[importers] Add processed stashing to eBird imports

vrobbler/apps/scrobbles/importers/webdav.py

@@ -70,7 +70,9 @@ def import_from_webdav_for_all_users(
             client, user_id, include_processed=include_processed
         )
         logger.info("Scanning WebDAV ebird for user %s", user_id)
-        ebird_count += scan_webdav_for_ebird(client, user_id)
+        ebird_count += scan_webdav_for_ebird(
+            client, user_id, include_processed=include_processed
+        )
         logger.info("Scanning WebDAV scale for user %s", user_id)
         scale_count += scan_webdav_for_scale(client, user_id)
 
@@ -550,8 +552,12 @@ def scan_webdav_for_bgstats(webdav_client, user_id, include_processed=False):
     return new_imports
 
 
-def scan_webdav_for_ebird(webdav_client, user_id):
-    """Download .csv files from WebDAV var/ebird/ and queue imports for new files."""
+def scan_webdav_for_ebird(webdav_client, user_id, include_processed=False):
+    """Download .csv files from WebDAV var/ebird/ and queue imports for new files.
+
+    After importing, files are moved to var/ebird/processed/ so they are
+    not re-imported on subsequent scans unless *include_processed* is True.
+    """
     from scrobbles.models import EBirdCSVImport
     from scrobbles.tasks import process_ebird_csv_import
 
@@ -571,6 +577,12 @@ def scan_webdav_for_ebird(webdav_client, user_id):
         )
         return 0
 
+    processed_dir = f"{ebird_path}processed/"
+    try:
+        webdav_client.mkdir(processed_dir, recursive=True)
+    except Exception:
+        pass
+
     new_imports = 0
     already_imported = set(
         EBirdCSVImport.objects.filter(user_id=user_id).values_list(
@@ -582,6 +594,8 @@ def scan_webdav_for_ebird(webdav_client, user_id):
         fname = os.path.basename(fname)
         if not fname.lower().endswith(".csv"):
             continue
+        if fname == "processed":
+            continue
         if fname in already_imported:
             logger.debug(f"Skipping already-imported {fname}")
             continue
@@ -589,7 +603,7 @@ def scan_webdav_for_ebird(webdav_client, user_id):
         tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
         try:
             webdav_client.download_sync(
-                remote_path=f"{ebird_path}/{fname}", local_path=tmp.name
+                remote_path=f"{ebird_path}{fname}", local_path=tmp.name
             )
             imp = EBirdCSVImport.objects.create(
                 user_id=user_id,
@@ -597,6 +611,14 @@ def scan_webdav_for_ebird(webdav_client, user_id):
             )
             with open(tmp.name, "rb") as f:
                 imp.csv_file.save(fname, f, save=True)
+
+            stem, ext = os.path.splitext(fname)
+            ts = datetime.now().strftime("%Y%m%dT%H%M%S")
+            webdav_client.move(
+                f"{ebird_path}{fname}",
+                f"{processed_dir}{stem}_{ts}{ext}",
+            )
+
             process_ebird_csv_import.delay(imp.id)
             new_imports += 1
         except Exception as e:
@@ -604,6 +626,39 @@ def scan_webdav_for_ebird(webdav_client, user_id):
         finally:
             os.unlink(tmp.name)
 
+    if include_processed:
+        try:
+            processed_files = webdav_client.list(processed_dir)
+        except Exception as e:
+            logger.warning(
+                "Could not list var/ebird/processed/",
+                extra={"user_id": user_id, "error": str(e)},
+            )
+            return new_imports
+
+        for fname in processed_files:
+            fname = os.path.basename(fname)
+            if not fname.lower().endswith(".csv"):
+                continue
+
+            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
+            try:
+                webdav_client.download_sync(
+                    remote_path=f"{processed_dir}{fname}", local_path=tmp.name
+                )
+                imp = EBirdCSVImport.objects.create(
+                    user_id=user_id,
+                    original_filename=fname,
+                )
+                with open(tmp.name, "rb") as f:
+                    imp.csv_file.save(fname, f, save=True)
+                process_ebird_csv_import.delay(imp.id)
+                new_imports += 1
+            except Exception as e:
+                logger.error(f"Failed to import processed eBird CSV file {fname}: {e}")
+            finally:
+                os.unlink(tmp.name)
+
     return new_imports