FabledCurator/backend/app/services/gallery_dl.py

"""Gallery-dl subprocess wrapper.

Ported from GallerySubscriber (~/Nextcloud/Projects/GallerySubscriber/
backend/app/services/gallery_dl.py). FC adaptations:
  - `artist_slug` replaces GS's `subscription_name` throughout
  - Validation uses backend/app/services/file_validator.py
  - Config dir under `<images_root>/.gallery-dl/`, not a separate config_path
  - PLATFORM_DEFAULTS scoped to the GS-6 platforms registered in FC-3b
"""

from __future__ import annotations

import asyncio
import json
import logging
import subprocess
import sys
import tempfile
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path

from .file_validator import is_validatable, quarantine_file, validate_file

log = logging.getLogger(__name__)


class ErrorType(StrEnum):
    AUTH_ERROR = "auth_error"
    RATE_LIMITED = "rate_limited"
    NOT_FOUND = "not_found"
    ACCESS_DENIED = "access_denied"
    NETWORK_ERROR = "network_error"
    NO_NEW_CONTENT = "no_new_content"
    TIER_LIMITED = "tier_limited"
    TIMEOUT = "timeout"
    HTTP_ERROR = "http_error"
    UNSUPPORTED_URL = "unsupported_url"
    VALIDATION_FAILED = "validation_failed"
    # Native Patreon ingester only (plan #697): a response parsed as JSON but
    # didn't match the JSON:API shape the ingester depends on (missing `data`,
    # media lacking `file_name`/`url`, etc.). Distinct from AUTH_ERROR — the fix
    # is updating the ingester's field-set/parser, not rotating credentials. The
    # contract test guards against silently shipping this.
    API_DRIFT = "api_drift"
    # Run made real progress (downloaded ≥1 file) but did not finish in the
    # subprocess budget. Distinct from UNKNOWN_ERROR — the downstream status
    # mapping classifies this as "ok" because the next tick continues.
    PARTIAL = "partial"
    UNKNOWN_ERROR = "unknown_error"


# Tick mode (routine cron polls): skip ≤20 contiguous already-archived
# items, then exit gallery-dl. Established subscription with zero new
# content exits in ~30s of HEAD requests instead of walking to the bottom
# of the post history (which can be hours for prolific creators). 20 (not
# 5) is operator-set headroom against any edge case where paywalled or
# otherwise-non-downloadable items might interleave with archived ones —
# 20 contiguous HEADs is still negligible.
TICK_SKIP_VALUE = "exit:20"

# Backfill mode (operator-triggered deep scan): walk the full history,
# one TIME-BOXED CHUNK per run (plan #693). config_overrides["_backfill_state"]
# == "running" selects this mode; the cursor checkpoint (plan #689) lets each
# chunk resume where the last stopped, so the walk advances across chunks
# until gallery-dl exits cleanly (= reached the bottom).
#
# The chunk budget is deliberately FAR below download_source's Celery
# soft_time_limit (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py), not
# "just under" it. Hitting this budget is the NORMAL chunk boundary, not a
# failure: subprocess.run raises TimeoutExpired (which captures partial
# stdout/stderr + the last emitted cursor), and download_service reclassifies
# a chunk that made progress as PARTIAL (status "ok"), not an error. The huge
# headroom means a stuck file or a slow chunk can never let Celery's
# SoftTimeLimitExceeded preempt TimeoutExpired (the failure mode behind Knuxy
# #38275 / Anduo #39912/#40411). Earlier we ran one ~1170s run-to-the-wall
# per arming; that died as a timeout error every time on large catalogs.
BACKFILL_SKIP_VALUE = True
BACKFILL_CHUNK_SECONDS = 600


# Sits well below download_source's Celery soft_time_limit
# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py). subprocess.run MUST
# raise TimeoutExpired before Celery raises SoftTimeLimitExceeded —
# otherwise Celery wins the race, SIGKILLs the worker, in-memory
# stdout/stderr is lost, and the DownloadEvent ends up empty-logged with
# "stranded by recovery sweep" (operator-flagged 2026-05-31, Knuxy event
# #38275; recurred in backfill mode as Anduo #39912). Per-source bumps
# still live in source.config_overrides for legitimately long syncs —
# keep any override below the soft limit, or the soft-limit salvage path
# in tasks/download.py (_finalize_soft_limited) is the only safety net.
_DEFAULT_GDL_TIMEOUT_SECONDS = 870


# --- Discord naming ---------------------------------------------------------
#
# Derived from a REAL sidecar (operator's instance, 2026-09-13), not from memory
# of gallery-dl's extractor. What gallery-dl's discord extractor actually emits
# for an attachment: `channel` is a plain STRING (the channel's name), the
# message is `message_id`, the attachment's position in it is `num`, and there
# is NO `id` key at all.
#
# The previous patterns asked for `{channel[name]}` and `{id}`. Both render as
# "None", so every Discord download since the platform was added landed in a
# directory called `None` as `<date>_None_<original name>`. Worse, the sidecar was
# named `{filename}.json` — the attachment's ORIGINAL name — which (a) `find_
# sidecar` can never pair with `<date>_None_<name>.png`, so no Discord file ever
# got a Post or a post date, and (b) collides: every `image.png` in a channel
# overwrote the same `image.json`, so the one sidecar that survived described
# whichever message happened to be written last.
#
# The fix names the sidecar EXACTLY like the media minus its extension, so
# `find_sidecar`'s first candidate (`media.with_suffix(".json")`) is the match
# and the name is unique per attachment. tests/test_gallery_dl_naming.py renders
# these patterns against a sanitized copy of the real sidecar, so a key that
# does not exist fails CI instead of silently becoming "None".
DISCORD_FILENAME = "{date:%Y%m%d}_{message_id}_{num:>02}_{filename}.{extension}"
DISCORD_DIRECTORY = ["{channel}"]


