FabledCurator/backend/app/services/importer.py

"""Single-file import pipeline.

The Importer is the synchronous core of FC-2a's import path. A Celery task
(tasks/import_file.py) instantiates one per job and calls import_one().

The service is intentionally not async — Celery tasks are run in a
synchronous worker process, and the DB session is a sync session passed in
by the task. The Quart side uses an async session via the same models.
"""

import hashlib
import json
import logging
import shutil
from dataclasses import dataclass, field
from enum import StrEnum
from pathlib import Path

from PIL import Image
from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from ..models import (
    Artist,
    ExternalLink,
    ImageProvenance,
    ImageRecord,
    ImportSettings,
    Post,
    PostAttachment,
    Source,
)
from ..utils import safe_probe
from ..utils.paths import (
    canonical_subdir,
    derive_subdir,
    derive_top_level_artist,
    filehash_from_url,
    hash_suffixed_name,
    safe_ext,
)
from ..utils.phash import compute_phash, find_similar, fingerprint_path, fingerprints_match
from ..utils.sidecar import find_sidecar, parse_sidecar
from ..utils.slug import slugify
from .archive_extractor import extract_archive, is_archive
from .attachment_store import AttachmentStore
from .audits import single_color
from .link_extract import extract_external_links
from .thumbnailer import Thumbnailer
from .wip_title import (
    WIP_TITLE_SOURCE,
    apply_wip_image_tags,
    matches_wip_title,
    resolve_wip_tag_id,
)

log = logging.getLogger(__name__)

# Sentinel for the lazily-resolved wip tag id (distinguishes "not resolved yet"
# from a genuine None = tag absent, so absence is cached and not re-queried).
_UNSET = object()


class SkipReason(StrEnum):
    too_small = "too_small"
    too_transparent = "too_transparent"
    single_color = "single_color"
    duplicate_hash = "duplicate_hash"
    duplicate_phash = "duplicate_phash"
    invalid_image = "invalid_image"


@dataclass(frozen=True)
class ImportResult:
    # 'imported'   — new ImageRecord row created
    # 'superseded' — existing ImageRecord row got the new file (larger) + sidecar
    # 'attached'   — non-media saved as PostAttachment
    # 'refreshed'  — deep scan re-applied sidecar / filled NULL phash / NULL
    #                artist on an already-imported row (no new ImageRecord)
    # 'skipped'    — no work done (true duplicate, too small, etc.)
    # 'failed'     — pipeline error
    status: str
    image_id: int | None = None
    skip_reason: SkipReason | None = None
    error: str | None = None
    member_image_ids: list[int] = field(default_factory=list)


IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"}
ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS

# A "high-resolution files" pack often wraps per-chapter .rar/.zip inside one
# outer archive (incase #718) — recurse into nested archives so their images
# land instead of being silently dropped. Cap the depth so a maliciously
# deep-nested archive can't recurse forever; the bomb-probe re-runs per level.
_ARCHIVE_MAX_DEPTH = 3

# Tier-1 video near-dup (#871). pHash is images-only, so videos deduped on sha256
# alone — a different encode/remux of the same clip imported as a distinct record.
# Identity = container duration within a tight tolerance + matching aspect ratio,
# scoped to the same artist (codec/bitrate are NOT part of identity — the point is
# to match ACROSS re-encodes). Quality axis for supersede = pixel dimensions, same
# as image pHash. Tight tolerances because a wrong video merge is destructive.
_VIDEO_DUP_DURATION_TOL_SECONDS = 1.0
_VIDEO_DUP_ASPECT_TOL = 0.02


def is_supported(path: Path) -> bool:
    return path.suffix.lower() in ALL_EXTS


def is_video(path: Path) -> bool:
    return path.suffix.lower() in VIDEO_EXTS


def _safe_ext(path: Path) -> str:
    """Conservatively extract a file extension for PostAttachment.ext
    (varchar(32)). Thin wrapper over the shared `utils.paths.safe_ext` (kept for
    the Path-typed call sites + the [[path_suffix_sanitize]] memory pointer)."""
    return safe_ext(path)


def _mime_for(path: Path) -> str:
    suffix = path.suffix.lower()
    image_mimes = {
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".webp": "image/webp",
        ".bmp": "image/bmp",
        ".tif": "image/tiff",
        ".tiff": "image/tiff",
    }
    video_mimes = {
        ".mp4": "video/mp4",
        ".mov": "video/quicktime",
        ".avi": "video/x-msvideo",
        ".mkv": "video/x-matroska",
        ".webm": "video/webm",
        ".m4v": "video/x-m4v",
        ".wmv": "video/x-ms-wmv",
        ".flv": "video/x-flv",
    }
    return image_mimes.get(suffix) or video_mimes.get(suffix) or "application/octet-stream"


def _sha256_of(path: Path, chunk_size: int = 1 << 20) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fp:
        while True:
            block = fp.read(chunk_size)
            if not block:
                break
            h.update(block)
    return h.hexdigest()


