feat(fc2a): add Importer service

Single-file pipeline: validate as image, apply filter rules (min dimensions, transparency), SHA256 hash dedup, atomic copy to /images/{subdir}/, create ImageRecord with origin='imported_filesystem' and integrity_status='unknown', auto-derive top-level folder name as Artist + artist tag. Importer is intentionally sync (consumed by a Celery worker process); the async Quart side uses the same ORM through its own async session. The db_sync fixture in conftest.py was added in Task 3 to support these tests. Thumbnail generation is NOT inlined; the calling Celery task enqueues a separate thumbnail task so the import queue keeps moving on big batches. pHash dedup is FC-2d, not here. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 12:07:01 -04:00
parent 1a2a120da1
commit bb9c183ff7
2 changed files with 380 additions and 0 deletions
@@ -0,0 +1,248 @@
+"""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 shutil
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+
+from PIL import Image
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ..models import Artist, ImageRecord, ImportSettings, Tag, TagKind
+from ..models.tag import image_tag
+from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
+from ..utils.slug import slugify
+from .thumbnailer import Thumbnailer
+
+
+class SkipReason(str, Enum):
+    too_small = "too_small"
+    too_transparent = "too_transparent"
+    single_color = "single_color"
+    duplicate_hash = "duplicate_hash"
+    invalid_image = "invalid_image"
+
+
+@dataclass(frozen=True)
+class ImportResult:
+    status: str  # 'imported' | 'skipped' | 'failed'
+    image_id: int | None = None
+    skip_reason: SkipReason | None = None
+    error: str | None = None
+
+
+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 _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,
+    ):
+        self.session = session
+        self.images_root = images_root
+        self.import_root = import_root
+        self.thumbnailer = thumbnailer
+        self.settings = settings
+
+    def import_one(self, source: Path) -> ImportResult:
+        """The full pipeline for a single file.
+
+        Returns an ImportResult; callers (the Celery task) are responsible for
+        translating that into ImportTask.status and ImageRecord linkage.
+        """
+        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 not is_video(source):
+            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:
+                pct = self._transparency_pct(source)
+                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(source)
+        existing_stmt = select(ImageRecord).where(ImageRecord.sha256 == sha)
+        existing = self.session.execute(existing_stmt).scalar_one_or_none()
+        if existing:
+            return ImportResult(
+                status="skipped", skip_reason=SkipReason.duplicate_hash,
+                image_id=existing.id, error="sha256 already present",
+            )
+
+        # Destination path.
+        subdir = derive_subdir(source, 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
+
+        # Atomic copy: write to .partial, rename. Avoids half-imported files on crash.
+        partial = dest.with_suffix(dest.suffix + ".partial")
+        shutil.copy2(source, partial)
+        partial.rename(dest)
+
+        record = ImageRecord(
+            path=str(dest),
+            sha256=sha,
+            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()
+
+        # Folder→artist auto-tag.
+        artist_name = derive_top_level_artist(source, self.import_root)
+        if artist_name:
+            self._attach_artist_tag(record.id, artist_name)
+
+        # 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 _attach_artist_tag(self, image_id: int, artist_name: str) -> None:
+        """Idempotently creates Artist + artist tag, attaches to image."""
+        slug = slugify(artist_name)
+        artist = self.session.execute(
+            select(Artist).where(Artist.slug == slug)
+        ).scalar_one_or_none()
+        if artist is None:
+            artist = Artist(name=artist_name, slug=slug, is_subscription=False)
+            self.session.add(artist)
+            self.session.flush()
+
+        tag = self.session.execute(
+            select(Tag).where(Tag.name == artist_name).where(Tag.kind == TagKind.artist)
+        ).scalar_one_or_none()
+        if tag is None:
+            tag = Tag(name=artist_name, kind=TagKind.artist)
+            self.session.add(tag)
+            self.session.flush()
+
+        # Idempotent association; using a raw insert + on-conflict-do-nothing
+        # is cleaner here than fetching first, but for a sync session we just
+        # check existence.
+        already = self.session.execute(
+            select(image_tag.c.tag_id)
+            .where(image_tag.c.image_record_id == image_id)
+            .where(image_tag.c.tag_id == tag.id)
+        ).scalar_one_or_none()
+        if already is None:
+            self.session.execute(
+                image_tag.insert().values(
+                    image_record_id=image_id, tag_id=tag.id, source="auto",
+                )
+            )
+
+    def _transparency_pct(self, source: Path) -> float:
+        """Fraction of fully-transparent pixels in the image. 0.0 if no alpha."""
+        with Image.open(source) as im:
+            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