FabledCurator/backend/app/api/sources.py

"""FC-3a: CRUD over Source rows. FC-3c adds POST /<id>/check."""

from quart import Blueprint, jsonify, request
from sqlalchemy import select

from ..extensions import get_session
from ..models import DownloadEvent, Source
from ..services.scheduler_service import active_platform_cooldowns, scheduler_status
from ..services.source_service import (
    KNOWN_PLATFORMS,
    ArtistNotFoundError,
    DuplicateSourceError,
    EmptyUrlError,
    InvalidConfigError,
    SourceService,
    UnknownPlatformError,
)
from ._responses import error_response as _bad

sources_bp = Blueprint("sources", __name__, url_prefix="/api/sources")


@sources_bp.route("", methods=["GET"])
async def list_sources():
    artist_id_raw = request.args.get("artist_id")
    artist_id = None
    if artist_id_raw is not None:
        try:
            artist_id = int(artist_id_raw)
        except ValueError:
            return _bad("invalid_artist_id", detail="artist_id must be an integer")
    failing = request.args.get("failing", "").lower() in ("1", "true", "yes")
    async with get_session() as session:
        records = await SourceService(session).list(artist_id=artist_id, failing=failing)
    return jsonify([r.to_dict() for r in records])


@sources_bp.route("/schedule-status", methods=["GET"])
async def schedule_status():
    """FC-dashboards: scheduler health for the Subscriptions hub."""
    async with get_session() as session:
        return jsonify(await scheduler_status(session))


@sources_bp.route("/<int:source_id>", methods=["GET"])
async def get_source(source_id: int):
    async with get_session() as session:
        record = await SourceService(session).get(source_id)
    if record is None:
        return _bad("not_found", status=404, detail=f"source id={source_id}")
    return jsonify(record.to_dict())


@sources_bp.route("", methods=["POST"])
async def create_source():
    body = await request.get_json()
    if not isinstance(body, dict):
        return _bad("invalid_body", detail="body must be a JSON object")
    try:
        artist_id = int(body["artist_id"])
        platform = body["platform"]
        url = body["url"]
    except (KeyError, TypeError, ValueError):
        return _bad("invalid_body", detail="artist_id, platform, url are required")

    optional = {
        k: body[k]
        for k in ("enabled", "config_overrides", "check_interval_override")
        if k in body
    }

    async with get_session() as session:
        svc = SourceService(session)
        try:
            record = await svc.create(
                artist_id=artist_id, platform=platform, url=url, **optional,
            )
        except ArtistNotFoundError:
            return _bad("artist_not_found", status=404, detail=f"artist id={artist_id}")
        except UnknownPlatformError as exc:
            return _bad("unknown_platform", detail=str(exc), known=sorted(KNOWN_PLATFORMS))
        except InvalidConfigError as exc:
            return _bad("invalid_config", detail=str(exc))
        except EmptyUrlError as exc:
            return _bad("empty_url", detail=str(exc))
        except DuplicateSourceError as exc:
            return _bad("duplicate", status=409, existing_id=exc.existing_id)

        # Immediate kickoff: a new enabled source is armed for backfill (#693)
        # but would otherwise sit idle until the next scheduler tick (~60s).
        # Enqueue the first walk now, skipping only if the platform is in a
        # rate-limit cooldown (the scheduler picks it up when that clears).
        dispatch_id = None
        if record.enabled:
            cooldowns = await active_platform_cooldowns(session)
            if record.platform not in cooldowns:
                session.add(DownloadEvent(source_id=record.id, status="pending"))
                await session.commit()
                dispatch_id = record.id

    if dispatch_id is not None:
        from ..tasks.download import download_source
        download_source.delay(dispatch_id)
    return jsonify(record.to_dict()), 201


