save

server/planning/feeding_services/onclusive_api_service.py

@@ -58,6 +58,13 @@ class OnclusiveApiService(HTTPFeedingServiceBase):
             "required": False,
             "default": 365,
         },
+        {
+            "id": "days_to_reingest",
+            "type": "text",
+            "label": lazy_gettext("Days to Reingest"),
+            "placeholder": lazy_gettext("Days"),
+            "required": False,
+        },
     ]
 
     HTTP_AUTH = False
@@ -86,6 +93,19 @@ def _update(self, provider, update):
             update["last_updated"] = utcnow().replace(
                 second=0
             )  # next time start from here, onclusive api does not use seconds
+
+            # force reingest starting from now - days_to_reingest
+            if provider["config"].get("days_to_reingest"):
+                start_date = datetime.now() - timedelta(days=int(provider["config"]["days_to_reingest"]))
+                logger.info("Reingesting from %s", start_date.date().isoformat())
+                update["config"] = provider["config"].copy()
+                update["config"]["days_to_ingest"] = ""
+                # override to reset
+                update["tokens"]["start_date"] = start_date
+                update["tokens"]["next_start"] = start_date
+                update["tokens"]["reingesting"] = True
+                update["tokens"]["import_finished"] = None
+
             if update["tokens"].get("import_finished"):
                 # populate it for cases when import was done before we introduced the field
                 update["tokens"].setdefault("next_start", update["tokens"]["import_finished"] - timedelta(hours=5))
@@ -120,6 +140,8 @@ def _update(self, provider, update):
                     if date > processed_date  # when continuing skip previously ingested days
                 )
             logger.info("ingest from onclusive %s with params %s", url, params)
+            print("DONE")
+            return
             try:
                 for i in iterations:
                     params[iterations_param] = i
@@ -129,9 +151,11 @@ def _update(self, provider, update):
                     logger.info("Onclusive returned %d items", len(items))
                     for item in items:
                         item.setdefault("language", self.language)
+                        print("ITEM", item)
                     yield items
                     update["tokens"][iterations_param] = i
                 update["tokens"].setdefault("import_finished", utcnow())
+                update["tokens"]["reingesting"] = False
             except SoftTimeLimitExceeded:
                 logger.warning("stopped due to time limit, tokens=%s", update["tokens"])