bvandeusen
6fc8ae3106
Two coupled bugs surfaced 2026-05-31 by the Subscriptions UI showing "phantom" subscriptions like `sidecar:patreon:dpmaker`: 1. `SourceService.list()` returned every Source, no filter on URL. alembic 0022 (2026-05-26) consolidated old per-post-URL Sources into one canonical row per (artist, platform); when no real campaign URL was salvageable it rewrote the canonical to `sidecar:<plat>:<slug>` enabled=false as a disabled anchor. The UI then listed those anchors as if they were polls — disabled, but visible. Fix: `list()` excludes `url LIKE 'sidecar:%'` by default; `include_synthetic=True` opts back in for admin tooling. 2. `importer._source_for_sidecar` picked the lowest-id Source for (artist, platform). When alembic 0022 had rewritten a per-post row into a synthetic anchor (lower id) AND the operator later added the real subscription (higher id), every gallery-dl download silently attached its Post to the SYNTHETIC instead of the real Source. Fix: prefer a non-`sidecar:%` URL when one exists; fall back to the synthetic; only create a new synthetic when nothing exists for (artist, platform). alembic 0028 is the data half: for every (artist, platform) with both a synthetic AND a real Source, pre-merge Post+ImageProvenance collisions on the canonical, bulk-repoint Posts/ImageProvenance/ DownloadEvent.source_id onto the real Source, and delete the synthetic. Lone synthetics (no real twin) are left intact — they anchor real imported content the operator may still want; the list-filter hides them so they no longer surface as phantoms.
1048 lines
43 KiB
Python
1048 lines
43 KiB
Python
"""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 select
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
Artist,
|
||
ImageProvenance,
|
||
ImageRecord,
|
||
ImportSettings,
|
||
Post,
|
||
PostAttachment,
|
||
Source,
|
||
)
|
||
from ..utils import safe_probe
|
||
from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
|
||
from ..utils.phash import compute_phash, find_similar
|
||
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 .thumbnailer import Thumbnailer
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
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
|
||
|
||
|
||
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)).
|
||
|
||
gallery-dl produces some filenames with URL-encoded query-string
|
||
artifacts embedded into the basename (e.g.
|
||
`79507046_media_..._https___www.patreon.com_media-u_Z0FBQUFBQm5q...`).
|
||
`Path.suffix` finds the LAST dot and returns everything after, which
|
||
in those cases yields a 50+ char "extension" of mostly base64-ish
|
||
junk. That blows the column. Operator-flagged 2026-05-25.
|
||
|
||
Real extensions are short and alphanumeric. We accept anything ≤ 16
|
||
chars where every post-dot character is alphanumeric; anything else
|
||
means the input wasn't a real extension and we return the empty
|
||
string. ext is nullable-ish (empty string still satisfies NOT NULL)
|
||
and consumers should treat "" as "no known extension".
|
||
"""
|
||
suffix = path.suffix.lower()
|
||
if not suffix or len(suffix) > 16:
|
||
return ""
|
||
if not all(c.isalnum() for c in suffix[1:]):
|
||
return ""
|
||
return suffix
|
||
|
||
|
||
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
|
||
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
|
||
|
||
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 _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,
|
||
_source_for_sidecar, 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 _source_for_sidecar(
|
||
self, *, artist_id: int, platform: str, artist_slug: str,
|
||
) -> Source:
|
||
"""Sidecar-import Source resolver. Used by both filesystem imports
|
||
and gallery-dl downloads (both write sidecar JSON, both flow through
|
||
_apply_sidecar / _capture_attachment).
|
||
|
||
Source represents a subscription feed (one per artist+platform — the
|
||
URL polled by the FC-3 downloader). The filesystem importer used to
|
||
call _find_or_create_source(url=sd.post_url), creating one Source
|
||
row per post URL — 100s of junk Sources per artist, all with
|
||
enabled=True, polluting the artist detail page and tricking the
|
||
subscription checker into trying to poll patreon post URLs as feeds.
|
||
Operator-flagged 2026-05-26; consolidated via alembic 0022.
|
||
|
||
Resolution order: prefer a real (non-sidecar) Source over a
|
||
synthetic anchor. When alembic 0022 ran, it may have rewritten
|
||
per-post Sources into `sidecar:<platform>:<slug>` synthetic
|
||
anchors. If the operator later added the real subscription, both
|
||
rows now coexist. A naive `ORDER BY id ASC LIMIT 1` lookup would
|
||
pick the older synthetic and silently attach every gallery-dl
|
||
download to the wrong Source — operator-flagged 2026-05-31 after
|
||
the Subscriptions UI surfaced the phantom anchors. Pick the real
|
||
one when one exists; fall back to the synthetic; only create a
|
||
new synthetic when nothing exists for (artist, platform).
|
||
"""
|
||
real_stmt = (
|
||
select(Source)
|
||
.where(
|
||
Source.artist_id == artist_id,
|
||
Source.platform == platform,
|
||
~Source.url.like("sidecar:%"),
|
||
)
|
||
.order_by(Source.id.asc())
|
||
.limit(1)
|
||
)
|
||
real = self.session.execute(real_stmt).scalar_one_or_none()
|
||
if real is not None:
|
||
return real
|
||
|
||
any_stmt = (
|
||
select(Source)
|
||
.where(
|
||
Source.artist_id == artist_id,
|
||
Source.platform == platform,
|
||
)
|
||
.order_by(Source.id.asc())
|
||
.limit(1)
|
||
)
|
||
return self._get_or_create(
|
||
any_stmt,
|
||
lambda: Source(
|
||
artist_id=artist_id,
|
||
platform=platform,
|
||
url=f"sidecar:{platform}:{artist_slug}",
|
||
enabled=False,
|
||
),
|
||
)
|
||
|
||
def _find_or_create_post(
|
||
self, *, source_id: int, external_post_id: str,
|
||
) -> Post:
|
||
"""Race-safe find-or-create on `post` keyed by
|
||
(source_id, external_post_id). Mirrors `_find_or_create_source`
|
||
— same savepoint + IntegrityError-recovery pattern."""
|
||
stmt = select(Post).where(
|
||
Post.source_id == source_id,
|
||
Post.external_post_id == external_post_id,
|
||
)
|
||
return self._get_or_create(
|
||
stmt,
|
||
lambda: Post(source_id=source_id, external_post_id=external_post_id),
|
||
)
|
||
|
||
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._source_for_sidecar(
|
||
artist_id=artist.id, platform=platform, artist_slug=artist.slug,
|
||
)
|
||
epid = sd.external_post_id or sc.stem
|
||
return self._find_or_create_post(
|
||
source_id=src.id, external_post_id=epid,
|
||
)
|
||
|
||
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)
|
||
existing = self.session.execute(
|
||
select(PostAttachment).where(PostAttachment.sha256 == sha)
|
||
).scalar_one_or_none()
|
||
if existing is None:
|
||
stored = self.attachments.store(source, sha)
|
||
self.session.add(PostAttachment(
|
||
post_id=post.id if post else None,
|
||
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()
|
||
self.session.commit()
|
||
return ImportResult(status="attached")
|
||
|
||
def _import_archive(self, source: Path) -> 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.
|
||
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 = self._resolve_artist(source)
|
||
post = self._post_for_sidecar(source, artist)
|
||
self._capture_attachment(source, post=post, artist=artist, resolved=True)
|
||
return ImportResult(status="attached")
|
||
|
||
artist = self._resolve_artist(source)
|
||
post = self._post_for_sidecar(source, artist)
|
||
member_ids: list[int] = []
|
||
with extract_archive(source) as members:
|
||
for _name, member_path in members:
|
||
if not is_supported(member_path):
|
||
continue # non-media preserved via the stored archive
|
||
res = self._import_media(member_path, source)
|
||
if res.status in ("imported", "superseded") and res.image_id:
|
||
member_ids.append(res.image_id)
|
||
# Preserve the archive itself (links to the same Post/Artist).
|
||
self._capture_attachment(
|
||
source, post=post, artist=artist, resolved=True
|
||
)
|
||
if member_ids:
|
||
return ImportResult(
|
||
status="imported", image_id=member_ids[0],
|
||
member_image_ids=member_ids,
|
||
)
|
||
return ImportResult(status="attached")
|
||
|
||
def _import_media(
|
||
self, source: Path, attribution_path: Path
|
||
) -> 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
|
||
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 for free (the
|
||
# importer didn't previously record video dimensions).
|
||
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 = probe.width, probe.height
|
||
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",
|
||
)
|
||
|
||
# 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:
|
||
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,
|
||
)
|
||
if rel == "larger_exists":
|
||
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
|
||
)
|
||
|
||
dest = self._copy_to_library(source, sha, attribution_path)
|
||
|
||
record = ImageRecord(
|
||
path=str(dest),
|
||
sha256=sha,
|
||
phash=phash,
|
||
size_bytes=dest.stat().st_size,
|
||
mime=_mime_for(source),
|
||
width=width,
|
||
height=height,
|
||
origin="imported_filesystem",
|
||
integrity_status="unknown",
|
||
)
|
||
self.session.add(record)
|
||
self.session.flush()
|
||
self._phash_cache_append(phash, width, height, record.id)
|
||
|
||
# Folder→artist (anchored to attribution_path).
|
||
artist = None
|
||
artist_name = derive_top_level_artist(
|
||
attribution_path, self.import_root
|
||
)
|
||
if artist_name:
|
||
artist = self._attach_artist(record, artist_name)
|
||
|
||
# Sidecar provenance (best-effort; never fails the import).
|
||
self._apply_sidecar(record, attribution_path, artist)
|
||
|
||
# Thumbnail is queued separately by the calling task; the importer
|
||
# does not generate thumbnails inline so the import queue stays moving.
|
||
|
||
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 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`.
|
||
|
||
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
|
||
- failed → file untouched; caller decides
|
||
"""
|
||
if not is_supported(path):
|
||
return ImportResult(
|
||
status="skipped", skip_reason=SkipReason.invalid_image,
|
||
error=f"unsupported extension {path.suffix}",
|
||
)
|
||
|
||
# Format / dimension / transparency filters (mirror _import_media).
|
||
width = height = None
|
||
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",
|
||
)
|
||
|
||
# Hash dedup
|
||
sha = _sha256_of(path)
|
||
existing = self.session.execute(
|
||
select(ImageRecord).where(ImageRecord.sha256 == sha)
|
||
).scalar_one_or_none()
|
||
if existing:
|
||
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,
|
||
)
|
||
if rel == "larger_exists":
|
||
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
|
||
)
|
||
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,
|
||
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)
|
||
|
||
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
|
||
|
||
def _attach_artist(self, record: ImageRecord, artist_name: str) -> Artist:
|
||
"""Upsert the Artist and set it as the image's canonical artist.
|
||
Artist-kind tags were retired in FC-2d-vii-c — the Artist row is
|
||
the single source of truth. Returns the Artist (sidecar
|
||
provenance attaches its Source to it)."""
|
||
artist = self._upsert_artist(artist_name)
|
||
if record.artist_id is None:
|
||
record.artist_id = artist.id
|
||
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 _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
|
||
|
||
if explicit_source is not None:
|
||
src = explicit_source
|
||
else:
|
||
platform = sd.platform or "unknown"
|
||
src = self._source_for_sidecar(
|
||
artist_id=artist.id, platform=platform,
|
||
artist_slug=artist.slug,
|
||
)
|
||
|
||
epid = sd.external_post_id or sc.stem
|
||
post = self._find_or_create_post(
|
||
source_id=src.id, external_post_id=epid,
|
||
)
|
||
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
|
||
|
||
# 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
|
||
# _apply_sidecar 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 adds the
|
||
# uq_image_provenance_image_post UNIQUE so this savepoint actually
|
||
# trips 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=src.id,
|
||
captured_metadata=sd.raw,
|
||
)
|
||
)
|
||
self.session.flush()
|
||
sp.commit()
|
||
except IntegrityError:
|
||
sp.rollback()
|
||
if record.primary_post_id is None:
|
||
record.primary_post_id = post.id
|
||
self.session.flush()
|
||
|
||
def _copy_to_library(
|
||
self, source: Path, sha: str, attribution_path: Path
|
||
) -> 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.
|
||
"""
|
||
subdir = derive_subdir(attribution_path, self.import_root)
|
||
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,
|
||
) -> 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:
|
||
dest = self._copy_to_library(source, sha, source)
|
||
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.thumbnail_path = None
|
||
existing.integrity_status = "unknown"
|
||
existing.tagger_predictions = None
|
||
existing.tagger_model_version = None
|
||
existing.siglip_embedding = None
|
||
existing.siglip_model_version = None
|
||
existing.centroid_scores = 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).
|
||
try:
|
||
self._apply_sidecar(existing, source, None)
|
||
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
|
||
|
||
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
|