FabledCurator/backend/app/tasks/external.py

"""Celery worker for off-platform file-host downloads (external_link ledger).

Walks the external_link table and fetches each pending/retryable link via
external_fetch, then routes the downloaded file(s) through the existing importer
(attach_in_place) so an archive becomes ImageRecords and any other file a
PostAttachment — linked to the SAME post the link came from (via a synthesized
sidecar). Long-running-task hygiene (rule #89): per-fetch wall-clock timeout,
attempt tracking + dead-letter, a recovery sweep, and retention of dead rows.

Concurrency: a per-host Redis lock serializes fetches to one-per-host (mega /
gdrive ban-avoidance), and an atomic status claim (pending/failed →
downloading) stops two workers grabbing the same link.

Files land in the artist's library tree (so an attached art image stays in
place, mirroring the gallery-dl download path); a captured archive/attachment is
copied into its store and the on-disk original removed.
"""
from __future__ import annotations

import json
import logging
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path

import redis
from sqlalchemy import delete, or_, select, update

from ..celery_app import celery
from ..config import get_config
from ..models import Artist, ExternalLink, ImageRecord, ImportSettings, Post, Source
from ..services.external_fetch import fetch_external
from ..services.importer import Importer
from ..services.thumbnailer import Thumbnailer
from ._sync_engine import sync_session_factory as _sync_session_factory

log = logging.getLogger(__name__)

IMAGES_ROOT = Path("/images")

# After this many failed attempts a link is dead-lettered (skipped by routine
# sweeps; the daily recovery retries it, sparingly). Mirrors the ingester ledger.
DEAD_LETTER_THRESHOLD = 3
# Per-fetch read + total budgets now live in external_fetch (a short read
# timeout fails a stalled host fast; a generous total caps a big-but-flowing
# download). The celery soft/hard time_limit is the outer backstop above those.
# Links enqueued per sweep — bounds the burst when a big backfill records many.
_SWEEP_BATCH = 50
# Recovery gives a dead link one more attempt once it has been dead for
# 2**recoveries days (1, 2, 4, 8, 16), and stops after _MAX_RECOVERIES. A host
# outage gets several later chances; a link that can never succeed (too big to
# import in time, a file made private) costs a handful of downloads, not three
# a day.
_MAX_RECOVERIES = 5
# A link recovery has given up on is pruned this long after its last attempt.
_RETENTION_DAYS = 30
# Per-host serialize lock. TTL is a safety net for a worker that dies holding it
# (normal completion/error releases it in `finally`); sized just past the fetch
# total budget (30 min) so a dead worker can't wedge a host's links much longer
# than one fetch would have taken.
_LOCK_PREFIX = "fc:extdl_lock:"
_LOCK_TTL = 2400
# A link 'downloading' this long has outlived fetch_external_link's hard limit
# (time_limit=3600), so its worker is gone and the sweep recovers it.
_STALL_MINUTES = 65
_SERIALIZE_COUNTDOWN = 120
_MAX_SERIALIZE_WAITS = 30

_redis_client: redis.Redis | None = None


def _redis() -> redis.Redis:
    global _redis_client
    if _redis_client is None:
        _redis_client = redis.from_url(get_config().celery_broker_url)
    return _redis_client


def _host_enabled(settings: ImportSettings, host: str) -> bool:
    """Per-host enable flag — defaults True (rule #26: works out of the box).
    The Settings UI slice adds real columns; getattr keeps this forward-
    compatible without a migration in this slice."""
    return bool(getattr(settings, f"extdl_{host}_enabled", True))


def _write_link_sidecar(file: Path, post: Post, platform: str, artist: Artist) -> None:
    """Sidecar next to a fetched file so the importer links it to `post`
    (category+id resolve the source+post; matches gallery-dl's sidecar shape)."""
    data = {
        "category": platform,
        "id": post.external_post_id,
        "url": post.post_url,
        "title": post.post_title,
        "artist": artist.name,
    }
    Path(str(file) + ".json").write_text(json.dumps(data))


