814 lines
27 KiB
Python
814 lines
27 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
from datetime import datetime
|
|
|
|
from books.koreader import fetch_file_from_webdav
|
|
from profiles.models import UserProfile
|
|
from scrobbles.models import KoReaderImport, TrailGPXImport
|
|
from scrobbles.tasks import process_koreader_import, process_trail_gpx_import
|
|
from scrobbles.utils import get_file_md5_hash
|
|
from webdav.client import get_webdav_client
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_GPX_PATH = "var/gpx/"
|
|
DEFAULT_KOREADER_PATH = "var/koreader/"
|
|
DEFAULT_RETROARCH_PATH = "var/retroarch/"
|
|
DEFAULT_BGSTATS_PATH = "var/bgstats/"
|
|
DEFAULT_EBIRD_PATH = "var/ebird/"
|
|
DEFAULT_SCALE_PATH = "var/scale/"
|
|
|
|
|
|
def import_from_webdav_for_all_users(
|
|
restart=False,
|
|
update_retroarch_hash=False,
|
|
update_koreader_etag=False,
|
|
include_processed=False,
|
|
):
|
|
"""Iterate all WebDAV-enabled users, scanning each media-type directory."""
|
|
webdav_enabled_user_ids = UserProfile.objects.filter(
|
|
webdav_url__isnull=False,
|
|
webdav_user__isnull=False,
|
|
webdav_pass__isnull=False,
|
|
webdav_auto_import=True,
|
|
).values_list("user_id", flat=True)
|
|
if include_processed:
|
|
logger.warning(
|
|
"Re-importing previously-processed files from processed/ subdirectories "
|
|
"— this may create duplicate scrobbles."
|
|
)
|
|
|
|
logger.info(f"Start import of {webdav_enabled_user_ids.count()} webdav accounts")
|
|
|
|
ko_count = 0
|
|
gpx_count = 0
|
|
retro_count = 0
|
|
bgstats_count = 0
|
|
ebird_count = 0
|
|
scale_count = 0
|
|
|
|
for user_id in webdav_enabled_user_ids:
|
|
client = get_webdav_client(user_id)
|
|
logger.info("Scanning WebDAV for user %s", user_id)
|
|
|
|
logger.info("Scanning WebDAV koreader for user %s", user_id)
|
|
ko_count += scan_webdav_for_koreader(
|
|
client, user_id, restart, update_etag_only=update_koreader_etag
|
|
)
|
|
logger.info("Scanning WebDAV gpx for user %s", user_id)
|
|
gpx_count += scan_webdav_for_gpx(
|
|
client, user_id, include_processed=include_processed
|
|
)
|
|
logger.info("Scanning WebDAV retroarch for user %s", user_id)
|
|
retro_count += scan_webdav_for_retroarch(
|
|
client,
|
|
user_id,
|
|
update_hash_only=update_retroarch_hash,
|
|
include_processed=include_processed,
|
|
)
|
|
logger.info("Scanning WebDAV bgstats for user %s", user_id)
|
|
bgstats_count += scan_webdav_for_bgstats(
|
|
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, include_processed=include_processed
|
|
)
|
|
logger.info("Scanning WebDAV scale for user %s", user_id)
|
|
scale_count += scan_webdav_for_scale(client, user_id)
|
|
|
|
logger.info(
|
|
"Started %d KOReader, %d Trail GPX, %d Retroarch, %d BGStats, %d eBird, %d Scale WebDAV imports",
|
|
ko_count,
|
|
gpx_count,
|
|
retro_count,
|
|
bgstats_count,
|
|
ebird_count,
|
|
scale_count,
|
|
extra={
|
|
"koreader": ko_count,
|
|
"trail_gpx": gpx_count,
|
|
"retroarch": retro_count,
|
|
"bgstats": bgstats_count,
|
|
"ebird": ebird_count,
|
|
"scale": scale_count,
|
|
},
|
|
)
|
|
return ko_count, gpx_count, retro_count, bgstats_count, ebird_count, scale_count
|
|
|
|
|
|
def scan_webdav_for_koreader(
|
|
webdav_client, user_id, restart=False, update_etag_only=False
|
|
):
|
|
"""Check for koreader statistics.sqlite3 and queue import if changed.
|
|
|
|
Uses the remote file's ETag (from a single PROPFIND) to detect changes
|
|
without downloading the file, making the common no-change case fast.
|
|
|
|
When *update_etag_only* is True, stores the remote ETag on the last
|
|
completed import without re-downloading — useful when migrating to the
|
|
ETag-based check to avoid a one-time re-import.
|
|
"""
|
|
if not update_etag_only:
|
|
# Don't queue if one is still being processed
|
|
if KoReaderImport.objects.filter(
|
|
user_id=user_id, processed_finished__isnull=True
|
|
).exists():
|
|
logger.info(
|
|
"KoReader import already pending for user",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
try:
|
|
files = webdav_client.list(DEFAULT_KOREADER_PATH, get_info=True)
|
|
except:
|
|
logger.info(
|
|
"No var/koreader/ directory on webdav",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
sqlite_info = None
|
|
for f in files:
|
|
if "statistics.sqlite3" in f.get("path"):
|
|
sqlite_info = f
|
|
break
|
|
|
|
if not sqlite_info:
|
|
logger.info(
|
|
"No statistics.sqlite3 found in var/koreader/",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
remote_etag = sqlite_info.get("etag") or ""
|
|
|
|
last_import = (
|
|
KoReaderImport.objects.filter(user_id=user_id, processed_finished__isnull=False)
|
|
.order_by("processed_finished")
|
|
.last()
|
|
)
|
|
|
|
if update_etag_only:
|
|
if last_import and remote_etag:
|
|
last_import.webdav_etag = remote_etag
|
|
last_import.save(update_fields=["webdav_etag"])
|
|
logger.info(
|
|
"Updated KoReader webdav_etag (%s) without re-import",
|
|
remote_etag[:16],
|
|
extra={"user_id": user_id, "last_import_id": last_import.id},
|
|
)
|
|
return 0
|
|
|
|
if last_import and last_import.webdav_etag and remote_etag:
|
|
if last_import.webdav_etag == remote_etag:
|
|
logger.info(
|
|
"koreader stats file unchanged (ETag match)",
|
|
extra={
|
|
"user_id": user_id,
|
|
"etag": remote_etag,
|
|
"last_import_id": last_import.id,
|
|
},
|
|
)
|
|
return 0
|
|
|
|
file_path = fetch_file_from_webdav(user_id)
|
|
if not file_path:
|
|
logger.warning(
|
|
"Could not fetch koreader file from webdav",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
new_hash = get_file_md5_hash(file_path)
|
|
old_hash = last_import.file_md5_hash() if last_import else None
|
|
|
|
if last_import and not remote_etag and new_hash == old_hash:
|
|
logger.info(
|
|
"koreader stats file has not changed (content hash match)",
|
|
extra={
|
|
"user_id": user_id,
|
|
"new_hash": new_hash,
|
|
"old_hash": old_hash,
|
|
"last_import_id": last_import.id,
|
|
},
|
|
)
|
|
os.unlink(file_path)
|
|
return 0
|
|
|
|
koreader_import, created = KoReaderImport.objects.get_or_create(
|
|
user_id=user_id, processed_finished__isnull=True
|
|
)
|
|
|
|
if not created and not restart:
|
|
logger.info(
|
|
f"Not resuming failed KoReader import {koreader_import.id} "
|
|
f"for user {user_id}, use restart=True to restart"
|
|
)
|
|
os.unlink(file_path)
|
|
return 0
|
|
|
|
koreader_import.webdav_etag = remote_etag
|
|
koreader_import.save_sqlite_file_to_self(file_path)
|
|
process_koreader_import.delay(koreader_import.id)
|
|
os.unlink(file_path)
|
|
return 1
|
|
|
|
|
|
def scan_webdav_for_gpx(webdav_client, user_id, include_processed=False):
|
|
gpx_path = DEFAULT_GPX_PATH # TODO allow this to be configured in user settings
|
|
try:
|
|
webdav_client.info(DEFAULT_GPX_PATH)
|
|
logger.info("var/gpx/ found, listing files")
|
|
except:
|
|
logger.info("No var/gpx/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
files = webdav_client.list(gpx_path)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/gpx/", extra={"user_id": user_id, "error": str(e)}
|
|
)
|
|
return 0
|
|
|
|
gpx_extensions = (".gpx", ".fit")
|
|
processed_dir = f"{gpx_path}processed/"
|
|
try:
|
|
webdav_client.mkdir(processed_dir, recursive=True)
|
|
except Exception:
|
|
pass
|
|
|
|
new_imports = 0
|
|
|
|
for fname in files:
|
|
fname = os.path.basename(fname)
|
|
if not fname.lower().endswith(gpx_extensions):
|
|
continue
|
|
if fname == "processed":
|
|
continue
|
|
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=f"{gpx_path}{fname}", local_path=tmp.name
|
|
)
|
|
imp = TrailGPXImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
)
|
|
with open(tmp.name, "rb") as f:
|
|
imp.gpx_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"{gpx_path}{fname}",
|
|
f"{processed_dir}{stem}_{ts}{ext}",
|
|
)
|
|
|
|
process_trail_gpx_import.delay(imp.id)
|
|
new_imports += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to import {fname}: {e}")
|
|
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/gpx/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(gpx_extensions):
|
|
continue
|
|
|
|
remote_path = f"{processed_dir}{fname}"
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=remote_path, local_path=tmp.name
|
|
)
|
|
imp = TrailGPXImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
)
|
|
with open(tmp.name, "rb") as f:
|
|
imp.gpx_file.save(fname, f, save=True)
|
|
|
|
process_trail_gpx_import.delay(imp.id)
|
|
new_imports += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to import processed GPX file {fname}: {e}")
|
|
finally:
|
|
os.unlink(tmp.name)
|
|
|
|
return new_imports
|
|
|
|
|
|
def scan_webdav_for_retroarch(
|
|
webdav_client, user_id, update_hash_only=False, include_processed=False
|
|
):
|
|
"""Check for new .lrtl files on WebDAV, download+import if changed.
|
|
|
|
Uses ETags from a single PROPFIND to detect changes without downloading
|
|
any file content, making the common no-change case very fast.
|
|
|
|
When *update_hash_only* is True, mismatched hashes update the last import's
|
|
stored hash without re-downloading — useful when migrating to a new hash
|
|
scheme to avoid a one-time re-import.
|
|
|
|
When *include_processed* is True, also imports files from the processed/
|
|
subdirectory. These are sorted chronologically by their YYYYMMDDHHMM-
|
|
timestamp prefix so that the importer always sees the earliest snapshot
|
|
first, which is critical for correct incremental scrobbling since
|
|
retroarch overwrites the same .lrtl filename.
|
|
"""
|
|
import hashlib
|
|
import shutil
|
|
import re
|
|
import zipfile
|
|
from datetime import datetime
|
|
|
|
from scrobbles.models import RetroarchImport
|
|
from scrobbles.tasks import process_retroarch_import
|
|
|
|
retroarch_path = DEFAULT_RETROARCH_PATH
|
|
processed_dir = f"{retroarch_path}processed/"
|
|
try:
|
|
webdav_client.info(retroarch_path)
|
|
except:
|
|
logger.info("No var/retroarch/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
webdav_client.mkdir(processed_dir, recursive=True)
|
|
except Exception:
|
|
pass
|
|
|
|
# Collect files from root directory
|
|
try:
|
|
root_files = webdav_client.list(retroarch_path, get_info=True)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/retroarch/",
|
|
extra={"user_id": user_id, "error": str(e)},
|
|
)
|
|
return 0
|
|
|
|
# Extract basename from path since the webdav adapter returns name=None
|
|
for f in root_files:
|
|
if f.get("path") and not f.get("name"):
|
|
f["name"] = os.path.basename(f["path"])
|
|
|
|
lrtl_files = [
|
|
f for f in root_files
|
|
if (f.get("name") or "").lower().endswith(".lrtl")
|
|
]
|
|
|
|
# Optionally include historical files from processed/
|
|
if include_processed:
|
|
try:
|
|
processed_files = webdav_client.list(processed_dir, get_info=True)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/retroarch/processed/",
|
|
extra={"user_id": user_id, "error": str(e)},
|
|
)
|
|
else:
|
|
for f in processed_files:
|
|
if f.get("path") and not f.get("name"):
|
|
f["name"] = os.path.basename(f["path"])
|
|
lrtl_files.extend([
|
|
f for f in processed_files
|
|
if (f.get("name") or "").lower().endswith(".lrtl")
|
|
])
|
|
|
|
if not lrtl_files:
|
|
logger.info("No .lrtl files found on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
# Sort chronologically: processed files (timestamp prefixed) first,
|
|
# current root files last.
|
|
ts_pattern = re.compile(r"^(\d{12})-")
|
|
def sort_key(f):
|
|
name = f.get("name") or ""
|
|
m = ts_pattern.match(name)
|
|
if m:
|
|
return m.group(1)
|
|
# Root files (no timestamp) go after all historical snapshots
|
|
return "999999999999"
|
|
|
|
lrtl_files.sort(key=sort_key)
|
|
|
|
# Don't queue if one is still being processed
|
|
if RetroarchImport.objects.filter(
|
|
user_id=user_id, processed_finished__isnull=True
|
|
).exists():
|
|
logger.info(
|
|
"Retroarch import already pending for user",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
# Compute hash from filenames + ETags (no downloads needed)
|
|
hasher = hashlib.md5()
|
|
for f in lrtl_files:
|
|
hasher.update((f.get("name") or "").encode())
|
|
hasher.update((f.get("etag") or f.get("modified") or "").encode())
|
|
content_hash = hasher.hexdigest()
|
|
|
|
last_import = (
|
|
RetroarchImport.objects.filter(
|
|
user_id=user_id, processed_finished__isnull=False
|
|
)
|
|
.order_by("-processed_finished")
|
|
.first()
|
|
)
|
|
|
|
# Skip unchanged root files only, never skip when re-processing history
|
|
if not include_processed and last_import and last_import.files_hash == content_hash:
|
|
logger.info(
|
|
"Retroarch lrtl files unchanged for user, skipping",
|
|
extra={"user_id": user_id},
|
|
)
|
|
return 0
|
|
|
|
if update_hash_only and last_import:
|
|
logger.info(
|
|
"Updating retroarch files_hash (%s) without re-import",
|
|
content_hash[:12],
|
|
extra={"user_id": user_id},
|
|
)
|
|
last_import.files_hash = content_hash
|
|
last_import.save(update_fields=["files_hash"])
|
|
return 0
|
|
|
|
# Something changed — download everything
|
|
download_dir = tempfile.mkdtemp()
|
|
try:
|
|
downloaded = []
|
|
imported_filenames = []
|
|
for f in lrtl_files:
|
|
basename = f.get("name")
|
|
# Determine source directory: processed files already live under processed/
|
|
is_processed = ts_pattern.match(basename)
|
|
source_dir = processed_dir if is_processed else retroarch_path
|
|
|
|
remote_path = f"{source_dir}{basename}"
|
|
dst = os.path.join(download_dir, basename)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=remote_path,
|
|
local_path=dst,
|
|
)
|
|
downloaded.append(basename)
|
|
|
|
if is_processed:
|
|
imported_filenames.append(basename)
|
|
else:
|
|
ts = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
processed_name = f"{ts}-{basename}"
|
|
imported_filenames.append(processed_name)
|
|
webdav_client.move(
|
|
remote_path,
|
|
f"{processed_dir}{processed_name}",
|
|
)
|
|
except Exception as e:
|
|
logger.error("Failed to download retroarch %s: %s", basename, e)
|
|
|
|
if not downloaded:
|
|
return 0
|
|
|
|
zip_path = os.path.join(
|
|
download_dir,
|
|
f"retroarch-batch-{datetime.now().strftime('%Y%m%d%H%M%S')}.zip",
|
|
)
|
|
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for basename in downloaded:
|
|
zf.write(os.path.join(download_dir, basename), basename)
|
|
|
|
imp = RetroarchImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=", ".join(imported_filenames),
|
|
files_hash=content_hash,
|
|
)
|
|
with open(zip_path, "rb") as f:
|
|
imp.lrtl_file.save(f"retroarch-batch-{imp.uuid}.zip", f, save=True)
|
|
process_retroarch_import.delay(imp.id)
|
|
logger.info(
|
|
"Queued retroarch import %s with %d file(s) (hash=%s)",
|
|
imp.uuid,
|
|
len(downloaded),
|
|
content_hash[:12],
|
|
)
|
|
return 1
|
|
finally:
|
|
shutil.rmtree(download_dir, ignore_errors=True)
|
|
|
|
|
|
def scan_webdav_for_bgstats(webdav_client, user_id, include_processed=False):
|
|
"""Download .bgsplay files from WebDAV and queue imports for new files.
|
|
|
|
After importing, files are moved to var/bgstats/processed/ so they are
|
|
not re-imported on subsequent scans unless *include_processed* is True.
|
|
"""
|
|
from scrobbles.models import BGStatsImport
|
|
from scrobbles.tasks import process_bgstats_import
|
|
|
|
bgstats_path = DEFAULT_BGSTATS_PATH
|
|
try:
|
|
webdav_client.info(bgstats_path)
|
|
except:
|
|
logger.info("No var/bgstats/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
files = webdav_client.list(bgstats_path)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/bgstats/",
|
|
extra={"user_id": user_id, "error": str(e)},
|
|
)
|
|
return 0
|
|
|
|
processed_dir = f"{bgstats_path}processed/"
|
|
try:
|
|
webdav_client.mkdir(processed_dir, recursive=True)
|
|
except Exception:
|
|
pass
|
|
|
|
new_imports = 0
|
|
already_imported = set(
|
|
BGStatsImport.objects.filter(user_id=user_id).values_list(
|
|
"original_filename", flat=True
|
|
)
|
|
)
|
|
|
|
for fname in files:
|
|
fname = os.path.basename(fname)
|
|
if not fname.lower().endswith(".bgsplay"):
|
|
continue
|
|
if fname == "processed":
|
|
continue
|
|
if fname in already_imported:
|
|
logger.debug(f"Skipping already-imported {fname}")
|
|
continue
|
|
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=f"{bgstats_path}{fname}", local_path=tmp.name
|
|
)
|
|
imp = BGStatsImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
)
|
|
with open(tmp.name, "rb") as f:
|
|
imp.bgsplay_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"{bgstats_path}{fname}",
|
|
f"{processed_dir}{stem}_{ts}{ext}",
|
|
)
|
|
|
|
process_bgstats_import.delay(imp.id)
|
|
new_imports += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to import BG Stats file {fname}: {e}")
|
|
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/bgstats/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(".bgsplay"):
|
|
continue
|
|
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=f"{processed_dir}{fname}", local_path=tmp.name
|
|
)
|
|
imp = BGStatsImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
)
|
|
with open(tmp.name, "rb") as f:
|
|
imp.bgsplay_file.save(fname, f, save=True)
|
|
process_bgstats_import.delay(imp.id)
|
|
new_imports += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to import processed BG Stats file {fname}: {e}")
|
|
finally:
|
|
os.unlink(tmp.name)
|
|
|
|
return new_imports
|
|
|
|
|
|
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
|
|
|
|
ebird_path = DEFAULT_EBIRD_PATH
|
|
try:
|
|
webdav_client.info(ebird_path)
|
|
except:
|
|
logger.info("No var/ebird/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
files = webdav_client.list(ebird_path)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/ebird/",
|
|
extra={"user_id": user_id, "error": str(e)},
|
|
)
|
|
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(
|
|
"original_filename", flat=True
|
|
)
|
|
)
|
|
|
|
for fname in files:
|
|
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
|
|
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=f"{ebird_path}{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)
|
|
|
|
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:
|
|
logger.error(f"Failed to import eBird CSV file {fname}: {e}")
|
|
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
|
|
|
|
|
|
def scan_webdav_for_scale(webdav_client, user_id):
|
|
"""Download .csv files from WebDAV var/scale/ and queue imports for new files.
|
|
|
|
Because the scale CSV always has the same filename but grows by appending
|
|
rows, we detect changes by downloading the file and comparing its content
|
|
hash against the last completed import, rather than checking the filename.
|
|
"""
|
|
from scrobbles.models import ScaleCSVImport
|
|
from scrobbles.tasks import process_scale_csv_import
|
|
|
|
scale_path = (
|
|
DEFAULT_SCALE_PATH # TODO Allow this to be configured in a user profile setting
|
|
)
|
|
try:
|
|
webdav_client.info(scale_path)
|
|
except:
|
|
logger.info("No var/scale/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
files = webdav_client.list(scale_path)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Could not list var/scale/",
|
|
extra={"user_id": user_id, "error": str(e)},
|
|
)
|
|
return 0
|
|
|
|
new_imports = 0
|
|
|
|
for fname in 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"{scale_path}/{fname}", local_path=tmp.name
|
|
)
|
|
|
|
new_hash = get_file_md5_hash(tmp.name)
|
|
|
|
last_import = (
|
|
ScaleCSVImport.objects.filter(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
processed_finished__isnull=False,
|
|
)
|
|
.order_by("-processed_finished")
|
|
.first()
|
|
)
|
|
if last_import and last_import.file_hash == new_hash:
|
|
logger.debug(
|
|
"Scale CSV %s unchanged (hash match), skipping",
|
|
fname,
|
|
extra={"user_id": user_id, "hash": new_hash},
|
|
)
|
|
continue
|
|
|
|
imp = ScaleCSVImport.objects.create(
|
|
user_id=user_id,
|
|
original_filename=fname,
|
|
file_hash=new_hash,
|
|
)
|
|
with open(tmp.name, "rb") as f:
|
|
imp.csv_file.save(fname, f, save=True)
|
|
process_scale_csv_import.delay(imp.id)
|
|
new_imports += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to import Scale CSV file {fname}: {e}")
|
|
finally:
|
|
os.unlink(tmp.name)
|
|
|
|
return new_imports
|