class Importer:
    """Imports one file at a time. Constructed per-task by the Celery task."""

    def __init__(
        self,
        session: Session,
        images_root: Path,
        import_root: Path,
        thumbnailer: Thumbnailer,
        settings: ImportSettings,
        deep: bool = False,
    ):
        self.session = session
        self.images_root = images_root
        self.import_root = import_root
        self.thumbnailer = thumbnailer
        self.settings = settings
        self.deep = deep
        # Post-first ingest (#856, milestone #67): on the NATIVE core ingester the
        # post-record (`upsert_post_record`) is the SOLE writer of a post's
        # body/links/metadata; the per-media import only links image provenance +
        # localization, NOT the post body. download_service sets this True for
        # native platforms before the phase-3 media loop. Default False keeps the
        # gallery-dl path writing post fields via `_apply_sidecar` (unchanged).
        # Safe as a per-instance flag: Importer is constructed per Celery task.
        self.post_first = False
        self.attachments = AttachmentStore(images_root)
        # phash near-dup candidate cache. Archive imports call _import_media
        # per-member; without this cache the per-member SELECT *FROM
        # image_record WHERE phash IS NOT NULL fetch repeats N times and a
        # large library × many-member archive blew past soft_time_limit
        # (300s) — operator-flagged 2026-05-25. Loaded lazily on first
        # need, appended to on every imported/superseded outcome, never
        # invalidated mid-Importer (Importer instances are per-task /
        # per-archive-import so cross-instance staleness is harmless).
        self._phash_candidates: list[tuple] | None = None
        # Lazily-resolved `wip` system tag id for title-based WIP auto-tagging
        # (task #1458). Sentinel _UNSET so a genuine None (tag absent) is cached
        # and not re-queried per media. Importer is per-task, so this can't stale.
        self._wip_tag_id: int | None = _UNSET

    def _phash_candidates_cache(self) -> list[tuple]:
        """Cached `(phash, width, height, id)` rows from image_record.
        Loaded on first call, appended-to on subsequent imported/
        superseded outcomes. Soft-timeout pattern: an archive with N
        members + a library of M existing rows used to do N × M-row
        fetches (operator-flagged 2026-05-25); now it's exactly one.

        The per-task lifecycle of Importer (instantiated fresh by
        import_media_file) bounds the cache's staleness window: cross-
        process changes (other workers importing concurrently) won't
        be reflected, but that's the same race the un-cached version
        had — `find_similar` is best-effort anyway."""
        if self._phash_candidates is None:
            rows = self.session.execute(
                select(
                    ImageRecord.phash,
                    ImageRecord.width,
                    ImageRecord.height,
                    ImageRecord.id,
                ).where(ImageRecord.phash.is_not(None))
            ).all()
            self._phash_candidates = [
                (r.phash, r.width or 0, r.height or 0, r.id) for r in rows
            ]
        return self._phash_candidates

    def _phash_cache_append(self, phash, width, height, image_id) -> None:
        """Append a freshly-imported row to the cache so subsequent
        members of the same archive can match against it."""
        if self._phash_candidates is not None and phash is not None:
            self._phash_candidates.append(
                (phash, width or 0, height or 0, image_id)
            )

    def _pixel_confirmer(self, source: Path):
        """Build `find_similar`'s gate-3 callback for an incoming file.

        pHash proposes; this accepts. A candidate is a duplicate only if its
        file really is the same picture as `source` at a different size —
        which is the only merge the operator asked for (#4223). Everything
        else (a missing file, an unreadable one, a deleted row) returns
        False: the destructive outcomes here are dropping a download and
        overwriting a kept file, so an unanswerable question must not read
        as "yes".

        Both sides' fingerprints are computed lazily and cached, so an
        import that matches nothing costs no I/O at all and an archive
        member that keeps hitting the same candidate pays for it once.
        """
        new_fp: list = []
        cand_fps: dict[int, object] = {}

        def confirm(candidate_id: int) -> bool:
            if not new_fp:
                new_fp.append(fingerprint_path(source))
            if new_fp[0] is None:
                return False
            if candidate_id not in cand_fps:
                rec = self.session.get(ImageRecord, candidate_id)
                cand_fps[candidate_id] = (
                    fingerprint_path(Path(rec.path)) if rec and rec.path else None
                )
            return fingerprints_match(new_fp[0], cand_fps[candidate_id])

        return confirm

    def _get_or_create(self, stmt, factory):
        """Race-safe find-or-create. Run `stmt` (scalar_one_or_none); if a
        row exists, return it. Otherwise open a savepoint and INSERT
        ``factory()``; on IntegrityError (a concurrent worker inserted the
        same row first) roll the savepoint back — NOT the outer transaction,
        which would lose the surrounding scan's progress — and re-run `stmt`
        (scalar_one) to return the row the other worker created.

        Centralizes the pattern shared by _find_or_create_source and
        _find_or_create_post. The plain SELECT-then-INSERT version lost
        races under the 5-min recovery sweep (operator-flagged
        2026-05-26)."""
        existing = self.session.execute(stmt).scalar_one_or_none()
        if existing is not None:
            return existing
        sp = self.session.begin_nested()
        try:
            row = factory()
            self.session.add(row)
            self.session.flush()
            sp.commit()
            return row
        except IntegrityError:
            sp.rollback()
            return self.session.execute(stmt).scalar_one()

    def _find_or_create_source(
        self, *, artist_id: int, platform: str, url: str,
    ) -> Source:
        """Race-safe find-or-create on `source` keyed by
        (artist_id, platform, url) — the same key as the
        `uq_source_artist_platform_url` constraint.

        Two concurrent workers processing different files in the same
        post can both find no existing Source row then both INSERT,
        which trips the unique constraint and poisons the session with
        `psycopg.errors.UniqueViolation`. Operator-flagged 2026-05-26.

        Pattern: select; if absent, open a savepoint and INSERT.
        On IntegrityError, roll the savepoint back (NOT the outer
        transaction, which would lose the surrounding scan's progress)
        and re-select — the concurrent op just created the row we
        wanted, so the second select will find it.
        """
        stmt = select(Source).where(
            Source.artist_id == artist_id,
            Source.platform == platform,
            Source.url == url,
        )
        return self._get_or_create(
            stmt,
            lambda: Source(artist_id=artist_id, platform=platform, url=url),
        )

    def _lookup_source_for_sidecar(
        self, *, artist_id: int, platform: str,
    ) -> Source | None:
        """Find the real subscription Source for (artist, platform), or
        None if no subscription exists.

        Pre-alembic-0030 this method would CREATE a synthetic
        `sidecar:<platform>:<slug>` Source when no real one existed —
        because `Post.source_id` was NOT NULL and the importer needed
        something to attach Posts to. Alembic 0030 relaxed both
        `Post.source_id` and `ImageProvenance.source_id` to nullable, so
        synthetic anchors are obsolete; the importer now leaves
        source_id as None when no subscription exists for the (artist,
        platform). Operator-asked 2026-06-01: synthetic Sources had
        leaked into the Subscriptions UI as phantom subscriptions and
        the operator wanted the data model to truthfully say "this
        content has no live subscription."
        """
        stmt = (
            select(Source)
            .where(
                Source.artist_id == artist_id,
                Source.platform == platform,
            )
            .order_by(Source.id.asc())
            .limit(1)
        )
        return self.session.execute(stmt).scalar_one_or_none()

    def _find_or_create_post(
        self, *, source_id: int | None, external_post_id: str,
        artist_id: int,
    ) -> Post:
        """Race-safe find-or-create on `post`. Keyed by
        (source_id, external_post_id) when source_id is set — the
        `uq_post_source_external_id` constraint guards. For NULL-source
        posts the existence check matches on (artist_id, external_post_id),
        which the partial unique index `uq_post_artist_external_id_null_source`
        (alembic 0030) guards. Same savepoint + IntegrityError-recovery
        pattern as the rest of the helpers."""
        if source_id is not None:
            stmt = select(Post).where(
                Post.source_id == source_id,
                Post.external_post_id == external_post_id,
            )
        else:
            stmt = select(Post).where(
                Post.source_id.is_(None),
                Post.artist_id == artist_id,
                Post.external_post_id == external_post_id,
            )
        return self._get_or_create(
            stmt,
            lambda: Post(
                source_id=source_id,
                artist_id=artist_id,
                external_post_id=external_post_id,
            ),
        )

    def _sync_external_links(self, post: Post) -> None:
        """Record off-platform file-host links (mega/gdrive/mediafire/dropbox/
        pixeldrain) found in the post body, so they're never silently dropped
        and the download worker can fetch them later. Shared by every platform's
        import (runs off Post.description). INSERT-MISSING only — an existing row
        keeps its status/attempts (a re-import must not reset a link already
        downloaded or dead-lettered). Identity is (post_id, url); the full url
        incl. #fragment is preserved by the extractor."""
        links = extract_external_links(post.description)
        if not links:
            return
        self.session.flush()  # ensure post.id is assigned before we reference it
        existing = set(self.session.execute(
            select(ExternalLink.url).where(ExternalLink.post_id == post.id)
        ).scalars().all())
        for link in links:
            if link.url in existing:
                continue
            self.session.add(ExternalLink(
                post_id=post.id,
                artist_id=post.artist_id,
                host=link.host,
                url=link.url,
                label=link.label,
            ))

    def import_one(self, source: Path) -> ImportResult:
        """Dispatch by kind. Media → normal pipeline. Archive → extract
        media members (one Post via the archive-adjacent sidecar) and
        preserve the archive itself. Other non-media → PostAttachment.
        Nothing is skipped for being non-media (FC-2d-iii)."""
        if source.suffix.lower() == ".json":
            return ImportResult(
                status="skipped", skip_reason=SkipReason.invalid_image,
                error="sidecar json is metadata, not content",
            )
        if is_archive(source):
            return self._import_archive(source)
        if not is_supported(source):
            return self._capture_attachment(source)
        return self._import_media(source, source)

    def _resolve_artist(self, attribution_path: Path) -> Artist | None:
        name = derive_top_level_artist(attribution_path, self.import_root)
        return self._upsert_artist(name) if name else None

    def _post_for_sidecar(
        self, source: Path, artist: Artist | None
    ) -> Post | None:
        """If a sidecar sits next to `source`, ensure its Source+Post
        exist (idempotent) and return the Post — so attachments can link
        to the same Post the per-member _apply_sidecar will reuse."""
        sc = find_sidecar(source)
        if sc is None or artist is None:
            return None
        try:
            data = json.loads(sc.read_text("utf-8"))
            if not isinstance(data, dict):
                raise ValueError("sidecar JSON is not an object")
        except Exception as exc:
            log.warning("sidecar parse failed for %s: %s", sc, exc)
            return None
        sd = parse_sidecar(data)
        platform = sd.platform or "unknown"
        src = self._lookup_source_for_sidecar(
            artist_id=artist.id, platform=platform,
        )
        epid = sd.external_post_id or sc.stem
        return self._find_or_create_post(
            source_id=src.id if src else None,
            external_post_id=epid,
            artist_id=artist.id,
        )

    def _capture_attachment(
        self, source: Path, *, post: Post | None = None,
        artist: Artist | None = None, resolved: bool = False,
    ) -> ImportResult:
        if not resolved:
            artist = self._resolve_artist(source)
            post = self._post_for_sidecar(source, artist)
        sha = _sha256_of(source)
        post_id = post.id if post else None
        # Dedup is PER-POST, not global: a non-art file the creator attaches to
        # many posts (a standard pdf/zip/link-card) must get a row on EVERY post
        # so none is left a bare shell. The on-disk blob stays sha-deduped
        # (attachments.store is sha-addressed + idempotent); only the rows are
        # per-post. PostAttachment's unique is partial (post_id, sha256) for
        # real posts and (sha256) for the NULL-post filesystem case. Before
        # 2026-06-08 the global UNIQUE(sha256) left every post after the first
        # with no attachment row → 1589 empty Anduo shells.
        if post_id is not None:
            select_existing = select(PostAttachment).where(
                PostAttachment.post_id == post_id,
                PostAttachment.sha256 == sha,
            )
        else:
            select_existing = select(PostAttachment).where(
                PostAttachment.post_id.is_(None),
                PostAttachment.sha256 == sha,
            )
        existing = self.session.execute(select_existing).scalar_one_or_none()
        if existing is not None:
            self.session.commit()
            return ImportResult(status="attached")
        # Savepoint + IntegrityError recovery — the partial UNIQUE means two
        # workers can both pass the SELECT and only the second INSERT fails.
        # Without savepoint, the outer transaction poisons and the calling task
        # crashes. attachments.store is sha-addressed so both workers race to
        # write the same target path; shutil.copy2 + rename is idempotent.
        sp = self.session.begin_nested()
        try:
            stored = self.attachments.store(source, sha)
            self.session.add(PostAttachment(
                post_id=post_id,
                artist_id=artist.id if artist else None,
                sha256=sha,
                path=stored,
                original_filename=source.name,
                ext=_safe_ext(source),
                mime=_mime_for(source),
                size_bytes=source.stat().st_size,
            ))
            self.session.flush()
            sp.commit()
        except IntegrityError:
            sp.rollback()
            # Lost the race — the other worker's row for this (post, sha) wins.
            self.session.execute(select_existing).scalar_one()
        self.session.commit()
        return ImportResult(status="attached")

    def _import_archive(
        self, source: Path, *,
        artist: Artist | None = None,
        source_row: Source | None = None,
    ) -> ImportResult:
        # Layer-3 isolation: bomb-size guard + integrity test in a
        # spawned child BEFORE extracting in this process. A
        # decompression bomb or a native-lib crash on a malformed
        # archive is contained to the child; we reject the file cleanly
        # instead of OOMing/segfaulting the import worker. extract_archive
        # is already fail-soft for plain exceptions, so this only adds
        # the hard-crash protection.
        #
        # Audit 2026-06-02: optional artist/source_row kwargs let the
        # download path thread its explicit subscription context
        # through instead of having _resolve_artist re-derive from
        # path-walk (which works by coincidence today because gallery-dl
        # lays files out under /images/<artist_slug>/...). Filesystem
        # import still calls bare _import_archive(source) and falls
        # back to the path-walk derivation as before.
        probe = safe_probe.probe_archive(source)
        if not probe.ok:
            if probe.crashed:
                return ImportResult(
                    status="failed",
                    error=f"archive probe crashed/timed out: {probe.reason}",
                )
            # Clean rejection (bomb cap exceeded, integrity mismatch):
            # still preserve the archive file itself as an attachment so
            # nothing silently vanishes, matching extract_archive's
            # fail-soft contract.
            artist_use = artist if artist is not None else self._resolve_artist(source)
            post = self._post_for_sidecar(source, artist_use)
            self._capture_attachment(
                source, post=post, artist=artist_use, resolved=True,
            )
            reason = f"archive probe rejected, captured unextracted: {probe.reason}"
            log.warning("%s: %s", source.name, reason)
            return ImportResult(status="attached", error=reason)

        artist_use = artist if artist is not None else self._resolve_artist(source)
        post = self._post_for_sidecar(source, artist_use)
        member_ids: list[int] = []
        # Every member image touched (new + superseded + deduped), so the
        # from_attachment_id stamp below covers files that already existed in the
        # library and were merely re-linked to this post — those matter most
        # (the HR copy a bundle re-ships). Separate from member_ids, which is
        # the NEWLY-imported subset feeding the ImportResult contract.
        member_record_ids: set[int] = set()
        # Per-outcome tally so the "no images" reason names the ACTUAL cause
        # (#718): nested-archive packs, all-deduped (benign), unsupported formats,
        # or failed/corrupt members — instead of one catch-all string.
        counts = {
            "media": 0, "deduped": 0, "unsupported": 0,
            "failed": 0, "nested": 0, "nested_rejected": 0,
        }
        self._collect_archive_members(
            source, attribution=source, source_row=source_row,
            depth=0, member_ids=member_ids, counts=counts,
            member_record_ids=member_record_ids,
        )
        # Preserve the archive itself (links to the same Post/Artist).
        self._capture_attachment(
            source, post=post, artist=artist_use, resolved=True
        )
        # Stamp each member's provenance row for THIS post with the archive it
        # came out of (milestone #87). Done as a post-pass rather than threaded
        # through _import_media/_apply_sidecar so the many dedup/supersede
        # branches stay untouched. NULL-only so a re-extract never re-stamps and
        # the backfill (reextract task → this same path) is idempotent. Nested
        # members link to this OUTER archive — the only one stored as a blob.
        self._stamp_member_archive(
            post.id if post is not None else None, source, member_record_ids,
        )
        if member_ids:
            return ImportResult(
                status="imported", image_id=member_ids[0],
                member_image_ids=member_ids,
            )
        # No NEW images landed. Name WHY precisely so a post showing "no images"
        # beside an archive is diagnosable — and DON'T flag the benign all-deduped
        # case as a problem: those images already exist in the library and were
        # re-linked to this post (enrich-on-duplicate), so the post DOES show them.
        if counts["media"] == 0 and counts["nested"] == 0:
            reason = ("archive yielded no members (unsupported/corrupt, or the "
                      "extractor backend failed)")
        elif counts["deduped"] and not (
            counts["unsupported"] or counts["failed"] or counts["nested_rejected"]
        ):
            log.info(
                "%s: all %d image member(s) already in library (deduped + linked "
                "to this post) — archive preserved, no new import",
                source.name, counts["deduped"],
            )
            return ImportResult(status="attached")
        else:
            reason = (
                f"no new images — media={counts['media']} deduped={counts['deduped']} "
                f"unsupported/non-media={counts['unsupported']} failed={counts['failed']} "
                f"nested={counts['nested']} nested-rejected={counts['nested_rejected']}"
            )
        log.warning("%s: %s", source.name, reason)
        return ImportResult(status="attached", error=reason)

    def _collect_archive_members(
        self, archive_path: Path, *, attribution: Path,
        source_row: Source | None, depth: int,
        member_ids: list[int], counts: dict,
        member_record_ids: set[int],
    ) -> None:
        """Extract `archive_path` and import its image/video members, RECURSING
        into nested archives (#718). Members attribute to `attribution` — the
        OUTER archive's path, so its sidecar resolves them to the right Post even
        when they came from a nested archive. Depth-capped + bomb-probed per
        nested level. Mutates `member_ids` and `counts` in place.

        extract_archive is fail-soft, but each level is wrapped so one bad nested
        archive can't abort the whole import."""
        try:
            with extract_archive(archive_path) as members:
                for _name, member_path in members:
                    if not member_path.is_file():
                        continue  # directory entries etc. — not media
                    if is_archive(member_path):
                        counts["nested"] += 1
                        if depth + 1 > _ARCHIVE_MAX_DEPTH:
                            counts["nested_rejected"] += 1
                            log.warning(
                                "nested archive past depth cap %d, skipped: %s",
                                _ARCHIVE_MAX_DEPTH, member_path.name,
                            )
                            continue
                        probe = safe_probe.probe_archive(member_path)
                        if not probe.ok:
                            counts["nested_rejected"] += 1
                            log.warning(
                                "nested archive rejected (%s): %s",
                                probe.reason, member_path.name,
                            )
                            continue
                        self._collect_archive_members(
                            member_path, attribution=attribution,
                            source_row=source_row, depth=depth + 1,
                            member_ids=member_ids, counts=counts,
                            member_record_ids=member_record_ids,
                        )
                        continue
                    counts["media"] += 1
                    if not is_supported(member_path):
                        counts["unsupported"] += 1
                        continue  # non-media preserved via the stored archive
                    res = self._import_media(
                        member_path, attribution, explicit_source=source_row,
                    )
                    if res.status in ("imported", "superseded") and res.image_id:
                        member_ids.append(res.image_id)
                        member_record_ids.add(res.image_id)
                    elif res.status == "skipped" and res.skip_reason in (
                        SkipReason.duplicate_hash, SkipReason.duplicate_phash
                    ):
                        counts["deduped"] += 1
                        # A deduped member still links provenance to this post
                        # (enrich-on-duplicate); record it so its archive origin
                        # gets stamped too.
                        if res.image_id:
                            member_record_ids.add(res.image_id)
                    else:
                        counts["failed"] += 1
        except Exception as exc:  # noqa: BLE001 — defensive per level; keep going
            log.warning(
                "archive extraction failed for %s (depth %d): %s",
                archive_path.name, depth, exc,
            )

    def _stamp_member_archive(
        self, post_id: int | None, archive_source: Path, member_record_ids: set[int],
    ) -> None:
        """Record which archive each extracted member came from (milestone #87).

        Resolves the archive's own PostAttachment (by post + sha — it was just
        captured) and stamps from_attachment_id on every member's provenance row
        FOR THIS POST. NULL-only, so re-extracting the same archive (the backfill
        path) never overwrites and stays idempotent. No-op when the archive isn't
        post-attached (filesystem import with no post) or yielded no members.
        """
        if post_id is None or not member_record_ids:
            return
        sha = _sha256_of(archive_source)
        att_id = self.session.execute(
            select(PostAttachment.id).where(
                PostAttachment.post_id == post_id,
                PostAttachment.sha256 == sha,
            )
        ).scalar_one_or_none()
        if att_id is None:
            return
        self.session.execute(
            update(ImageProvenance)
            .where(
                ImageProvenance.image_record_id.in_(member_record_ids),
                ImageProvenance.post_id == post_id,
                ImageProvenance.from_attachment_id.is_(None),
            )
            .values(from_attachment_id=att_id)
        )
        self.session.commit()

    @staticmethod
    def _video_aspect_matches(w, h, cw, ch) -> bool:
        """True when two (w,h) pairs share an aspect ratio within tolerance.
        Missing dims → don't block (duration is the primary signal)."""
        if not (w and h and cw and ch):
            return True
        return abs((w / h) - (cw / ch)) <= _VIDEO_DUP_ASPECT_TOL

    def _find_similar_video(
        self, duration: float | None, width: int | None, height: int | None,
        artist_id: int | None,
    ) -> tuple[str, int | None]:
        """Tier-1 video near-dup (#871). Find a same-artist video whose duration
        matches within tolerance AND whose aspect ratio matches — treat it as the
        same content (a different encode/remux). Returns find_similar's contract:
        ("none"|"larger_exists"|"smaller_exists", id), quality judged by pixel
        dimensions so a higher-res copy supersedes a smaller one."""
        if duration is None or artist_id is None:
            return ("none", None)
        lo = duration - _VIDEO_DUP_DURATION_TOL_SECONDS
        hi = duration + _VIDEO_DUP_DURATION_TOL_SECONDS
        rows = self.session.execute(
            select(ImageRecord.id, ImageRecord.width, ImageRecord.height)
            .where(
                ImageRecord.artist_id == artist_id,
                ImageRecord.mime.like("video/%"),
                ImageRecord.duration_seconds.is_not(None),
                ImageRecord.duration_seconds >= lo,
                ImageRecord.duration_seconds <= hi,
            )
        ).all()
        w, h = width or 0, height or 0
        for cid, cw, ch in rows:
            if not self._video_aspect_matches(width, height, cw, ch):
                continue
            cw0, ch0 = cw or 0, ch or 0
            if cw0 >= w and ch0 >= h:
                return ("larger_exists", cid)
            if w > cw0 or h > ch0:
                return ("smaller_exists", cid)
        return ("none", None)

    def _import_media(
        self, source: Path, attribution_path: Path,
        *, explicit_source: Source | None = None,
    ) -> ImportResult:
        """The media import pipeline (filters, dedup, copy, provenance).

        `attribution_path` anchors artist/subdir/sidecar derivation: for
        a normal file it equals `source`; for an archive member it is the
        archive's real path (the member lives in a tempdir).
        """
        if not is_supported(source):
            return ImportResult(
                status="skipped", skip_reason=SkipReason.invalid_image,
                error=f"unsupported extension {source.suffix}",
            )

        # Compute file dimensions (images only) and apply filters.
        width = height = None
        duration = None  # video container duration (Tier-1 near-dup key, #871)
        has_alpha = False
        if is_video(source):
            # Layer-3 isolation: validate the container via ffprobe (a
            # separate process) before the rest of the pipeline touches
            # it. A corrupt video that would crash a decoder is rejected
            # cleanly here, and we capture width/height + duration for free.
            probe = safe_probe.probe_video(source)
            if not probe.ok:
                if probe.crashed:
                    return ImportResult(
                        status="failed",
                        error=f"video probe crashed/timed out: {probe.reason}",
                    )
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.invalid_image,
                    error=probe.reason,
                )
            width, height, duration = probe.width, probe.height, probe.duration
        else:
            try:
                with Image.open(source) as im:
                    im.verify()
                with Image.open(source) as im:
                    width, height = im.size
                    has_alpha = im.mode in ("RGBA", "LA") or (
                        im.mode == "P" and "transparency" in im.info
                    )
            except Exception as exc:
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.invalid_image, error=str(exc),
                )

            if (self.settings.min_width and width < self.settings.min_width) or (
                self.settings.min_height and height < self.settings.min_height
            ):
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.too_small,
                    error=f"{width}x{height} below min {self.settings.min_width}x{self.settings.min_height}",
                )

            if self.settings.skip_transparent and has_alpha:
                try:
                    pct = self._transparency_pct(source)
                except OSError as exc:
                    # PIL.verify() at line 263 only validates header structure;
                    # truncated/corrupt pixel data only surfaces when load()
                    # actually decodes (here via getchannel('A')). Convert to
                    # invalid_image skip so the Celery autoretry loop doesn't
                    # bounce the same broken file forever.
                    return ImportResult(
                        status="skipped", skip_reason=SkipReason.invalid_image,
                        error=f"PIL load failed during transparency check: {exc}",
                    )
                if pct >= self.settings.transparency_threshold:
                    return ImportResult(
                        status="skipped", skip_reason=SkipReason.too_transparent,
                        error=f"{pct:.2%} transparent",
                    )

            if self.settings.skip_single_color and self._single_color_hit(source):
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.single_color,
                    error=(f"one color dominates >"
                           f"{self.settings.single_color_threshold:.0%}"),
                )

        # Artist anchored to the attribution path (folder→artist), resolved
        # UP-FRONT so the enrich-on-duplicate branches link provenance with the
        # right artist even when the sidecar carries none — which is now the norm
        # for archive members under post-first (the per-media sidecar is minimal).
        # Without this a cross-posted / re-packed archive image deduped and was
        # left UNLINKED from the new post = a post showing "no images" (#718).
        _artist_name = derive_top_level_artist(attribution_path, self.import_root)
        path_artist = self._upsert_artist(_artist_name) if _artist_name else None

        # Hash dedup (exact).
        sha = _sha256_of(source)
        existing_stmt = select(ImageRecord).where(ImageRecord.sha256 == sha)
        existing = self.session.execute(existing_stmt).scalar_one_or_none()
        if existing:
            if not self.deep:
                # Enrich-on-duplicate (parity with attach_in_place): a re-scanned
                # file that is a byte-dup of an existing image still links its
                # post via _apply_sidecar, so a cross-posted image shows on every
                # post.
                self._apply_sidecar(
                    existing, attribution_path, path_artist,
                    explicit_source=explicit_source,
                )
                self.session.commit()
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.duplicate_hash,
                    image_id=existing.id, error="sha256 already present",
                )
            return self._deep_rederive(existing, source, attribution_path)

        # Perceptual near-dup (images only; videos keep phash NULL).
        phash = None
        if not is_video(source):
            try:
                with Image.open(source) as im:
                    phash = compute_phash(im)
            except OSError as exc:
                # Same rationale as the transparency-check guard above:
                # broken-pixel-data files pass verify() but blow up here.
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.invalid_image,
                    error=f"PIL load failed during phash compute: {exc}",
                )
            if phash is not None:
                candidates = self._phash_candidates_cache()
                rel, match_id = find_similar(
                    phash, width or 0, height or 0,
                    candidates, self.settings.phash_threshold,
                    confirm=self._pixel_confirmer(source),
                )
                if rel == "larger_exists":
                    # Enrich-on-duplicate (parity with attach_in_place).
                    larger = self.session.get(ImageRecord, match_id)
                    if larger is not None:
                        self._apply_sidecar(
                            larger, attribution_path, path_artist,
                            explicit_source=explicit_source,
                        )
                        self.session.commit()
                    return ImportResult(
                        status="skipped",
                        skip_reason=SkipReason.duplicate_phash,
                        image_id=match_id,
                        error="perceptual near-duplicate of larger existing image",
                    )
                if rel == "smaller_exists":
                    target = self.session.get(ImageRecord, match_id)
                    self._supersede(target, source, sha, phash, width, height)
                    return ImportResult(
                        status="superseded", image_id=match_id
                    )
        elif duration is not None:
            # Tier-1 video near-dup (#871): same-artist, matching duration+aspect
            # → same content across re-encodes. Mirror the image flow.
            rel, match_id = self._find_similar_video(
                duration, width, height,
                path_artist.id if path_artist else None,
            )
            if rel == "larger_exists":
                larger = self.session.get(ImageRecord, match_id)
                if larger is not None:
                    self._apply_sidecar(
                        larger, attribution_path, path_artist,
                        explicit_source=explicit_source,
                    )
                    self.session.commit()
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.duplicate_phash,
                    image_id=match_id,
                    error="video near-duplicate (duration+aspect) of existing equal/larger video",
                )
            if rel == "smaller_exists":
                target = self.session.get(ImageRecord, match_id)
                self._supersede(
                    target, source, sha, None, width, height, duration=duration,
                )
                return ImportResult(status="superseded", image_id=match_id)

        dest = self._copy_to_library(source, sha, attribution_path, path_artist)

        record = ImageRecord(
            path=str(dest),
            sha256=sha,
            phash=phash,
            size_bytes=dest.stat().st_size,
            mime=_mime_for(source),
            width=width,
            height=height,
            duration_seconds=duration,
            origin="imported_filesystem",
            integrity_status="unknown",
        )
        self.session.add(record)
        self.session.flush()
        self._phash_cache_append(phash, width, height, record.id)

        # Folder→artist (path_artist resolved up-front above); bind it to the
        # new record so it carries the canonical Artist.
        if path_artist is not None and record.artist_id is None:
            record.artist_id = path_artist.id
            self.session.flush()

        # Sidecar provenance (best-effort; never fails the import).
        # explicit_source lets the FC-3c download path bind the new
        # ImageProvenance row to its subscription Source instead of
        # having _apply_sidecar re-derive via _lookup_source_for_sidecar.
        # Audit 2026-06-02 — archive members extracted from a
        # subscription-downloaded zip previously lost subscription
        # linkage if the on-disk layout didn't match assumptions.
        self._apply_sidecar(
            record, attribution_path, path_artist, explicit_source=explicit_source,
        )

        # Thumbnail is queued separately by the calling task; the importer
        # does not generate thumbnails inline so the import queue stays moving.

        # Title-based WIP auto-tag (task #1458): fresh import only, after the
        # sidecar has linked the post so record.primary_post_id / its title exist.
        self._maybe_apply_wip_title(record)

        self.session.commit()
        return ImportResult(status="imported", image_id=record.id)

    def _deep_rederive(
        self, existing: ImageRecord, source: Path, attribution_path: Path
    ) -> ImportResult:
        """Deep scan: backfill phash/provenance/artist on an
        already-imported record. METADATA ONLY — never re-runs the pHash
        near-dup / supersede path. NULL-only on phash/artist, additive on
        sidecar Post/Source/ImageProvenance (via _apply_sidecar).
        Idempotent: a second deep-scan over the same file finds nothing
        to refresh and is a no-op.

        Returns status="refreshed" so the UI can surface the work done
        instead of the prior misleading "skipped/duplicate_hash" reading.
        Operator-flagged 2026-05-25 — IR has had this; FC inherited it
        as a no-op skip during the original port and the UI showed deep
        scan as "completed with no changes" even when sidecar metadata
        actually got re-applied to N existing rows.
        """
        if existing.phash is None and not is_video(source):
            try:
                with Image.open(source) as im:
                    ph = compute_phash(im)
                if ph is not None:
                    existing.phash = ph
                    # Promoted from NULL to non-NULL → cache is now stale
                    # (this row would newly qualify for the candidates set).
                    self._phash_candidates = None
            except Exception as exc:
                log.warning("deep rephash failed for %s: %s", source, exc)

        artist = None
        name = derive_top_level_artist(attribution_path, self.import_root)
        if name:
            artist = self._upsert_artist(name)
            if existing.artist_id is None:
                existing.artist_id = artist.id

        self._apply_sidecar(existing, attribution_path, artist)
        self.session.commit()
        return ImportResult(status="refreshed", image_id=existing.id)

    def _maybe_apply_wip_title(self, record: ImageRecord) -> None:
        """Auto-apply the `wip` system tag to a FRESHLY-imported image when its
        primary post's TITLE explicitly declares work-in-progress (task #1458 —
        the artist's own "WIP" / "work in progress" label).

        Called ONLY from the two new-record paths (never deep-scan / supersede),
        so a manually-removed WIP tag is never re-applied by a routine re-scan —
        removal sticks. The existing catalogue is covered separately by the
        operator-triggered backfill sweep. Gated by the settings toggle, and
        best-effort: any failure is logged, never allowed to fail the import."""
        if not self.settings.wip_title_tagging_enabled:
            return
        if record.primary_post_id is None:
            return
        try:
            title = self.session.execute(
                select(Post.post_title).where(Post.id == record.primary_post_id)
            ).scalar_one_or_none()
            if not matches_wip_title(title):
                return
            if self._wip_tag_id is _UNSET:
                self._wip_tag_id = resolve_wip_tag_id(self.session)
            if self._wip_tag_id is None:
                return
            apply_wip_image_tags(
                self.session, [record.id], self._wip_tag_id, source=WIP_TITLE_SOURCE
            )
        except Exception as exc:  # noqa: BLE001 — a tag must never fail an import
            log.warning(
                "wip-title auto-tag failed for image %s: %s", record.id, exc
            )

    def _apply_post_fields(self, post: Post, sd) -> None:
        """Write a parsed sidecar's post-level fields onto a Post — the SINGLE
        predicate shared by BOTH ingest paths: the per-media path (_apply_sidecar)
        and the post-record path (upsert_post_record). Fill-with-non-empty:
        parse_sidecar yields None for empty values, so a None field is left
        untouched (an empty feed body never wipes a populated one). raw_metadata +
        external-link sync always run (latest snapshot). Keeping ONE copy stops
        the two paths from diverging on how a post body/links get stored — they
        were verbatim duplicates (#842 DRY pass; see [[feedback_preview_apply_parity]]).
        """
        if sd.post_url is not None:
            post.post_url = sd.post_url
        if sd.post_title is not None:
            post.post_title = sd.post_title
        if sd.post_date is not None:
            post.post_date = sd.post_date
        if sd.description is not None:
            post.description = sd.description
        if sd.attachment_count is not None:
            post.attachment_count = sd.attachment_count
        post.raw_metadata = sd.raw
        self._sync_external_links(post)

    def upsert_post_record(
        self, sidecar: Path, *, artist: Artist | None = None,
        source: Source | None = None,
    ) -> bool:
        """Upsert the Post for a post-ONLY sidecar (a media-less post), so the
        artist archive includes text posts (their body + external links).

        Reuses `_find_or_create_post` (keyed on external_post_id) so it UPDATES
        the SAME Post a media import would create — never doubles. Fields are
        FILLED, never clobbered with empty: parse_sidecar yields None for empty
        values, and a None field is left untouched (an empty feed body never
        wipes a populated one). Returns True if a Post was upserted, False if the
        sidecar was unusable (parse failure / no artist)."""
        try:
            data = json.loads(sidecar.read_text("utf-8"))
            if not isinstance(data, dict):
                raise ValueError("sidecar JSON is not an object")
        except Exception as exc:
            log.warning("post-record sidecar parse failed for %s: %s", sidecar, exc)
            return False

        sd = parse_sidecar(data)

        if artist is None:
            name = self._sidecar_artist_name(data)
            artist = self._upsert_artist(name) if name else None
        if artist is None:
            log.warning("post-record sidecar %s has no artist; skipping", sidecar)
            return False

        if source is not None:
            src = source
        else:
            platform = sd.platform or "unknown"
            src = self._lookup_source_for_sidecar(
                artist_id=artist.id, platform=platform,
            )

        epid = sd.external_post_id or sidecar.stem
        post = self._find_or_create_post(
            source_id=src.id if src else None,
            external_post_id=epid,
            artist_id=artist.id,
        )
        if post.artist_id is None:
            post.artist_id = artist.id
        self._apply_post_fields(post, sd)
        self._redate_post_images(post)
        self.session.commit()
        return True

    def _redate_post_images(self, post: Post) -> None:
        """Carry a post's date onto the images already linked to it (#4431).

        The native ingesters import a post's media BEFORE its record: the
        per-media sidecar holds only the image identity (post-first, #856), and
        the date arrives with `_post.json`. So `_attach_provenance` links each
        image to a post that has no date yet, and the image keeps its download
        time. This runs when the record lands, and applies the same two rules
        `_attach_provenance` applies: `effective_date` is the PRIMARY post's
        date, and `earliest_post_date` is the earliest date across every post
        the image is linked to. Only rows that differ are written."""
        if post.post_date is None:
            return
        self.session.flush()
        self.session.execute(
            update(ImageRecord)
            .where(ImageRecord.primary_post_id == post.id)
            .where(ImageRecord.effective_date.is_distinct_from(post.post_date))
            .values(effective_date=post.post_date)
            .execution_options(synchronize_session=False)
        )
        linked = select(ImageProvenance.image_record_id).where(
            ImageProvenance.post_id == post.id
        )
        earliest = (
            select(func.min(Post.post_date))
            .select_from(ImageProvenance)
            .join(Post, Post.id == ImageProvenance.post_id)
            .where(ImageProvenance.image_record_id == ImageRecord.id)
            .where(Post.post_date.is_not(None))
            .correlate(ImageRecord)
            .scalar_subquery()
        )
        self.session.execute(
            update(ImageRecord)
            .where(ImageRecord.id.in_(linked))
            .where(ImageRecord.earliest_post_date.is_distinct_from(earliest))
            .values(earliest_post_date=earliest)
            .execution_options(synchronize_session=False)
        )

    def attach_in_place(
        self,
        path: Path,
        *,
        sidecar_path: Path | None = None,
        artist: Artist | None = None,
        source: Source | None = None,
    ) -> ImportResult:
        """Attach a file ALREADY at its final library location (FC-3c).

        Mirrors _import_media but skips the copy step — the file is
        where it'll permanently live. Caller (DownloadService) already
        has artist/source context from the subscription Source row; pass
        them through. The sidecar JSON gallery-dl emits next to each
        downloaded file is read by `_apply_sidecar` via `find_sidecar`.

        File-type dispatch parity with `import_one` (FC-2d-iii): zips,
        PDFs, audio etc. become PostAttachments; archives are extracted.
        Without this dispatch, gallery-dl-downloaded non-media bounced
        back as `skipped+invalid_image`, which DownloadService counted
        as an ingest error and flipped otherwise-successful runs to
        status="error". Operator-flagged 2026-06-02 after a Lustria
        patreon run with a 94MB OST zip went red despite 21 successful
        image attaches.

        Caller's responsibilities after this returns:
          - duplicate_hash / duplicate_phash skip → delete the on-disk file
          - superseded → file stays where it is (now canonical)
          - imported → file stays where it is
          - attached → the file's been copied into the attachments store;
            caller may delete the on-disk original (mirrors duplicate_hash)
          - failed → file untouched; caller decides
        """
        if path.suffix.lower() == ".json":
            return ImportResult(
                status="skipped", skip_reason=SkipReason.invalid_image,
                error="sidecar json is metadata, not content",
            )
        if is_archive(path):
            return self._import_archive(
                path, artist=artist, source_row=source,
            )
        if not is_supported(path):
            post = self._post_for_sidecar(path, artist) if artist else None
            return self._capture_attachment(
                path, post=post, artist=artist, resolved=True,
            )

        # Format / dimension / transparency filters (mirror _import_media).
        width = height = None
        duration = None  # video container duration (Tier-1 near-dup key, #871)
        has_alpha = False
        if not is_video(path):
            try:
                with Image.open(path) as im:
                    im.verify()
                with Image.open(path) as im:
                    width, height = im.size
                    has_alpha = im.mode in ("RGBA", "LA") or (
                        im.mode == "P" and "transparency" in im.info
                    )
            except Exception as exc:
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.invalid_image,
                    error=str(exc),
                )

            if (self.settings.min_width and width < self.settings.min_width) or (
                self.settings.min_height and height < self.settings.min_height
            ):
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.too_small,
                    error=f"{width}x{height} below min {self.settings.min_width}x{self.settings.min_height}",
                )

            if self.settings.skip_transparent and has_alpha:
                pct = self._transparency_pct(path)
                if pct >= self.settings.transparency_threshold:
                    return ImportResult(
                        status="skipped", skip_reason=SkipReason.too_transparent,
                        error=f"{pct:.2%} transparent",
                    )

            if self.settings.skip_single_color and self._single_color_hit(path):
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.single_color,
                    error=(f"one color dominates >"
                           f"{self.settings.single_color_threshold:.0%}"),
                )
        else:
            # Best-effort probe for dims + duration so downloaded videos can dedup
            # (#871). LENIENT: unlike _import_media this path does not reject on a
            # probe failure — preserve the existing "import the downloaded video"
            # behavior; we just skip dedup when we can't read the duration.
            vp = safe_probe.probe_video(path)
            if vp.ok:
                width, height, duration = vp.width, vp.height, vp.duration

        # Hash dedup
        sha = _sha256_of(path)
        existing = self.session.execute(
            select(ImageRecord).where(ImageRecord.sha256 == sha)
        ).scalar_one_or_none()
        if existing:
            # Enrich-on-duplicate (image_provenance docstring / spec §3): the
            # same bytes appearing in another post must show on THAT post too,
            # so append a provenance row for the new post rather than dropping
            # the linkage. primary_post_id stays on the first post — _apply_sidecar
            # only sets it when NULL. Without this the new post is left with no
            # image AND (if its other content also deduped) no attachment, i.e. a
            # bare shell — operator-flagged 2026-06-08 (1589 empty Anduo posts).
            self._apply_sidecar(existing, path, artist, explicit_source=source)
            self.session.commit()
            return ImportResult(
                status="skipped", skip_reason=SkipReason.duplicate_hash,
                image_id=existing.id, error="sha256 already present",
            )

        # phash dedup + supersede
        phash = None
        if not is_video(path):
            try:
                with Image.open(path) as im:
                    phash = compute_phash(im)
            except Exception:
                phash = None
            if phash is not None:
                candidates = self._phash_candidates_cache()
                rel, match_id = find_similar(
                    phash, width or 0, height or 0,
                    candidates, self.settings.phash_threshold,
                    confirm=self._pixel_confirmer(path),
                )
                if rel == "larger_exists":
                    # Enrich-on-duplicate: link the near-dup's post to the
                    # existing larger image so it shows on both (see duplicate_hash
                    # above). smaller_exists already links via _supersede.
                    larger = self.session.get(ImageRecord, match_id)
                    if larger is not None:
                        self._apply_sidecar(
                            larger, path, artist, explicit_source=source,
                        )
                        self.session.commit()
                    return ImportResult(
                        status="skipped",
                        skip_reason=SkipReason.duplicate_phash,
                        image_id=match_id,
                        error="perceptual near-duplicate of larger existing image",
                    )
                if rel == "smaller_exists":
                    target = self.session.get(ImageRecord, match_id)
                    self._supersede(
                        target, path, sha, phash, width, height,
                        new_path=path, artist=artist, source_row=source,
                    )
                    return ImportResult(status="superseded", image_id=match_id)
        elif duration is not None:
            # Tier-1 video near-dup (#871): same content re-downloaded from another
            # source/encode. artist is the subscription artist (passed explicitly).
            rel, match_id = self._find_similar_video(
                duration, width, height, artist.id if artist else None,
            )
            if rel == "larger_exists":
                larger = self.session.get(ImageRecord, match_id)
                if larger is not None:
                    self._apply_sidecar(larger, path, artist, explicit_source=source)
                    self.session.commit()
                return ImportResult(
                    status="skipped", skip_reason=SkipReason.duplicate_phash,
                    image_id=match_id,
                    error="video near-duplicate (duration+aspect) of existing equal/larger video",
                )
            if rel == "smaller_exists":
                target = self.session.get(ImageRecord, match_id)
                self._supersede(
                    target, path, sha, None, width, height,
                    new_path=path, artist=artist, source_row=source,
                    duration=duration,
                )
                return ImportResult(status="superseded", image_id=match_id)

        # Create record — the path IS where the file lives.
        record = ImageRecord(
            path=str(path),
            sha256=sha,
            phash=phash,
            size_bytes=path.stat().st_size,
            mime=_mime_for(path),
            width=width,
            height=height,
            duration_seconds=duration,
            origin="downloaded",
            integrity_status="unknown",
        )
        if artist is not None:
            record.artist_id = artist.id
        self.session.add(record)
        self.session.flush()
        self._phash_cache_append(phash, width, height, record.id)

        # Sidecar provenance (best-effort). When `source` is passed, link
        # the post to that subscription Source instead of creating a new
        # per-post Source row.
        self._apply_sidecar(record, path, artist, explicit_source=source)

        # Title-based WIP auto-tag (task #1458): fresh import only, see the
        # matching call in _import_media.
        self._maybe_apply_wip_title(record)

        self.session.commit()
        return ImportResult(status="imported", image_id=record.id)

    def _upsert_artist(self, name: str) -> Artist:
        slug = slugify(name)
        artist = self.session.execute(
            select(Artist).where(Artist.slug == slug)
        ).scalar_one_or_none()
        if artist is None:
            artist = Artist(name=name, slug=slug, is_subscription=False)
            self.session.add(artist)
            self.session.flush()
        return artist

    @staticmethod
    def _sidecar_artist_name(data: dict) -> str | None:
        for k in ("artist", "author_name"):
            v = data.get(k)
            if isinstance(v, str) and v.strip():
                return v.strip()
        for k in ("user", "creator"):
            v = data.get(k)
            if isinstance(v, str) and v.strip():
                return v.strip()
            if isinstance(v, dict):
                for sub in ("name", "full_name", "vanity", "account"):
                    sv = v.get(sub)
                    if isinstance(sv, str) and sv.strip():
                        return sv.strip()
        return None

    def relink_source_filehash(
        self, path: Path, source_url: str, *, artist: Artist | None = None,
    ) -> bool:
        """#830 recapture: backfill source_url + source_filehash on an EXISTING
        on-disk image's ImageRecord (matched by sha256) so its post-body inline
        `<img src=CDN>` remaps to the local copy at render time — WITHOUT
        re-downloading and WITHOUT unlinking the file (unlike the import path's
        duplicate-hash branch). NULL-only, mirroring _apply_sidecar: never
        clobbers an already-set filehash. Returns True only when a row was
        updated. No-op when the file is gone, the url has no filehash, no record
        matches the bytes, or the record already carries a filehash.
        """
        fh = filehash_from_url(source_url)
        if not fh:
            return False
        try:
            sha = _sha256_of(path)
        except OSError:
            return False
        record = self.session.execute(
            select(ImageRecord).where(ImageRecord.sha256 == sha)
        ).scalar_one_or_none()
        if record is None or record.source_filehash is not None:
            return False
        record.source_url = source_url
        record.source_filehash = fh
        self.session.commit()
        return True

    def link_existing_image_to_post(
        self, path: Path, external_post_id: str, *,
        source: Source | None = None, artist: Artist | None = None,
    ) -> bool:
        """Recapture back-link: associate an ALREADY-on-disk image (matched by
        its stored path) with its post — the link _apply_sidecar normally makes
        at per-media import time.

        Recapture disk-skips downloaded media, so a pre-existing image (e.g. one
        pulled under the old gallery-dl path, imported as a bare record with no
        post) never gets its `image_provenance` row / `primary_post_id`. This
        backfills it from the walk's (on-disk path, post external id) pairing:
        find the ImageRecord by path, find/attach the Post by (source,
        external_post_id), upsert provenance. Idempotent (issue #1288). Returns
        True when a record matched and was linked; no-op when the file has no
        record or no id is given."""
        if not external_post_id:
            return False
        record = self.session.execute(
            select(ImageRecord).where(ImageRecord.path == str(path))
        ).scalar_one_or_none()
        if record is None:
            return False
        artist_id = record.artist_id or (artist.id if artist else None)
        if artist_id is None:
            return False
        if record.artist_id is None:
            record.artist_id = artist_id
        post = self._find_or_create_post(
            source_id=source.id if source else None,
            external_post_id=str(external_post_id),
            artist_id=artist_id,
        )
        self._attach_provenance(
            record, post, source_id=source.id if source else None,
        )
        self.session.commit()
        return True

    def _attach_provenance(
        self, record: ImageRecord, post: Post, *,
        source_id: int | None, captured_metadata: dict | None = None,
    ) -> None:
        """Upsert the (image, post) `image_provenance` link + keep
        `primary_post_id` and the denormalized gallery sort keys aligned. Shared
        by _apply_sidecar (fresh per-media import) and link_existing_image_to_post
        (recapture back-link, #1288) so the two paths can't diverge. Idempotent.

        Race-safe (image_record_id, post_id) upsert — mirrors the
        _find_or_create_source/post savepoint pattern. The plain
        SELECT-then-INSERT pattern lost a race when two workers ran on the same
        (image, post) pair (e.g. the 5-min recovery sweep re-enqueued a
        still-running long import), planting duplicates that then broke
        .scalar_one_or_none() on every later deep-scan rederive
        (MultipleResultsFound). Alembic 0021's uq_image_provenance_image_post
        UNIQUE makes this savepoint trip on collision."""
        exists = self.session.execute(
            select(ImageProvenance.id).where(
                ImageProvenance.image_record_id == record.id,
                ImageProvenance.post_id == post.id,
            )
        ).scalar_one_or_none()
        if exists is None:
            sp = self.session.begin_nested()
            try:
                self.session.add(
                    ImageProvenance(
                        image_record_id=record.id,
                        post_id=post.id,
                        source_id=source_id,
                        captured_metadata=captured_metadata,
                    )
                )
                self.session.flush()
                sp.commit()
            except IntegrityError:
                sp.rollback()
        if record.primary_post_id is None:
            record.primary_post_id = post.id
        # Keep the denormalized gallery sort key (alembic 0035) aligned with
        # the primary post's publish date so /scroll orders off
        # ix_image_record_effective_date instead of COALESCE-ing across the
        # post join. Only override when THIS post is the primary AND carries
        # a date; otherwise the column keeps its created_at-equivalent server
        # default (matches the old COALESCE(post_date, created_at) fallback).
        if record.primary_post_id == post.id and post.post_date is not None:
            record.effective_date = post.post_date
        # earliest_post_date (alembic 0071) = MIN(post_date) across ALL of this
        # image's provenance posts, not just the primary — so the gallery can
        # sort by original publish rather than the download/repost the primary
        # points at. Recompute from provenance whenever a dated post is linked;
        # the provenance row for THIS post was committed above, so the MIN
        # includes it. Leaves the created_at default when no linked post is dated.
        if post.post_date is not None:
            earliest = self.session.execute(
                select(func.min(Post.post_date))
                .select_from(ImageProvenance)
                .join(Post, Post.id == ImageProvenance.post_id)
                .where(
                    ImageProvenance.image_record_id == record.id,
                    Post.post_date.is_not(None),
                )
            ).scalar_one_or_none()
            if earliest is not None:
                record.earliest_post_date = earliest
        self.session.flush()

    def _apply_sidecar(
        self,
        record: ImageRecord,
        source: Path,
        artist: Artist | None,
        *,
        explicit_source: Source | None = None,
    ) -> None:
        """Read the sidecar adjacent to `source` and wire up provenance.

        If `explicit_source` is passed (FC-3c attach_in_place case), the
        sidecar's post is linked to that Source instead of looking up /
        creating one keyed by the sidecar's post_url. Without it, the
        filesystem importer's per-post-Source creation is preserved.
        """
        sc = find_sidecar(source)
        if sc is None:
            return
        try:
            data = json.loads(sc.read_text("utf-8"))
            if not isinstance(data, dict):
                raise ValueError("sidecar JSON is not an object")
        except Exception as exc:
            log.warning("sidecar parse failed for %s: %s", sc, exc)
            return

        sd = parse_sidecar(data)

        if artist is None:
            name = self._sidecar_artist_name(data)
            if name:
                artist = self._upsert_artist(name)
        if artist is None:
            log.warning(
                "sidecar present for %s but no artist; skipping provenance",
                source,
            )
            return

        if record.artist_id is None:
            record.artist_id = artist.id

        # #830 Phase 2: persist the file's CDN source identity (NULL-only, so a
        # re-import / enrich-on-duplicate never clobbers it) — the filehash is
        # the join key that lets post_feed_service remap the body's inline
        # `<img src=CDN>` to this local copy at render time.
        if sd.source_url and record.source_filehash is None:
            record.source_url = sd.source_url
            record.source_filehash = filehash_from_url(sd.source_url)

        if explicit_source is not None:
            src = explicit_source
        else:
            platform = sd.platform or "unknown"
            src = self._lookup_source_for_sidecar(
                artist_id=artist.id, platform=platform,
            )

        epid = sd.external_post_id or sc.stem
        post = self._find_or_create_post(
            source_id=src.id if src else None,
            external_post_id=epid,
            artist_id=artist.id,
        )
        # Post-first (#856): on the native path the post-record owns the body/
        # links/metadata, so the per-media import must NOT write post fields — its
        # minimal sidecar carries no body, and applying it would clobber
        # raw_metadata + re-run link-sync off empty data. The image still links
        # provenance + localization below. gallery-dl (post_first False) is
        # unchanged: its sidecar IS the only body source, so it still applies.
        if not self.post_first:
            self._apply_post_fields(post, sd)

        # Link the (image, post) provenance + keep primary_post_id / the gallery
        # sort keys aligned — shared with the recapture back-link path (#1288).
        self._attach_provenance(
            record, post, source_id=src.id if src else None,
            captured_metadata=sd.raw,
        )

    def _copy_to_library(
        self, source: Path, sha: str, attribution_path: Path,
        artist: Artist | None = None,
    ) -> Path:
        """Copy `source` to its final library path. Returns the destination.

        Atomic write: copies to .partial then renames. Shared by
        _import_media (filesystem scan) and _supersede (when new_path is
        not passed). FC-3c's attach_in_place skips this helper entirely
        — the file is already at its final home.

        `artist`, when resolved, decides the top-level directory: the
        library is keyed on the Artist row's slug, NOT on however the
        import folder happened to be capitalised. Without that, an import
        from `/import/Conto/` and a download for the same artist write to
        `Conto/` and `conto/` respectively and the library grows a second
        home for one artist (milestone #421).
        """
        subdir = canonical_subdir(
            derive_subdir(attribution_path, self.import_root),
            artist.slug if artist else None,
        )
        dest_dir = self.images_root / subdir if subdir else self.images_root
        dest_dir.mkdir(parents=True, exist_ok=True)
        dest_name = hash_suffixed_name(source.stem, sha, source.suffix)
        dest = dest_dir / dest_name
        partial = dest.with_suffix(dest.suffix + ".partial")
        shutil.copy2(source, partial)
        partial.rename(dest)
        return dest

    def _supersede(
        self, existing: ImageRecord, source: Path, sha: str,
        phash: str, width: int | None, height: int | None,
        *, new_path: Path | None = None,
        artist: Artist | None = None,
        source_row: Source | None = None,
        duration: float | None = None,
    ) -> None:
        """Replace `existing`'s file with the larger `source`, keeping the
        row id (so tags/series/curation stay attached). ML is cleared so
        the import task re-derives it on the new pixels.

        After the file swap, the new file's adjacent gallery-dl sidecar
        (if any) is applied via _apply_sidecar — operator-flagged
        2026-05-25: scanning a GS download dir with smaller IR-migrated
        images on the receiving end used to swap files but lose the GS
        sidecar's post metadata entirely. _apply_sidecar is additive
        (find-or-create Post / Source / ImageProvenance, NULL-only
        primary_post_id update) so any pre-existing Post linkage
        survives untouched.

        If `new_path` is provided, `source` is assumed to ALREADY be at
        that path (FC-3c attach_in_place case) — skip the copy step.
        Otherwise the file is copied via _copy_to_library."""
        if new_path is None:
            # The KEPT row's artist decides the destination, not the incoming
            # file's folder — a supersede rewrites `existing.path`, so writing
            # it anywhere but that artist's canonical directory would move a
            # row OUT of the tree milestone #421 is consolidating. ImageRecord
            # carries `artist_id` with no relationship attribute, so this is a
            # lookup rather than `existing.artist`.
            kept_artist = artist
            if kept_artist is None and existing.artist_id is not None:
                kept_artist = self.session.get(Artist, existing.artist_id)
            dest = self._copy_to_library(source, sha, source, kept_artist)
        else:
            dest = new_path

        old_path = existing.path
        old_thumb = existing.thumbnail_path

        existing.path = str(dest)
        existing.sha256 = sha
        existing.phash = phash
        existing.size_bytes = dest.stat().st_size
        existing.mime = _mime_for(source)
        existing.width = width
        existing.height = height
        existing.duration_seconds = duration  # #871: keep the kept copy's duration
        existing.thumbnail_path = None
        existing.integrity_status = "unknown"
        existing.siglip_embedding = None
        existing.siglip_model_version = None
        # created_at intentionally preserved; updated_at auto-bumps.
        self.session.flush()
        self.session.commit()
        # The phash candidate cache (used to avoid N+1 selects during
        # archive imports) is now stale for `existing.id` — the row's
        # phash/dimensions changed. Invalidate; the next call re-fetches.
        self._phash_candidates = None

        # Sidecar enrichment from the new (larger) file's location.
        # _apply_sidecar resolves artist from the sidecar itself if the
        # existing row has none, and is internally guarded against
        # missing-or-malformed sidecars (silent return).
        # Audit 2026-06-02: thread artist/source_row from the
        # download-path caller (attach_in_place smaller_exists branch)
        # so the supersede preserves explicit subscription linkage
        # instead of re-deriving via path-walk.
        try:
            self._apply_sidecar(
                existing, source, artist, explicit_source=source_row,
            )
        except Exception as exc:
            # Don't unwind the supersede DB swap if sidecar parsing
            # blows up unexpectedly — the file replacement is the
            # critical operation, sidecar is enrichment.
            log.warning(
                "sidecar enrichment failed during supersede of "
                "image_record.id=%s from %s: %s",
                existing.id, source, exc,
            )

        for stale in (old_path, old_thumb):
            if not stale or stale == str(dest):
                # If the supersede kept the file in place (new_path == old
                # location), don't delete it.
                continue
            try:
                p = Path(stale)
                if p.exists():
                    p.unlink()
            except Exception:
                # Benign orphan; the DB swap already committed. Don't undo it.
                pass

    # Matches the Cleanup audit card's default tolerance: the import-side
    # filter and the retroactive audit must agree on what "single color" MEANS
    # (Euclidean RGB distance to the dominant color); only the match threshold
    # is operator-tunable per surface.
    _SINGLE_COLOR_TOLERANCE = 30

    def _single_color_hit(self, source: Path) -> bool:
        """True when one color dominates beyond the configured threshold — the
        same canonical predicate the Cleanup audit runs (audits.single_color,
        whose docstring anticipated this adoption; the skip_single_color
        setting existed but was never wired until 2026-07-02). Never raises:
        unreadable files were already rejected by verify() upstream, and a
        residual decode error just declines to match (the import proceeds)."""
        try:
            with Image.open(source) as im:
                if getattr(im, "is_animated", False):
                    # Frame 0 only would misjudge animations; skip like the
                    # transparency check does.
                    return False
                return single_color.evaluate(
                    im,
                    threshold=self.settings.single_color_threshold,
                    tolerance=self._SINGLE_COLOR_TOLERANCE,
                )
        except Exception:
            return False

    def _transparency_pct(self, source: Path) -> float:
        """Fraction of fully-transparent pixels in the image. 0.0 if no alpha.

        For animated formats (multi-frame WebP / GIF / APNG), short-circuit
        to 0.0 instead of decoding every frame. PIL's `getchannel("A")`
        forces a full decode of all frames in an animated image, which for
        a large animated WebP takes 5+ minutes and blows past the Celery
        soft+hard time limits (300s/360s → SIGKILL). Operator-flagged
        2026-05-26. Transparency analysis on a multi-frame image isn't
        meaningful for art-curation purposes anyway — different frames
        have different alpha — so the existing too_transparent skip rule
        is bypassed entirely for animated content.
        """
        with Image.open(source) as im:
            if getattr(im, "is_animated", False):
                log.info(
                    "skipping transparency check for animated image %s "
                    "(n_frames=%d) — avoids multi-frame decode timeout",
                    source, getattr(im, "n_frames", 0),
                )
                return 0.0
            if im.mode not in ("RGBA", "LA") and not (
                im.mode == "P" and "transparency" in im.info
            ):
                return 0.0
            if im.mode != "RGBA":
                im = im.convert("RGBA")
            alpha = im.getchannel("A")
            histogram = alpha.histogram()
            transparent = histogram[0]
            total = sum(histogram)
            return transparent / total if total else 0.0