@celery.task(
    name="backend.app.tasks.external.fetch_external_link",
    bind=True,
    soft_time_limit=3300,
    time_limit=3600,
)
def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict:
    """Fetch one external_link and route its file(s) through the importer."""
    SessionLocal = _sync_session_factory()

    # Atomic claim: only an actionable (pending/failed) row transitions to
    # downloading, so a duplicate enqueue (sweep + post-download hook) no-ops.
    with SessionLocal() as session:
        claimed = session.execute(
            update(ExternalLink)
            .where(
                ExternalLink.id == link_id,
                ExternalLink.status.in_(("pending", "failed")),
            )
            .values(status="downloading", started_at=datetime.now(UTC))
            .returning(ExternalLink.id)
        ).first()
        session.commit()
        if claimed is None:
            return {"link_id": link_id, "skipped": "not claimable"}

        link = session.get(ExternalLink, link_id)
        post = session.get(Post, link.post_id) if link.post_id else None
        if post is None:
            link.status = "dead"
            link.last_error = "post missing"
            link.completed_at = datetime.now(UTC)
            session.commit()
            return {"link_id": link_id, "error": "post missing"}
        artist = session.get(Artist, post.artist_id)
        source = session.get(Source, post.source_id) if post.source_id else None
        platform = source.platform if source is not None else "patreon"
        settings = ImportSettings.load_sync(session)

        if not _host_enabled(settings, link.host):
            log.info("extdl skip (host disabled): link=%s host=%s", link_id, link.host)
            link.status = "skipped"
            link.last_error = f"{link.host} disabled"
            link.completed_at = datetime.now(UTC)
            session.commit()
            return {"link_id": link_id, "skipped": "host disabled"}

        host, url, attempts = link.host, link.url, link.attempts
        log.info(
            "extdl start: link=%s host=%s post=%s artist=%s attempt=%d url=%s",
            link_id, host, post.id, artist.slug, attempts + 1, url,
        )

    # Per-host serialize: one fetch per host at a time. If busy, requeue.
    lock = None
    try:
        lock = _redis().lock(f"{_LOCK_PREFIX}{host}", timeout=_LOCK_TTL, blocking=False)
        got = bool(lock.acquire(blocking=False))
    except redis.RedisError as exc:
        log.warning("extdl lock unavailable for %s: %s — running uncapped", host, exc)
        lock, got = None, True
    if not got:
        # Release the claim back to pending so a later run can pick it up, then
        # requeue with a countdown (bounded, like download_source).
        with SessionLocal() as session:
            session.execute(
                update(ExternalLink).where(ExternalLink.id == link_id)
                .values(status="pending")
            )
            session.commit()
        log.info(
            "extdl requeue (host %s busy): link=%s wait=%d", host, link_id, _serialize_waits,
        )
        if _serialize_waits < _MAX_SERIALIZE_WAITS:
            fetch_external_link.apply_async(
                (link_id,), {"_serialize_waits": _serialize_waits + 1},
                countdown=_SERIALIZE_COUNTDOWN,
            )
        else:
            log.warning("extdl giving up requeue (host %s busy): link=%s", host, link_id)
        return {"link_id": link_id, "requeued": "host busy"}

    started = time.monotonic()
    # Per-LINK staging dir (not just per-post): two links on the same post (e.g.
    # the same film from mega + gdrive) can emit the same filename — sharing one
    # external/<post_id>/ dir let the second overwrite the first's file, then the
    # dedup-skip unlink orphaned a record → 404 on playback. Isolating per link
    # keeps each link's file at its own path. (#859)
    post_dir = (
        IMAGES_ROOT / artist.slug / platform / "external"
        / str(post.external_post_id) / str(link_id)
    )
    try:
        # Clear what an earlier, interrupted attempt left behind, so this fetch
        # starts from the link's imported files only.
        with SessionLocal() as session:
            _discard_unrecorded(session, post_dir)
        result = fetch_external(host, url, post_dir)
        with SessionLocal() as session:
            link = session.get(ExternalLink, link_id)
            if not result.ok:
                log.warning(
                    "extdl fetch failed: link=%s host=%s error=%s",
                    link_id, host, result.error,
                )
                _record_failure(link, attempts, result.error or "fetch failed")
                session.commit()
                _discard_unrecorded(session, post_dir)
                return {"link_id": link_id, "error": result.error}

            log.info(
                "extdl fetched: link=%s host=%s files=%d bytes=%d",
                link_id, host, len(result.files), result.bytes,
            )
            # Re-load post/artist/source attached to THIS session before handing
            # them to the importer (the claim block's instances are detached).
            post = session.get(Post, link.post_id)
            artist = session.get(Artist, post.artist_id)
            source = session.get(Source, post.source_id) if post.source_id else None
            image_ids = _route_files(session, result.files, post, platform, artist, source)
            link.status = "downloaded"
            link.last_error = None
            link.completed_at = datetime.now(UTC)
            link.duration_seconds = time.monotonic() - started
            session.commit()
            log.info(
                "extdl done: link=%s host=%s post=%s files=%d image(s)=%d dur=%.1fs",
                link_id, host, post.id, len(result.files), len(image_ids),
                link.duration_seconds,
            )

        # Thumbnails + ML for any newly-attached images (mirrors the download
        # path). Lazy import to dodge a task-module import cycle.
        if image_ids:
            from .ml import cpu_embed_enabled, embed_image
            from .thumbnail import generate_thumbnail
            do_embed = cpu_embed_enabled()
            for img_id in image_ids:
                generate_thumbnail.delay(img_id)
                if do_embed:
                    embed_image.delay(img_id)
        return {"link_id": link_id, "files": len(result.files), "images": len(image_ids)}
    except Exception as exc:  # never leave a link stuck in 'downloading'
        log.exception("external fetch task failed for link %s", link_id)
        with SessionLocal() as session:
            link = session.get(ExternalLink, link_id)
            if link is not None and link.status == "downloading":
                _record_failure(link, attempts, f"{type(exc).__name__}: {exc}")
                session.commit()
            _discard_unrecorded(session, post_dir)
        raise
    finally:
        if lock is not None:
            try:
                lock.release()
            except Exception:  # noqa: BLE001 — TTL may have already freed it
                pass