def sidecar_name_for(media_pattern: str) -> str | None:
    """The metadata filename pattern that names a sidecar exactly like its media.

    Returns None for a pattern that does not end in `.{extension}`, since then
    there is no media stem to mirror and the caller must fall back.
    """
    suffix = ".{extension}"
    if not media_pattern.endswith(suffix):
        return None
    return media_pattern[: -len(suffix)] + ".json"


def metadata_postprocessor(filename: str) -> dict:
    return {"name": "metadata", "mode": "json", "directory": ".", "filename": filename}


def archive_path(images_root: Path) -> Path:
    """gallery-dl's download archive: the record of what it has already fetched.

    One definition, because the Discord repair (services/discord_repair.py) has
    to find the same file the downloader writes, without constructing a service
    whose __init__ creates directories.
    """
    return Path(images_root) / ".gallery-dl" / "archive.sqlite3"


@dataclass
class SourceConfig:
    """Per-source overrides loaded from Source.config_overrides JSON.

    Note: the gallery-dl `skip` value (tick vs backfill, see TICK_SKIP_VALUE /
    BACKFILL_SKIP_VALUE) is NOT carried here — it derives from the
    Source.backfill_runs_remaining column at the download_service layer
    and is passed to _build_config_for_source as `skip_value`. Same for
    the per-run subprocess timeout.

    `resume_cursor` is RUNTIME state, not persisted operator config: the
    cursor-paged backfill checkpoint (see parse_last_cursor + the
    download_service backfill lifecycle). It is consumed only by the native
    Patreon ingester (the sole cursor-paged platform; gallery-dl's Patreon path
    was removed at the #697 cutover): on a backfill/recovery run the ingester
    resumes its newest→oldest walk from that pagination cursor instead of
    restarting from the top. download_service threads it from
    config_overrides["_backfill_cursor"]; SourceConfig deliberately does NOT read
    it in from_dict (config_overrides also holds operator config, and
    resume_cursor must only apply in backfill/recovery mode).
    """
    content_types: list[str] = field(default_factory=lambda: ["all"])
    sleep: float | None = None
    sleep_request: float | None = None
    directory_pattern: str | None = None
    filename_pattern: str | None = None
    save_metadata: bool = True
    timeout: int = _DEFAULT_GDL_TIMEOUT_SECONDS
    resume_cursor: str | None = None

    @classmethod
    def from_dict(cls, data: dict) -> SourceConfig:
        return cls(
            content_types=data.get("content_types", ["all"]),
            sleep=data.get("sleep"),
            sleep_request=data.get("sleep_request"),
            directory_pattern=data.get("directory_pattern"),
            filename_pattern=data.get("filename_pattern"),
            save_metadata=data.get("save_metadata", True),
            timeout=data.get("timeout", _DEFAULT_GDL_TIMEOUT_SECONDS),
        )


@dataclass
class DownloadResult:
    success: bool
    url: str
    artist_slug: str
    platform: str
    files_downloaded: int = 0
    files_quarantined: int = 0
    quarantined_paths: list[str] = field(default_factory=list)
    written_paths: list[str] = field(default_factory=list)
    # Native ingester only: post-only sidecar paths for posts that have NO
    # downloadable media (pure-text posts), so the importer can still upsert the
    # Post + its body. Empty on the gallery-dl path. Phase 3 imports these via
    # Importer.upsert_post_record (keyed on external_post_id → updates, never
    # doubles, the same Post a media import would create).
    post_record_paths: list[str] = field(default_factory=list)
    # Native ingester, recapture mode only (#830): (on-disk path, CDN source_url)
    # pairs for already-present media whose ImageRecord.source_filehash should be
    # backfilled (inline-image localization) WITHOUT re-download or unlink. Empty
    # on the gallery-dl path and outside recapture.
    relink_source_paths: list[tuple[str, str, str]] = field(default_factory=list)
    stdout: str = ""
    stderr: str = ""
    return_code: int = 0
    error_type: ErrorType | None = None
    error_message: str | None = None
    duration_seconds: float = 0.0
    started_at: str | None = None
    completed_at: str | None = None
    # Plan #704 — structured fields the NATIVE ingester populates directly (None
    # on the gallery-dl path, which keeps the regex-over-stdout route). When set,
    # phase 3 reads these instead of scraping the text it would otherwise have to
    # reconstruct: `run_stats` mirrors _compute_run_stats' shape; `cursor` is the
    # backfill checkpoint the ingester knows exactly (no parse_last_cursor).
    run_stats: dict | None = None
    cursor: str | None = None
    posts_processed: int = 0
    # Plan #708 B1 — the server's Retry-After seconds on a RATE_LIMITED result, so
    # the platform cooldown matches the hint instead of a flat default. None when
    # unknown (no header, or not a rate-limit failure).
    retry_after_seconds: float | None = None
    # Native ingester only: marks this walk's fetched media seen in its ledger.
    # Phase 3 calls it AFTER the import loop, never before — a file marked seen
    # but not yet imported is invisible to every later walk, so a run killed in
    # between orphaned it for good (TamadaHeijun's 12PCG post lost 7 of 13
    # images to a stranded run, 2026-09-24). Unmarked, the next walk finds the
    # file on disk with no ImageRecord and imports it. None on gallery-dl.
    mark_seen_after_import: Callable[[], None] | None = None


