|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
| 3 | +import logging |
3 | 4 | from typing import Any, Optional |
4 | 5 |
|
5 | 6 | from flask import Response, session |
|
15 | 16 | from ckanext.datavic_odp_theme.views import get_blueprints |
16 | 17 | from ckanext.datavic_odp_theme.helpers import get_helpers, group_list |
17 | 18 |
|
| 19 | +log = logging.getLogger(__name__) |
| 20 | + |
18 | 21 |
|
19 | 22 | class DatavicODPTheme(p.SingletonPlugin): |
20 | 23 | p.implements(p.IConfigurer) |
@@ -198,24 +201,103 @@ def after_dataset_create(self, context, pkg_dict): |
198 | 201 | group.add_package_by_name(pkg_dict.get('name', None)) |
199 | 202 |
|
200 | 203 | def after_dataset_update(self, context, pkg_dict): |
201 | | - self._trigger_after_resource_create(pkg_dict) |
| 204 | + self._submit_new_resources_only(pkg_dict) |
| 205 | + |
202 | 206 | group_id = pkg_dict.get('category', None) |
203 | 207 | if group_id: |
204 | 208 | group = model.Group.get(group_id) |
205 | 209 | groups = context.get('package').get_groups('group') |
206 | 210 | if group not in groups: |
207 | 211 | group.add_package_by_name(pkg_dict.get('name')) |
208 | 212 |
|
| 213 | + def _submit_new_resources_only(self, pkg_dict): |
| 214 | + """Submit only newly added resources to xloader during dataset update. |
| 215 | +
|
| 216 | + Compares current resource IDs against the previous activity snapshot |
| 217 | + to detect new resources. URL changes for existing resources are |
| 218 | + handled by the parent xloaderPlugin via ``notify()``. |
| 219 | +
|
| 220 | + Falls back to submitting all resources if no activity data is |
| 221 | + available (e.g. activity plugin disabled, data migration). |
| 222 | + """ |
| 223 | + current_resources = pkg_dict.get("resources", []) |
| 224 | + current_res_ids = { |
| 225 | + r.get("id") for r in current_resources if r.get("id") |
| 226 | + } |
| 227 | + |
| 228 | + previous_res_ids = self._get_previous_resource_ids(pkg_dict.get("id")) |
| 229 | + |
| 230 | + if previous_res_ids is None: |
| 231 | + log.info( |
| 232 | + "No previous activity for package %s — " |
| 233 | + "submitting all %d resources", |
| 234 | + pkg_dict.get("id"), |
| 235 | + len(current_resources), |
| 236 | + ) |
| 237 | + for resource in current_resources: |
| 238 | + self._infer_format_and_submit(resource) |
| 239 | + return |
| 240 | + |
| 241 | + new_res_ids = current_res_ids - previous_res_ids |
| 242 | + |
| 243 | + if not new_res_ids: |
| 244 | + return |
| 245 | + |
| 246 | + log.info( |
| 247 | + "Detected %d new resource(s) for package %s: %s", |
| 248 | + len(new_res_ids), |
| 249 | + pkg_dict.get("id"), |
| 250 | + new_res_ids, |
| 251 | + ) |
| 252 | + |
| 253 | + for resource in current_resources: |
| 254 | + if resource.get("id") in new_res_ids: |
| 255 | + self._infer_format_and_submit(resource) |
| 256 | + |
| 257 | + def _get_previous_resource_ids(self, pkg_id): |
| 258 | + """Return resource IDs from the most recent activity, or ``None`` |
| 259 | + if unavailable. |
| 260 | + """ |
| 261 | + if not pkg_id or not p.plugin_loaded("activity"): |
| 262 | + return None |
| 263 | + |
| 264 | + try: |
| 265 | + activities = tk.get_action("package_activity_list")( |
| 266 | + {"ignore_auth": True}, |
| 267 | + { |
| 268 | + "id": pkg_id, |
| 269 | + "limit": 1, |
| 270 | + "include_hidden_activity": True, |
| 271 | + }, |
| 272 | + ) |
| 273 | + except Exception: |
| 274 | + return None |
| 275 | + |
| 276 | + if not activities: |
| 277 | + return None |
| 278 | + |
| 279 | + prev_pkg = activities[0].get("data", {}).get("package", {}) |
| 280 | + prev_resources = prev_pkg.get("resources", []) |
| 281 | + return {r.get("id") for r in prev_resources if r.get("id")} |
| 282 | + |
| 283 | + def _infer_format_and_submit(self, resource): |
| 284 | + """Infer the resource format from its URL if missing, then submit.""" |
| 285 | + if resource and not resource.get("format"): |
| 286 | + if not resource.get("url_type"): |
| 287 | + url_without_params = resource.get("url", "").split("?")[0] |
| 288 | + resource["format"] = ( |
| 289 | + url_without_params.split(".")[-1].lower() |
| 290 | + ) |
| 291 | + self._submit_to_xloader(resource) |
| 292 | + |
209 | 293 | def _trigger_after_resource_create(self, pkg_dict): |
210 | | - """Dataset syndication doesn't trigger the `after_resource_create` method. |
211 | | - So here we want to run submit for each resource after dataset creation. |
| 294 | + """Submit all resources after dataset creation. |
| 295 | +
|
| 296 | + Syndication via ``package_create`` does not trigger |
| 297 | + ``after_resource_create``, so we handle it here. |
212 | 298 | """ |
213 | 299 | for resource in pkg_dict.get("resources", []): |
214 | | - if resource and not resource.get("format"): |
215 | | - if not resource["url_type"]: |
216 | | - url_without_params = resource["url"].split('?')[0] |
217 | | - resource["format"] = url_without_params.split('.')[-1].lower() |
218 | | - self._submit_to_datapusher(resource) |
| 300 | + self._infer_format_and_submit(resource) |
219 | 301 |
|
220 | 302 | def _submit_to_datapusher(self, resource_dict): |
221 | 303 | """The original method doesn't check if `url_type` is here. Seems like |
|
0 commit comments