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, 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; an operator recovery still re-attempts). 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
# Dead rows older than this are pruned (retention).
_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
_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")
            .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:
        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()
                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()
        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 / archive captured as a PostAttachment (copied into the
            # store) — drop the on-disk original. An archive captured WITHOUT
            # extracting any image carries the reason (the recurring "zip but no
            # images" symptom, #718) — surface it loudly.
            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)
            f.unlink(missing_ok=True)
        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,
            )
            if reason in ("duplicate_hash", "duplicate_phash"):
                # Path-safe unlink: external files are imported IN PLACE, so `f`
                # can BE the canonical record's backing file (same content
                # re-fetched / a colliding name). Only delete `f` when it is NOT
                # the existing record's file — otherwise we orphan the record and
                # playback 404s. (#859)
                canonical = None
                if result.image_id is not None:
                    rec = session.get(ImageRecord, result.image_id)
                    if rec is not None:
                        canonical = (IMAGES_ROOT / rec.path).resolve()
                if canonical != f.resolve():
                    f.unlink(missing_ok=True)  # a redundant copy — safe to drop
                else:
                    log.info(
                        "extdl keep: %s IS record %s's canonical file — not unlinking",
                        f.name, result.image_id,
                    )
        elif result.status == "failed":
            log.warning(
                "extdl import FAILED: post=%s file=%s error=%s",
                post.id, f.name, result.error,
            )
            f.unlink(missing_ok=True)
        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,
            )
        # The synthesized sidecar has done its job — don't litter the tree.
        Path(str(f) + ".json").unlink(missing_ok=True)
    return image_ids


@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."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        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)}


@celery.task(name="backend.app.tasks.external.recover_external_links")
def recover_external_links() -> dict:
    """Recovery sweep (rule #89): reset dead links back to retryable so a stuck
    host outage or a since-fixed bug gets another pass, then enqueue. Bounded."""
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        session.execute(
            update(ExternalLink).where(ExternalLink.status == "dead")
            .values(status="failed", attempts=0, last_error=None)
        )
        session.commit()
    return sweep_external_links()


@celery.task(name="backend.app.tasks.external.prune_external_links")
def prune_external_links() -> dict:
    """Retention (rule #89): delete long-dead links 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.created_at < cutoff,
            )
        )
        session.commit()
    return {"pruned": res.rowcount or 0}