From 13dd5b67d05d88af3c65c6d809259191e01e8d4c Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Thu, 28 May 2026 17:40:19 -0400 Subject: [PATCH] [importers] Add processed stashing to eBird imports --- vrobbler/apps/scrobbles/importers/webdav.py | 63 +++++++++++++++++++-- 1 file changed, 59 insertions(+), 4 deletions(-) diff --git a/vrobbler/apps/scrobbles/importers/webdav.py b/vrobbler/apps/scrobbles/importers/webdav.py index b0e64ac..c1dcadd 100644 --- a/vrobbler/apps/scrobbles/importers/webdav.py +++ b/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