def extract_errors_warnings(stderr: str) -> str:
    """Keep only the `[error]`/`[warning]` lines from a captured stderr blob.

    A generic download-log helper (module-level so the native-ingester result
    path can shape its logs without reaching through a GalleryDLService instance).
    """
    if not stderr:
        return ""
    kept = [
        line for line in stderr.splitlines()
        if "][error]" in line.lower() or "][warning]" in line.lower()
    ]
    return "\n".join(kept)


def truncate_log(text: str, max_bytes: int = 500_000) -> str:
    """Cap a captured log to `max_bytes`, eliding the middle (head + tail kept)."""
    if not text:
        return text
    encoded = text.encode("utf-8")
    if len(encoded) <= max_bytes:
        return text
    half = max_bytes // 2
    head = encoded[:half].decode("utf-8", errors="ignore")
    tail = encoded[-half:].decode("utf-8", errors="ignore")
    head_lines = head.count("\n")
    tail_lines = tail.count("\n")
    total_lines = text.count("\n")
    elided = max(0, total_lines - head_lines - tail_lines)
    marker = f"\n\n... [{elided} lines elided, {len(encoded) - max_bytes} bytes] ...\n\n"
    return head + marker + tail


def _summarize_validation_failures(failures: list[dict]) -> str:
    if not failures:
        return "Validation failed"
    reasons: dict[str, int] = {}
    for f in failures:
        key = f.get("reason") or "unknown"
        reasons[key] = reasons.get(key, 0) + 1
    top_reason, top_count = max(reasons.items(), key=lambda kv: kv[1])
    n = len(failures)
    if top_count == n:
        return f"{n} file{'s' if n != 1 else ''} quarantined: {top_reason}"
    return f"{n} files quarantined ({top_count}× {top_reason}, mixed)"


# (parse_last_cursor was removed in plan #704: the native ingester now carries
# its checkpoint cursor as a structured DownloadResult.cursor field, so there is
# no log text to scrape — and gallery-dl platforms never had a cursor.)


def make_run_stats(
    *,
    exit_code: int = 0,
    downloaded_count: int = 0,
    skipped_count: int = 0,
    per_item_failures: int = 0,
    warning_count: int = 0,
    tier_gated_count: int = 0,
    quarantined_count: int = 0,
    dead_lettered_count: int = 0,
) -> dict:
    """The canonical `run_stats` dict shape, in ONE place.

    Both result producers — gallery-dl's `_compute_run_stats` (log scrape) and the
    native ingester's per-outcome tally (ingest_core) — build through this so the
    key set can't drift between backends. Phase 3 + the Logs UI read these keys.
    """
    return {
        "exit_code": exit_code,
        "downloaded_count": downloaded_count,
        "skipped_count": skipped_count,
        "per_item_failures": per_item_failures,
        "warning_count": warning_count,
        "tier_gated_count": tier_gated_count,
        "quarantined_count": quarantined_count,
        "dead_lettered_count": dead_lettered_count,
    }


# --- tier-gated classification, shared by BOTH backends ---------------------
#
# These three live together because the native ingester and the gallery-dl
# subprocess must reach the same verdict from the same number. They did not:
# gallery-dl classified TIER_LIMITED while ingest_core counted gated posts and
# threw the count away, so the platforms FC owns reported a paywalled creator as
# a silent one (#874 follow-up). One predicate, spread into both, rather than
# the condition re-derived per backend.


def classify_tier_gated(tier_gated_count: int) -> ErrorType | None:
    """TIER_LIMITED when a walk saw tier-gated posts and nothing else failed.

    Deliberately NOT conditioned on `downloaded == 0`. A creator whose top-tier
    posts we cannot see is tier-limited even in a week we did get their cheaper
    ones — the fact the operator needs ("there is content here you are not
    paying for") is true either way. gallery-dl has classified it this way since
    the paywall-as-"needs attention" complaint (see `_categorize_error`), and the
    native path now matches rather than inventing a stricter rule.

    Callers must apply this only AFTER the real error categories (auth, rate
    limit, drift, …) have had their turn; tier-gating is the weakest signal and
    must never mask a genuine failure.
    """
    return ErrorType.TIER_LIMITED if tier_gated_count else None


def tier_gated_message(count: int) -> str:
    """The one wording for the tier-gated verdict, so the two backends can't
    describe the same state differently in the Logs UI."""
    return (
        f"Subscription tier does not grant access to "
        f"{count} post{'s' if count != 1 else ''}"
    )


