feat(fc3c): Importer.attach_in_place + _supersede(new_path=) + _copy_to_library helper

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/services/importer.py

@@ -335,17 +335,7 @@ class Importer:
                         status="superseded", image_id=match_id
                     )
 
-        # Destination path (artist/subdir anchored to attribution_path).
-        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
-
-        # 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)
+        dest = self._copy_to_library(source, sha, attribution_path)
 
         record = ImageRecord(
             path=str(dest),
@@ -407,6 +397,143 @@ class Importer:
             image_id=existing.id, error="deep: re-derived",
         )
 
+    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:
+                cand_rows = self.session.execute(
+                    select(
+                        ImageRecord.phash,
+                        ImageRecord.width,
+                        ImageRecord.height,
+                        ImageRecord.id,
+                    ).where(ImageRecord.phash.is_not(None))
+                ).all()
+                candidates = [
+                    (c.phash, c.width or 0, c.height or 0, c.id)
+                    for c in cand_rows
+                ]
+                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="imported_download",
+            integrity_status="unknown",
+        )
+        if artist is not None:
+            record.artist_id = artist.id
+        self.session.add(record)
+        self.session.flush()
+
+        # 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(
@@ -447,8 +574,20 @@ class Importer:
         return None
 
     def _apply_sidecar(
-        self, record: ImageRecord, source: Path, artist: Artist | None
+        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
@@ -476,19 +615,22 @@ class Importer:
         if record.artist_id is None:
             record.artist_id = artist.id
 
-        platform = sd.platform or "unknown"
-        url = sd.post_url or f"sidecar:{platform}"
-        src = self.session.execute(
-            select(Source).where(
-                Source.artist_id == artist.id,
-                Source.platform == platform,
-                Source.url == url,
-            )
-        ).scalar_one_or_none()
-        if src is None:
-            src = Source(artist_id=artist.id, platform=platform, url=url)
-            self.session.add(src)
-            self.session.flush()
+        if explicit_source is not None:
+            src = explicit_source
+        else:
+            platform = sd.platform or "unknown"
+            url = sd.post_url or f"sidecar:{platform}"
+            src = self.session.execute(
+                select(Source).where(
+                    Source.artist_id == artist.id,
+                    Source.platform == platform,
+                    Source.url == url,
+                )
+            ).scalar_one_or_none()
+            if src is None:
+                src = Source(artist_id=artist.id, platform=platform, url=url)
+                self.session.add(src)
+                self.session.flush()
 
         epid = sd.external_post_id or sc.stem
         post = self.session.execute(
@@ -532,20 +674,42 @@ class Importer:
             record.primary_post_id = post.id
         self.session.flush()
 
-    def _supersede(
-        self, existing: ImageRecord, source: Path, sha: str,
-        phash: str, width: int | None, height: int | 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."""
-        subdir = derive_subdir(source, self.import_root)
+    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 = dest_dir / hash_suffixed_name(source.stem, sha, source.suffix)
+        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.
+
+        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
@@ -569,7 +733,9 @@ class Importer:
         self.session.commit()
 
         for stale in (old_path, old_thumb):
-            if not stale:
+            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)