def _record_failure(link: ExternalLink, attempts: int, error: str) -> None:
    nxt = attempts + 1
    link.attempts = nxt
    link.last_error = error[:1000]
    link.status = "dead" if nxt >= DEAD_LETTER_THRESHOLD else "failed"
    link.completed_at = datetime.now(UTC)
    log.warning(
        "extdl link %s -> %s (attempt %d/%d): %s",
        link.id, link.status, nxt, DEAD_LETTER_THRESHOLD, error[:200],
    )


def _route_files(session, files, post, platform, artist, source) -> list[int]:
    """Import each fetched file through the SAME pipeline extracted-zip members
    use — importer.attach_in_place — so a downloaded archive is extracted to
    ImageRecords and provenance-linked to the post (via the synthesized sidecar),
    a non-art file is captured as a PostAttachment, and an art image is attached
    in place. Returns new image ids so the caller enqueues thumbnail + ML
    (tagging) for them, exactly as download_service does for downloaded media.

    Result handling mirrors download_service._phase3_persist branch-for-branch
    (duplicate cleanup, the unextracted-archive warning of #718) and logs every
    decision — this path is new and the operator expects to iterate on it."""
    settings = ImportSettings.load_sync(session)
    importer = Importer(
        session=session,
        images_root=IMAGES_ROOT,
        import_root=IMAGES_ROOT,
        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
        settings=settings,
    )
    image_ids: list[int] = []
    for f in files:
        _write_link_sidecar(f, post, platform, artist)
        result = importer.attach_in_place(f, artist=artist, source=source)
        session.commit()
        if result.status in ("imported", "superseded"):
            ids = list(getattr(result, "member_image_ids", []) or [])
            if result.image_id is not None and result.image_id not in ids:
                ids.append(result.image_id)
            image_ids.extend(ids)
            log.info(
                "extdl import: post=%s file=%s status=%s -> %d image(s) "
                "(provenance-linked, queued for tagging)",
                post.id, f.name, result.status, len(ids),
            )
        elif result.status == "attached":
            # Non-art, or an archive with no new images, captured as a
            # PostAttachment. An archive captured WITHOUT extracting any image
            # carries the reason, so the "zip but no images" case is visible.
            if result.error:
                log.warning(
                    "extdl archive captured UNEXTRACTED: post=%s file=%s reason=%s",
                    post.id, f.name, result.error,
                )
            else:
                log.info("extdl attachment captured: post=%s file=%s", post.id, f.name)
        elif result.status == "skipped":
            reason = result.skip_reason.value if result.skip_reason else None
            log.info(
                "extdl skipped: post=%s file=%s reason=%s", post.id, f.name, reason,
            )
        elif result.status == "failed":
            log.warning(
                "extdl import FAILED: post=%s file=%s error=%s",
                post.id, f.name, result.error,
            )
        else:  # 'refreshed' or any future status — work happened, no new images
            log.info(
                "extdl import: post=%s file=%s status=%s", post.id, f.name, result.status,
            )
        # Same policy as the subscription download path: drop the file unless
        # the library needs it at this path. A file imported in place is kept.
        importer.discard_after_attach(f, result)
        # The synthesized sidecar has done its job — don't litter the tree.
        Path(str(f) + ".json").unlink(missing_ok=True)
    return image_ids


