535 lines
17 KiB
Python
535 lines
17 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
|
|
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):
|
|
"""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)
|
|
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)
|
|
logger.info("Scanning WebDAV gpx for user %s", user_id)
|
|
gpx_count += scan_webdav_for_gpx(client, user_id)
|
|
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
|
|
)
|
|
logger.info("Scanning WebDAV bgstats for user %s", user_id)
|
|
bgstats_count += scan_webdav_for_bgstats(client, user_id)
|
|
logger.info("Scanning WebDAV ebird for user %s", user_id)
|
|
ebird_count += scan_webdav_for_ebird(client, user_id)
|
|
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):
|
|
"""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.
|
|
"""
|
|
# 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 (f.get("name") or "") == "statistics.sqlite3":
|
|
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 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):
|
|
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")
|
|
new_imports = 0
|
|
already_imported = set(
|
|
TrailGPXImport.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(gpx_extensions):
|
|
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"{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)
|
|
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)
|
|
|
|
return new_imports
|
|
|
|
|
|
def scan_webdav_for_retroarch(webdav_client, user_id, update_hash_only=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.
|
|
"""
|
|
import hashlib
|
|
import shutil
|
|
import zipfile
|
|
from datetime import datetime
|
|
|
|
from scrobbles.models import RetroarchImport
|
|
from scrobbles.tasks import process_retroarch_import
|
|
|
|
retroarch_path = DEFAULT_RETROARCH_PATH + "playlogs/"
|
|
try:
|
|
webdav_client.info(retroarch_path)
|
|
except:
|
|
logger.info("No var/retroarch/ directory on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
try:
|
|
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
|
|
|
|
lrtl_files = sorted(
|
|
[f for f in files if (f.get("name") or "").lower().endswith(".lrtl")],
|
|
key=lambda x: (x.get("name") or ""),
|
|
)
|
|
if not lrtl_files:
|
|
logger.info("No .lrtl files found on webdav", extra={"user_id": user_id})
|
|
return 0
|
|
|
|
# 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()
|
|
|
|
# Skip if the last completed import already has this hash
|
|
last_import = (
|
|
RetroarchImport.objects.filter(
|
|
user_id=user_id, processed_finished__isnull=False
|
|
)
|
|
.order_by("-processed_finished")
|
|
.first()
|
|
)
|
|
if 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 = []
|
|
for f in lrtl_files:
|
|
basename = f.get("name")
|
|
dst = os.path.join(download_dir, basename)
|
|
try:
|
|
webdav_client.download_sync(
|
|
remote_path=f.get("path"),
|
|
local_path=dst,
|
|
)
|
|
downloaded.append(basename)
|
|
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=f"batch-{len(downloaded)}-files",
|
|
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):
|
|
"""Download .bgsplay files from WebDAV and queue imports for new files."""
|
|
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
|
|
|
|
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 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)
|
|
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)
|
|
|
|
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."""
|
|
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
|
|
|
|
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 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)
|
|
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)
|
|
|
|
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."""
|
|
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
|
|
already_imported = set(
|
|
ScaleCSVImport.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 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"{scale_path}/{fname}", local_path=tmp.name
|
|
)
|
|
imp = ScaleCSVImport.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_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
|