@sources_bp.route("/<int:source_id>", methods=["PATCH"])
async def patch_source(source_id: int):
    body = await request.get_json()
    if not isinstance(body, dict):
        return _bad("invalid_body", detail="body must be a JSON object")
    async with get_session() as session:
        svc = SourceService(session)
        try:
            record = await svc.update(source_id, **body)
        except LookupError:
            return _bad("not_found", status=404)
        except UnknownPlatformError as exc:
            return _bad("unknown_platform", detail=str(exc), known=sorted(KNOWN_PLATFORMS))
        except InvalidConfigError as exc:
            return _bad("invalid_config", detail=str(exc))
        except EmptyUrlError as exc:
            return _bad("empty_url", detail=str(exc))
        except DuplicateSourceError as exc:
            return _bad("duplicate", status=409, existing_id=exc.existing_id)
    return jsonify(record.to_dict())


@sources_bp.route("/<int:source_id>", methods=["DELETE"])
async def delete_source(source_id: int):
    async with get_session() as session:
        try:
            await SourceService(session).delete(source_id)
        except LookupError:
            return _bad("not_found", status=404)
    return "", 204


@sources_bp.route("/<int:source_id>/backfill", methods=["POST"])
async def set_backfill(source_id: int):
    """Plan #693/#697 + #830: start/stop a backfill, or start a recovery /
    recapture. Body: `{"action": "start" | "stop" | "recover" | "recapture"}`
    (default "start"). 'start' walks the full post history in time-boxed chunks
    until it reaches the bottom (then the source shows 'complete'); 'recover' is
    the same walk but bypasses the Patreon seen-ledger to re-fetch
    dropped-and-deleted near-dups under the current pHash threshold; 'recapture'
    re-grabs EVERY post's body + external links and localizes on-disk inline
    images WITHOUT re-downloading media; 'stop' cancels any back to tick mode.
    Returns the updated source dict (incl. backfill_state / backfill_chunks /
    backfill_bypass_seen / backfill_recapture)."""
    from pathlib import Path

    from ..services.credential_service import CredentialService
    from ..services.download_backends import (
        uses_native_ingester,
        verify_source_credential,
    )
    from .credentials import _get_crypto

    payload = await request.get_json(silent=True) or {}
    action = payload.get("action", "start")
    if action not in ("start", "stop", "recover", "recapture"):
        return _bad(
            "invalid_action",
            detail="action must be 'start', 'stop', 'recover', or 'recapture'",
        )

    # Pre-flight (plan #703 #2): before arming a deep walk on a native-ingester
    # platform (where verify is one cheap API page), refuse if the credential is
    # DEFINITIVELY rejected — don't burn chunks against expired cookies. Proceed
    # on valid OR inconclusive (a network blip shouldn't block). Gated to native
    # platforms: gallery-dl verify is a slow --simulate subprocess, too heavy for
    # an arm action. The credential read happens in a session that's CLOSED
    # before the verify network call (don't hold a DB conn across the request).
    if action in ("start", "recover", "recapture"):
        async with get_session() as session:
            rec = await SourceService(session).get(source_id)
            if rec is None:
                return _bad("not_found", status=404)
            native = uses_native_ingester(rec.platform)
            if native:
                cred = CredentialService(session, _get_crypto())
                cookies_path = await cred.get_cookies_path(rec.platform)
                auth_token = await cred.get_token(rec.platform)
        if native:
            ok, message = await verify_source_credential(
                platform=rec.platform,
                url=rec.url,
                artist_slug=rec.artist_slug,
                config_overrides=rec.config_overrides or {},
                cookies_path=str(cookies_path) if cookies_path else None,
                auth_token=auth_token,
                images_root=Path("/images"),
            )
            if ok is False:
                return _bad("credential_rejected", detail=message, status=409)

    async with get_session() as session:
        try:
            svc = SourceService(session)
            if action == "start":
                record = await svc.start_backfill(source_id)
            elif action == "recover":
                record = await svc.start_recovery(source_id)
            elif action == "recapture":
                record = await svc.start_recapture(source_id)
            else:
                record = await svc.stop_backfill(source_id)
        except LookupError:
            return _bad("not_found", status=404)
    return jsonify(record.to_dict())