def _discard_unrecorded(session, link_dir: Path) -> int:
    """Delete every file in a link's staging dir that no ImageRecord points at,
    then the dirs left empty. The dir belongs to one link, so anything in it
    without a record is a partial or abandoned download from an earlier
    attempt (megatools' `.megatmp.*`, a killed transfer, a rejected file).
    Images imported in place stay. Returns the number of files removed."""
    if not link_dir.is_dir():
        return 0
    files = [p for p in link_dir.rglob("*") if p.is_file()]
    if not files:
        return 0
    recorded = set(session.execute(
        select(ImageRecord.path).where(
            ImageRecord.path.in_([str(p) for p in files])
        )
    ).scalars())
    removed = 0
    for p in files:
        if str(p) in recorded:
            continue
        try:
            p.unlink(missing_ok=True)
            removed += 1
        except OSError as exc:
            log.warning("extdl could not remove %s: %s", p, exc)
    for d in sorted((p for p in link_dir.rglob("*") if p.is_dir()), reverse=True):
        try:
            d.rmdir()
        except OSError:
            pass  # not empty: it holds an imported file
    try:
        link_dir.rmdir()
    except OSError:
        pass
    if removed:
        log.info("extdl removed %d leftover file(s) from %s", removed, link_dir)
    return removed


@celery.task(name="backend.app.tasks.external.sweep_external_links")
def sweep_external_links() -> dict:
    """Enqueue a bounded batch of actionable links (pending, or failed below the
    dead-letter threshold). Driven by the post-download hook and a beat tick.

    First it recovers links whose worker died mid-fetch: a link 'downloading'
    for longer than the task's hard limit has no worker left, so it counts as a
    failed attempt and goes back to the retry path (dead-lettered at the
    threshold like any other failure). The next attempt clears its leftovers."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        now = datetime.now(UTC)
        stale = session.execute(
            select(ExternalLink).where(
                ExternalLink.status == "downloading",
                or_(
                    ExternalLink.started_at.is_(None),
                    ExternalLink.started_at < now - timedelta(minutes=_STALL_MINUTES),
                ),
            )
        ).scalars().all()
        for link in stale:
            _record_failure(link, link.attempts, "worker stopped mid-fetch")
        session.commit()
        ids = session.execute(
            select(ExternalLink.id).where(
                ExternalLink.status.in_(("pending", "failed")),
                ExternalLink.attempts < DEAD_LETTER_THRESHOLD,
            ).order_by(ExternalLink.id.asc()).limit(_SWEEP_BATCH)
        ).scalars().all()
    for link_id in ids:
        fetch_external_link.delay(link_id)
    return {"enqueued": len(ids), "recovered": len(stale)}


@celery.task(name="backend.app.tasks.external.recover_external_links")
def recover_external_links() -> dict:
    """Recovery sweep: give each dead link one more attempt once it
    has been dead for 2**recoveries days, then enqueue. A recovered link that
    fails again goes straight back to dead, and after _MAX_RECOVERIES it stays
    dead until retention prunes it."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        now = datetime.now(UTC)
        candidates = session.execute(
            select(ExternalLink).where(
                ExternalLink.status == "dead",
                ExternalLink.recoveries < _MAX_RECOVERIES,
            )
        ).scalars().all()
        recovered = 0
        for link in candidates:
            wait = timedelta(days=2 ** link.recoveries)
            if link.completed_at is not None and link.completed_at > now - wait:
                continue
            link.status = "failed"
            link.attempts = DEAD_LETTER_THRESHOLD - 1
            link.recoveries += 1
            recovered += 1
        session.commit()
    return {**sweep_external_links(), "revived": recovered}


@celery.task(name="backend.app.tasks.external.prune_external_links")
def prune_external_links() -> dict:
    """Retention: delete links recovery has given up on, once their
    last attempt is _RETENTION_DAYS old, so the ledger doesn't grow unbounded.
    Downloaded links are kept (they're the record of what we have)."""
    cutoff = datetime.now(UTC) - timedelta(days=_RETENTION_DAYS)
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        res = session.execute(
            delete(ExternalLink).where(
                ExternalLink.status == "dead",
                ExternalLink.recoveries >= _MAX_RECOVERIES,
                or_(
                    ExternalLink.completed_at.is_(None),
                    ExternalLink.completed_at < cutoff,
                ),
            )
        )
        session.commit()
    return {"pruned": res.rowcount or 0}