m038
bd8fe59b49
A later successful asset could push last_ingest_at past an earlier asset whose thumbnail download failed, permanently excluding it from later incremental runs (only --full recovered it). Now the cursor never advances to/past the earliest failed asset, never below the prior cursor, and not at all on a --subset run. Regression tests added. (review finding #9)
103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
import datetime
|
|
import logging
|
|
import os
|
|
|
|
from photoflow.core.models import Asset
|
|
from photoflow.immich import pipeline
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
|
|
|
|
def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None,
|
|
tag=None, subset=None, full=False) -> dict:
|
|
os.makedirs(thumbs_dir, exist_ok=True)
|
|
|
|
tag_ids = None
|
|
if tag:
|
|
tid = client.resolve_tag_id(tag)
|
|
if not tid:
|
|
raise ValueError(f"Tag {tag!r} not found in Immich")
|
|
tag_ids = [tid]
|
|
|
|
updated_after = None if full else store.get_meta("last_ingest_at")
|
|
|
|
assets = client.search_assets(taken_after=date_from, taken_before=date_to,
|
|
tag_ids=tag_ids, updated_after=updated_after)
|
|
if subset is not None:
|
|
assets = assets[:subset]
|
|
|
|
processed_marked = 0
|
|
tag_counts: dict = {}
|
|
successful_updated: list = [] # updated_at of assets ingested this run
|
|
failed_updated: list = [] # updated_at of assets we could NOT ingest
|
|
|
|
for a in assets:
|
|
# Revision (2): sanitize the Immich-provided id before using it as a
|
|
# path component, guarding against path traversal (mirror /thumb proxy).
|
|
safe_id = os.path.basename(a["id"])
|
|
thumb_path = os.path.join(thumbs_dir, f"{safe_id}.jpg")
|
|
# Revision (1): treat an existing 0-byte file as missing.
|
|
need_thumb = (not os.path.exists(thumb_path)
|
|
or os.path.getsize(thumb_path) == 0)
|
|
if need_thumb:
|
|
try:
|
|
data = client.download_thumbnail(a["id"])
|
|
except Exception:
|
|
log.exception("Thumbnail download failed for asset %s", a["id"])
|
|
failed_updated.append(a["updated_at"])
|
|
continue
|
|
# Write to a temp path and os.replace into place so a failure
|
|
# never leaves a 0-byte <id>.jpg behind; clean up the temp file
|
|
# if the write itself fails (e.g. disk full).
|
|
tmp_path = f"{thumb_path}.tmp"
|
|
try:
|
|
with open(tmp_path, "wb") as f:
|
|
f.write(data)
|
|
os.replace(tmp_path, thumb_path)
|
|
except Exception:
|
|
log.exception("Thumbnail write failed for asset %s", a["id"])
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
failed_updated.append(a["updated_at"])
|
|
continue
|
|
|
|
store.upsert_asset(Asset(
|
|
immich_id=a["id"], taken_at=a["taken_at"], gps_lat=a["gps_lat"],
|
|
gps_lon=a["gps_lon"], place_city=a["place_city"],
|
|
place_country=a["place_country"], type=a["type"],
|
|
thumb_path=thumb_path, ingested_at=_now(), updated_at=a["updated_at"]))
|
|
store.set_asset_tags(a["id"], a["tags"])
|
|
for t in a["tags"]:
|
|
tag_counts[t] = tag_counts.get(t, 0) + 1
|
|
|
|
if pipeline.PROCESSED in a["tags"]:
|
|
store.mark_processed(a["id"])
|
|
processed_marked += 1
|
|
|
|
successful_updated.append(a["updated_at"])
|
|
|
|
for name, count in tag_counts.items():
|
|
store.upsert_tag(name, count=count)
|
|
|
|
# Advance the incremental cursor, but never past an asset we failed to ingest
|
|
# (a later success must not strand an earlier failure on the next run), never
|
|
# below the prior cursor, and not at all on a --subset sampling run.
|
|
if subset is None:
|
|
new_watermark = updated_after or ""
|
|
floor = min((u for u in failed_updated if u), default=None)
|
|
for u in successful_updated:
|
|
if not u or (floor is not None and u >= floor):
|
|
continue
|
|
if u > new_watermark:
|
|
new_watermark = u
|
|
if new_watermark:
|
|
store.set_meta("last_ingest_at", new_watermark)
|
|
|
|
return {"fetched": len(assets), "processed_marked": processed_marked}
|