@sources_bp.route("/<int:source_id>/preview", methods=["POST"])
async def preview_source_endpoint(source_id: int):
    """Plan #708 B4: dry-run — count what a backfill WOULD download for a native
    platform (Patreon today), without downloading. Walks the first few feed pages
    and counts media not already in the seen/dead ledgers. Returns
    {total_new, posts_scanned, pages_scanned, has_more, sample[]} or 409 + reason
    (unresolvable campaign id / auth / drift). 400 for gallery-dl platforms (no
    cheap dry-run — their verify is a slow --simulate)."""
    from pathlib import Path

    from ..services.credential_service import CredentialService
    from ..services.download_backends import preview_source, uses_native_ingester
    from ..tasks._sync_engine import sync_session_factory
    from .credentials import _get_crypto

    async with get_session() as session:
        rec = await SourceService(session).get(source_id)
        if rec is None:
            return _bad("not_found", status=404)
        if not uses_native_ingester(rec.platform):
            return _bad(
                "unsupported",
                detail="Preview is only available for native-ingester platforms.",
                status=400,
            )
        cred = CredentialService(session, _get_crypto())
        cookies_path = await cred.get_cookies_path(rec.platform)
        auth_token = await cred.get_token(rec.platform)

    # The walk + ledger reads are sync (run off the request loop); the process
    # sync engine is the same one the download task uses.
    result = await preview_source(
        platform=rec.platform,
        url=rec.url,
        source_id=source_id,
        config_overrides=rec.config_overrides or {},
        cookies_path=str(cookies_path) if cookies_path else None,
        images_root=Path("/images"),
        sync_session_factory=sync_session_factory(),
        auth_token=auth_token,
    )
    if "error" in result:
        return _bad("preview_failed", detail=result["error"], status=409)
    return jsonify(result)


@sources_bp.route("/<int:source_id>/check", methods=["POST"])
async def check_source(source_id: int):
    """FC-3c: enqueue a download for this source.

    Returns 202 with the new DownloadEvent id. If a pending/running
    event already exists for this source, returns 409 with that id. If
    the source's platform is currently in a rate-limit cooldown, returns
    **202 with `{status: "deferred", cooldown_until, platform}`** and
    does NOT create an event or dispatch — the bulk retry path uses this
    to avoid bowling N sources right back into the rate limit the
    cooldown is preventing. Single-click "retry this one source" passes
    `?force=true` to override the cooldown (operator-explicit, useful
    for rapid auth-fix testing). The in-flight guard always applies.
    """
    force = (request.args.get("force") or "").lower() in ("1", "true", "yes")
    async with get_session() as session:
        source = (await session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            return _bad("not_found", status=404, detail=f"source id={source_id}")
        if not source.enabled:
            return _bad("source_disabled", detail="enable the source first")

        # Cooldown gate (unless explicitly overridden). Checked before the
        # in-flight guard because a deferred retry doesn't need to create
        # or check for an event at all.
        if not force:
            cooldowns = await active_platform_cooldowns(session)
            expires_at = cooldowns.get(source.platform)
            if expires_at is not None:
                return jsonify({
                    "status": "deferred",
                    "platform": source.platform,
                    "cooldown_until": expires_at.isoformat(),
                }), 202

        in_flight = (await session.execute(
            select(DownloadEvent.id).where(
                DownloadEvent.source_id == source_id,
                DownloadEvent.status.in_(["pending", "running"]),
            ).order_by(DownloadEvent.id.desc()).limit(1)
        )).scalar_one_or_none()
        if in_flight is not None:
            return jsonify(
                {"download_event_id": in_flight, "status": "already_running"}
            ), 409

        event = DownloadEvent(source_id=source_id, status="pending")
        session.add(event)
        await session.commit()
        await session.refresh(event)
        event_id = event.id

    from ..tasks.download import download_source
    download_source.delay(source_id)

    return jsonify({"download_event_id": event_id, "status": "pending"}), 202