# `Source.error_type` doubles as the failure-class chip, and a status of "ok"
# CLEARS it (alembic 0032). TIER_LIMITED breaks that assumption: it rides an
# otherwise-successful run, so without an exemption the chip is wiped the moment
# it is set and `FailingSourcesCard`'s `tier_limited` palette entry can never
# render. Informational classes are the exemption — they describe the source,
# not a failure of the run.
INFORMATIONAL_ERROR_TYPES = frozenset({ErrorType.TIER_LIMITED.value})


def is_informational(error_type) -> bool:
    """True for a class that reports a state rather than a failure. Accepts an
    ErrorType or the plain string persisted on Source.error_type."""
    return error_type is not None and str(error_type) in INFORMATIONAL_ERROR_TYPES


def walk_completed(result: DownloadResult) -> bool:
    """Did this walk reach the bottom cleanly?

    The backfill lifecycle's completion test. An informational error_type still
    counts as complete: a fully-paywalled creator's backfill DID finish, and
    treating it as unfinished re-walks the same wall until the stall counter
    trips — the creator we can see least becoming the one we fetch most.
    """
    return (
        result.success
        and result.return_code == 0
        and (result.error_type is None or is_informational(result.error_type))
    )


class GalleryDLService:
    """Service for executing gallery-dl downloads."""

    AUTH_ERROR_PATTERNS = [
        "unauthorized", "login required", "please login", "must be logged in",
        "authentication required", "authenticationerror", "refresh-token",
        "session expired", "invalid cookie", "cookies are expired",
        "cookies have expired", "not logged in",
    ]
    RATE_LIMIT_PATTERNS = [
        '" 429 ', "rate limit", "too many requests", "ratelimit", "throttl",
    ]
    NOT_FOUND_PATTERNS = [
        '" 404 ', "error 404", "status: 404", "status code: 404",
        "404 not found", "page not found", "user not found",
        "creator not found", "artist not found",
        "content no longer available", "has been deleted",
        "account deleted", "profile does not exist",
    ]
    GENERIC_NOT_FOUND_PATTERNS = ["does not exist", "no longer available"]
    NETWORK_ERROR_PATTERNS = [
        "timed out", "read timed out", "connect timed out",
        "connection timed out", "connection refused", "connection reset",
        "network unreachable", "name resolution failed",
        "name or service not known", "nodename nor servname provided",
        "ssl: certificate_verify_failed", "ssl: wrong_version_number",
        "certificate verify failed", "[errno",
    ]
    ACCESS_DENIED_PATTERNS = [
        '" 403 ', "error 403", "forbidden", "access denied",
        "permission denied", "tier required", "pledge required",
    ]

    # Per-platform defaults for the gallery-dl-backed platforms. Patreon was
    # removed at the plan-#697 cutover — it now uses the native ingester
    # (services/patreon_ingester.py), not gallery-dl.
    PLATFORM_DEFAULTS = {
        # subscribestar removed — native-ingester platform now (#71); pixiv
        # removed likewise (#129); deviantart removed at #3069 as a dropped
        # platform, not a migrated one. The remaining entries are the
        # gallery-dl platforms not yet migrated.
        "hentaifoundry": {
            "content_types": ["all"],
            "directory": [],
            "filename": "{category}_{index:>03}_{title[:50]}.{extension}",
            "include": "all",
        },
        "discord": {
            "content_types": ["all"],
            "directory": DISCORD_DIRECTORY,
            "filename": DISCORD_FILENAME,
            # Overrides the global `{filename}.json` sidecar for this extractor
            # only — see the Discord naming note above.
            "postprocessors": [metadata_postprocessor(sidecar_name_for(DISCORD_FILENAME))],
            "embeds": "all",
            "stickers": True,
            "reactions": False,
            "threads": True,
        },
    }

    def __init__(
        self,
        images_root: Path,
        rate_limit: float = 3.0,
        validate_files: bool = True,
    ):
        self.images_root = Path(images_root)
        self._rate_limit = rate_limit
        self._validate_files = validate_files
        self._config_dir = self.images_root / ".gallery-dl"
        self._config_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
        (self._config_dir / "temp").mkdir(parents=True, exist_ok=True, mode=0o700)

    def _get_default_config(self) -> dict:
        config = {
            "extractor": {
                "base-directory": str(self.images_root),
                "archive": str(archive_path(self.images_root)),
                "skip": True,
                "sleep": self._rate_limit,
                "sleep-request": max(0.5, self._rate_limit / 4),
                # 2 (not 3) retries — a stuck CDN host shouldn't burn a whole
                # backfill chunk on one file. Anduo #40838 spent ~600s on a
                # single image (4 extractor HEAD retries × 30s + 4 downloader
                # GET retries × 120s); halving retries + the downloader
                # timeout below caps a wedged file at ~1-2 min instead.
                "retries": 2,
                "timeout": 30.0,
                "verify": True,
                "postprocessors": [
                    {
                        "name": "metadata",
                        "mode": "json",
                        "directory": ".",
                        "filename": "{filename}.json",
                    }
                ],
            },
            "downloader": {
                "part": True,
                "part-directory": str(self._config_dir / "temp"),
                # See the extractor retries note above (Anduo #40838): 2
                # retries + a 60s read-timeout (was 120) so a stalled
                # connection fails fast. 60s is a per-read timeout, not a
                # transfer cap — large GIFs that keep streaming are unaffected.
                "retries": 2,
                "timeout": 60.0,
                # NOTE: the Patreon/Mux yt-dlp Referer/Origin forwarding lived
                # here until the plan-#697 cutover. It was Patreon-specific (and
                # would have wrongly tagged the other platforms' yt-dlp fetches);
                # Patreon video is now handled by the native ingester's
                # downloader (patreon_downloader._VIDEO_HEADERS), so it's gone.
            },
            "output": {"progress": True},
        }
        for platform_name, defaults in self.PLATFORM_DEFAULTS.items():
            config["extractor"][platform_name] = {
                key: value for key, value in defaults.items() if key != "content_types"
            }
        return config

    def _build_config_for_source(
        self,
        platform: str,
        source_config: SourceConfig,
        artist_slug: str,
        skip_value: bool | str = BACKFILL_SKIP_VALUE,
    ) -> dict:
        """`skip_value` controls gallery-dl's archive-walk behavior:
            - True (BACKFILL_SKIP_VALUE): walk full post history, skipping
              archived items but continuing past them. Used in backfill mode.
            - "exit:20" (TICK_SKIP_VALUE): exit gallery-dl after 20
              contiguous archived items. Used in tick (routine catch-up)
              mode for fast no-op syncs on creators with deep history.
            - False: don't skip — redownload everything (not used in FC).
        The caller (download_service) chooses based on
        Source.backfill_runs_remaining.
        """
        config = json.loads(json.dumps(self._get_default_config()))  # deep copy

        destination = str(self.images_root / artist_slug / platform)
        config["extractor"]["base-directory"] = destination

        if source_config.sleep is not None:
            config["extractor"]["sleep"] = source_config.sleep
        if source_config.sleep_request is not None:
            config["extractor"]["sleep-request"] = source_config.sleep_request
        config["extractor"]["skip"] = skip_value

        if source_config.save_metadata:
            config["extractor"]["postprocessors"] = [
                {
                    "name": "metadata",
                    "mode": "json",
                    "directory": ".",
                    "filename": "{filename}.json",
                }
            ]
        else:
            config["extractor"].pop("postprocessors", None)

        platform_section = config["extractor"].setdefault(platform, {})

        if platform == "hentaifoundry":
            if "pictures" in source_config.content_types or "all" in source_config.content_types:
                platform_section["include"] = "all"

        if source_config.directory_pattern:
            platform_section["directory"] = source_config.directory_pattern
        if source_config.filename_pattern:
            platform_section["filename"] = source_config.filename_pattern

        # A platform that names its sidecar after its media must keep doing so
        # under a per-source filename override, or the pairing breaks exactly the
        # way Discord's did. No metadata wanted means no platform postprocessor
        # either — the global list was already dropped above.
        if "postprocessors" in platform_section:
            mirrored = sidecar_name_for(platform_section.get("filename") or "")
            if not source_config.save_metadata or mirrored is None:
                platform_section.pop("postprocessors")
            else:
                platform_section["postprocessors"] = [metadata_postprocessor(mirrored)]

        platform_section["metadata"] = source_config.save_metadata

        return config

    def _categorize_error(
        self, return_code: int, stdout: str, stderr: str,
    ) -> tuple[ErrorType, str]:
        combined = f"{stdout} {stderr}".lower()

        skip_line_count = len([
            line for line in stdout.split("\n") if line.strip().startswith("#")
        ])
        skip_text_indicators = ["skipping", "already exists", "archive"]
        has_skip_text = any(ind in combined for ind in skip_text_indicators)

        actual_error_indicators = [
            "][error]", "exception", "traceback",
            "download failed", "extraction failed",
        ]
        per_item_patterns = [
            "not allowed to view", "unable to get post",
            "failed to extract campaign id", "failed to download",
            "cannot import yt-dlp", "cannot import youtube-dl",
        ]
        all_lines = combined.split("\n")
        error_lines = [
            line for line in all_lines
            if any(ind in line for ind in actual_error_indicators)
        ]
        per_item_error_lines = [
            line for line in error_lines
            if any(p in line for p in per_item_patterns)
        ]
        source_level_error_lines = [
            line for line in error_lines
            if line not in set(per_item_error_lines)
        ]
        has_actual_error = bool(source_level_error_lines)

        if per_item_error_lines:
            per_item_set = set(per_item_error_lines)
            combined = "\n".join(line for line in all_lines if line not in per_item_set)

        if (skip_line_count > 0 or has_skip_text) and not has_actual_error:
            return ErrorType.NO_NEW_CONTENT, "No new content to download"

        if return_code == 0 and not stdout.strip():
            return ErrorType.NO_NEW_CONTENT, "No new content to download"

        has_401_error = (
            '" 401 ' in combined
            or "401 unauthorized" in combined
            or "error 401" in combined
            or "status: 401" in combined
            or "status code: 401" in combined
        )
        if has_401_error or any(p in combined for p in self.AUTH_ERROR_PATTERNS):
            return ErrorType.AUTH_ERROR, "Authentication failed — cookies may be expired or invalid"

        if any(p in combined for p in self.RATE_LIMIT_PATTERNS):
            return ErrorType.RATE_LIMITED, "Rate limited by server — try increasing sleep time"

        has_specific_404 = any(p in combined for p in self.NOT_FOUND_PATTERNS)
        has_generic_with_error = any(
            p in combined and "][error]" in combined
            for p in self.GENERIC_NOT_FOUND_PATTERNS
        )
        if has_specific_404 or has_generic_with_error:
            return ErrorType.NOT_FOUND, "URL not found — artist may have changed username or deleted content"

        if any(p in combined for p in self.NETWORK_ERROR_PATTERNS):
            return ErrorType.NETWORK_ERROR, "Network error — check internet connection"

        if any(p in combined for p in self.ACCESS_DENIED_PATTERNS):
            return ErrorType.ACCESS_DENIED, "Access denied — may need higher subscription tier"

        if "http error" in combined or "httperror" in combined:
            return ErrorType.HTTP_ERROR, "HTTP error occurred"

        if "no suitable" in combined or "unsupported" in combined or "no extractor" in combined:
            return ErrorType.UNSUPPORTED_URL, "URL format not supported by gallery-dl"

        if return_code in (1, 4) and (skip_line_count > 0 or has_skip_text) and not has_actual_error:
            return ErrorType.NO_NEW_CONTENT, "No new content to download"

        # Tier-gated classification used to require `return_code in (1, 4)`,
        # which silently fell through to UNKNOWN_ERROR when gallery-dl
        # returned a different exit code for mixed-failure runs (e.g.
        # paywall warnings + a missing yt-dlp dep flipping the exit bits).
        # The artist then surfaced as "needs attention" purely because a
        # paywall blocked posts the operator wasn't paying to see —
        # operator-flagged 2026-05-31. Now: if no source-level error
        # category fired AND tier-gated warnings are present, classify
        # as TIER_LIMITED regardless of return code. Same priority order
        # as before (auth/rate/access/not_found/network/http still win).
        if not has_actual_error:
            tier_gated_lines = [
                line for line in combined.split("\n")
                if "][warning]" in line and "not allowed to view post" in line
            ]
            # Same predicate + wording the native path uses, so the two backends
            # can't drift on what counts as tier-gated or how it reads.
            count = len(tier_gated_lines)
            gated = classify_tier_gated(count)
            if gated is not None:
                return (gated, tier_gated_message(count))

        # Partial-success: the subprocess exited non-zero (typically because
        # the wall-clock timeout fired mid-walk), but it had downloaded ≥1
        # file by then and no source-level error category fired. The work
        # the run DID do is real; gallery-dl's archive will pick up where
        # it left off on the next tick. Mapped to status="ok" downstream
        # (download_service.py) so this doesn't flag the source as
        # "needs attention." Operator-flagged 2026-06-01 after a Knuxy
        # patreon run downloaded hundreds of files then ran red on timeout.
        files_downloaded = self._count_downloaded_files(stdout)
        if not has_actual_error and files_downloaded > 0:
            return (
                ErrorType.PARTIAL,
                f"Downloaded {files_downloaded} file{'s' if files_downloaded != 1 else ''}; "
                "run did not complete in budget — next tick will continue",
            )

        return ErrorType.UNKNOWN_ERROR, f"Unknown error (return code: {return_code})"

    def _count_downloaded_files(self, stdout: str) -> int:
        if not stdout:
            return 0
        return sum(1 for line in stdout.splitlines() if line.strip().startswith("/"))

    def _written_paths(self, stdout: str) -> list[Path]:
        if not stdout:
            return []
        return [
            Path(stripped)
            for line in stdout.splitlines()
            if (stripped := line.strip()).startswith("/")
        ]

    def _validate_and_quarantine(
        self,
        written_paths: list[Path],
        artist_slug: str,
        platform: str,
        url: str,
    ) -> tuple[list[str], list[dict]]:
        quarantined_relpaths: list[str] = []
        failures: list[dict] = []
        if not written_paths:
            return quarantined_relpaths, failures

        for path in written_paths:
            if not is_validatable(path):
                continue
            try:
                result = validate_file(path)
            except Exception as exc:
                log.warning("Validator raised on %s: %s", path, exc)
                continue
            if result.ok:
                continue
            # Shared move+sidecar (file_validator.quarantine_file) — same impl the
            # native ingester uses, so the layout + provenance sidecar can't drift.
            dest = quarantine_file(
                self.images_root, path, artist_slug, platform,
                url=url, result=result,
            )
            if dest is None:
                continue  # move failed → left in place, not counted
            log.warning(
                "Quarantined corrupt file: %s → %s (%s: %s)",
                path, dest, result.format, result.reason,
            )
            quarantined_relpaths.append(str(dest))
            failures.append(
                {
                    "path": str(path),
                    "format": result.format,
                    "reason": result.reason,
                    "size": result.size,
                }
            )
        return quarantined_relpaths, failures

    def _compute_run_stats(self, return_code: int, stdout: str, stderr: str) -> dict:
        stdout = stdout or ""
        stderr = stderr or ""

        skipped_stdout = sum(
            1 for line in stdout.splitlines() if line.strip().startswith("#")
        )
        skipped_stderr = sum(
            1 for line in stderr.splitlines() if "] skipping " in line.lower()
        )
        per_item_failures = sum(
            1 for line in stderr.splitlines()
            if "[download][error]" in line.lower() and "failed to download" in line.lower()
        )
        warning_count = sum(
            1 for line in stderr.splitlines() if "][warning]" in line.lower()
        )
        tier_gated_count = sum(
            1 for line in stderr.splitlines()
            if "][warning]" in line.lower() and "not allowed to view post" in line.lower()
        )

        return make_run_stats(
            exit_code=return_code,
            downloaded_count=self._count_downloaded_files(stdout),
            skipped_count=skipped_stdout + skipped_stderr,
            per_item_failures=per_item_failures,
            warning_count=warning_count,
            tier_gated_count=tier_gated_count,
        )

    @staticmethod
    def _extract_errors_warnings(stderr: str) -> str:
        # Thin delegator to the module-level helper (kept for existing callers).
        return extract_errors_warnings(stderr)

    @staticmethod
    def _truncate_log(text: str, max_bytes: int = 500_000) -> str:
        # Thin delegator to the module-level helper (kept for existing callers).
        return truncate_log(text, max_bytes)

    async def download(
        self,
        url: str,
        artist_slug: str,
        platform: str,
        source_config: SourceConfig | None = None,
        cookies_path: str | None = None,
        auth_token: str | None = None,
        skip_value: bool | str = BACKFILL_SKIP_VALUE,
    ) -> DownloadResult:
        start_time = time.time()
        started_at = datetime.now(UTC).isoformat()

        if source_config is None:
            source_config = SourceConfig()

        config = self._build_config_for_source(
            platform, source_config, artist_slug, skip_value=skip_value,
        )

        if cookies_path:
            config["extractor"]["cookies"] = cookies_path
        if auth_token and platform == "discord":
            config["extractor"].setdefault("discord", {})
            config["extractor"]["discord"]["token"] = auth_token

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, dir=str(self._config_dir),
        ) as fh:
            json.dump(config, fh, indent=2)
            temp_config_path = fh.name

        try:
            cmd = [
                sys.executable, "-m", "gallery_dl",
                "--config", temp_config_path, "--verbose", url,
            ]
            log.info("Executing gallery-dl for %s/%s: %s", artist_slug, platform, url)

            loop = asyncio.get_running_loop()
            proc = await loop.run_in_executor(
                None,
                lambda: subprocess.run(
                    cmd, capture_output=True, text=True,
                    timeout=source_config.timeout,
                ),
            )

            duration = time.time() - start_time
            completed_at = datetime.now(UTC).isoformat()

            quarantined_paths: list[str] = []
            quarantine_failures: list[dict] = []
            if self._validate_files:
                quarantined_paths, quarantine_failures = self._validate_and_quarantine(
                    written_paths=self._written_paths(proc.stdout),
                    artist_slug=artist_slug,
                    platform=platform,
                    url=url,
                )

            written_paths = [
                str(p) for p in self._written_paths(proc.stdout)
                if str(p) not in set(quarantined_paths)
            ]
            files_written = self._count_downloaded_files(proc.stdout)
            files_quarantined = len(quarantined_paths)
            files_kept = max(0, files_written - files_quarantined)

            if proc.returncode == 0:
                if files_quarantined > 0:
                    return DownloadResult(
                        success=False, url=url, artist_slug=artist_slug, platform=platform,
                        files_downloaded=files_kept,
                        files_quarantined=files_quarantined,
                        quarantined_paths=quarantined_paths,
                        written_paths=written_paths,
                        stdout=proc.stdout, stderr=proc.stderr,
                        return_code=proc.returncode,
                        error_type=ErrorType.VALIDATION_FAILED,
                        error_message=_summarize_validation_failures(quarantine_failures),
                        duration_seconds=duration,
                        started_at=started_at, completed_at=completed_at,
                    )
                return DownloadResult(
                    success=True, url=url, artist_slug=artist_slug, platform=platform,
                    files_downloaded=files_kept,
                    written_paths=written_paths,
                    stdout=proc.stdout, stderr=proc.stderr,
                    return_code=proc.returncode,
                    duration_seconds=duration,
                    started_at=started_at, completed_at=completed_at,
                )

            error_type, error_message = self._categorize_error(
                proc.returncode, proc.stdout, proc.stderr
            )
            success = error_type in (ErrorType.NO_NEW_CONTENT, ErrorType.TIER_LIMITED)

            if files_quarantined > 0:
                error_type = ErrorType.VALIDATION_FAILED
                error_message = _summarize_validation_failures(quarantine_failures)
                success = False

            return DownloadResult(
                success=success, url=url, artist_slug=artist_slug, platform=platform,
                files_downloaded=files_kept if success else 0,
                files_quarantined=files_quarantined,
                quarantined_paths=quarantined_paths,
                written_paths=written_paths,
                stdout=proc.stdout, stderr=proc.stderr,
                return_code=proc.returncode,
                error_type=error_type, error_message=error_message,
                duration_seconds=duration,
                started_at=started_at, completed_at=completed_at,
            )

        except subprocess.TimeoutExpired as e:
            duration = time.time() - start_time
            # subprocess.run(text=True) makes these str if non-None, but the
            # caller may have raised TimeoutExpired manually with None or
            # bytes (tests do); coerce both cases to str.
            partial_stdout = e.stdout or ""
            partial_stderr = e.stderr or ""
            if isinstance(partial_stdout, bytes):
                partial_stdout = partial_stdout.decode("utf-8", "replace")
            if isinstance(partial_stderr, bytes):
                partial_stderr = partial_stderr.decode("utf-8", "replace")

            files_so_far = self._count_downloaded_files(partial_stdout)
            written_so_far = [str(p) for p in self._written_paths(partial_stdout)]
            stderr_lines = partial_stderr.strip().splitlines()
            tail_hint = stderr_lines[-1] if stderr_lines else "no stderr output"

            # If the partial output already shows a rate-limit pattern, the
            # timeout was almost certainly gallery-dl spinning on retries —
            # promote to RATE_LIMITED so _update_source_health stamps the
            # platform cooldown (same code path as a clean-exit rate limit).
            # Otherwise stay TIMEOUT and let the captured stdout/stderr +
            # files_so_far tell the operator whether it was "lots of
            # content" vs "stuck retrying" vs "hung silent".
            combined = (partial_stdout + "\n" + partial_stderr).lower()
            if any(p in combined for p in self.RATE_LIMIT_PATTERNS):
                error_type = ErrorType.RATE_LIMITED
                error_message = (
                    f"Rate-limited and never completed within "
                    f"{source_config.timeout}s ({files_so_far} files written)"
                )
            else:
                error_type = ErrorType.TIMEOUT
                error_message = (
                    f"Download timed out after {source_config.timeout}s — "
                    f"{files_so_far} file(s) written; last stderr: {tail_hint}"
                )

            log.error(
                "Download timeout for %s/%s after %.1fs (%d files written, "
                "last stderr: %s)",
                artist_slug, platform, duration, files_so_far, tail_hint,
            )

            return DownloadResult(
                success=False, url=url, artist_slug=artist_slug, platform=platform,
                files_downloaded=files_so_far,
                written_paths=written_so_far,
                stdout=partial_stdout, stderr=partial_stderr,
                return_code=-1,  # killed by timeout, no real exit code
                error_type=error_type, error_message=error_message,
                duration_seconds=duration,
                started_at=started_at,
                completed_at=datetime.now(UTC).isoformat(),
            )
        except Exception as exc:
            duration = time.time() - start_time
            log.exception("Unexpected error downloading %s/%s: %s", artist_slug, platform, exc)
            return DownloadResult(
                success=False, url=url, artist_slug=artist_slug, platform=platform,
                error_type=ErrorType.UNKNOWN_ERROR, error_message=str(exc),
                duration_seconds=duration,
                started_at=started_at,
                completed_at=datetime.now(UTC).isoformat(),
            )
        finally:
            try:
                Path(temp_config_path).unlink()  # noqa: ASYNC240
            except Exception:
                pass

    async def verify(
        self,
        url: str,
        artist_slug: str,
        platform: str,
        source_config: SourceConfig | None = None,
        cookies_path: str | None = None,
        auth_token: str | None = None,
        timeout: float = 45.0,  # noqa: ASYNC109 — subprocess.run timeout, not a coroutine deadline
    ) -> tuple[bool, str]:
        """Test that credentials authenticate against `url` WITHOUT
        downloading anything. Runs gallery-dl in --simulate mode limited
        to the first item; if auth is bad the extractor errors before it
        can list, which _categorize_error flags as AUTH_ERROR. Returns
        (ok, message). Used by the credential Verify button."""
        if source_config is None:
            source_config = SourceConfig()
        config = self._build_config_for_source(platform, source_config, artist_slug)
        if cookies_path:
            config["extractor"]["cookies"] = cookies_path
        if auth_token and platform == "discord":
            config["extractor"].setdefault("discord", {})["token"] = auth_token

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, dir=str(self._config_dir),
        ) as fh:
            json.dump(config, fh, indent=2)
            temp_config_path = fh.name
        try:
            cmd = [
                sys.executable, "-m", "gallery_dl",
                "--config", temp_config_path,
                "--simulate", "--range", "1-1", "--verbose", url,
            ]
            loop = asyncio.get_running_loop()
            proc = await loop.run_in_executor(
                None,
                lambda: subprocess.run(
                    cmd, capture_output=True, text=True, timeout=timeout,
                ),
            )
            etype, msg = self._categorize_error(proc.returncode, proc.stdout, proc.stderr)
            # TIER_LIMITED proves auth worked — gallery-dl reached the
            # post, was told it's tier-gated. The download path treats
            # this as success (line 712); verify must too, or operators
            # rotate working cookies for no reason. Audit 2026-06-02.
            if proc.returncode == 0 or etype in (
                ErrorType.NO_NEW_CONTENT, ErrorType.TIER_LIMITED,
            ):
                return True, "Credentials valid — the feed authenticated."
            if etype == ErrorType.AUTH_ERROR:
                return False, msg
            # Network / not-found / rate-limit / unknown: inconclusive,
            # not a definitive credential failure. Surface the reason.
            return False, f"Could not confirm ({etype.value}): {msg}"
        except subprocess.TimeoutExpired:
            return False, f"Verification timed out after {timeout:.0f}s"
        except Exception as exc:  # noqa: BLE001
            return False, f"Verification error: {exc}"
        finally:
            try:
                Path(temp_config_path).unlink()  # noqa: ASYNC240
            except Exception:
                pass