fc5: ir_ingest — Tag rows (skip artist/post) + manifest JSON for sha256-keyed tag_apply

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

backend/app/services/migrators/ir_ingest.py

@@ -0,0 +1,141 @@
+"""ImageRepo export → FabledCurator ingest.
+
+Reads a parsed imagerepo-export-v1.json dict (no DB connection to IR).
+Creates Tag rows (skipping artist/post kinds, resolving fandom_name to
+FK). Writes the per-image-sha256 artist assignments + tag associations
++ series page assignments to /images/_migration_state/ir_tag_manifest.json
+so tag_apply.py can join them to ImageRecord rows AFTER the operator
+runs FC's filesystem scan.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ...models import Tag, TagKind
+
+_SKIP_KINDS = frozenset({"artist", "post"})
+_MIGRATION_STATE_DIRNAME = "_migration_state"
+_IR_MANIFEST_FILENAME = "ir_tag_manifest.json"
+
+
+def _zero_counts() -> dict:
+    return {
+        "rows_processed": 0, "rows_inserted": 0, "rows_skipped": 0,
+        "files_copied": 0, "bytes_copied": 0, "conflicts": 0,
+    }
+
+
+def manifest_path(images_root: Path | None = None) -> Path:
+    root = images_root if images_root is not None else Path("/images")
+    p = root / _MIGRATION_STATE_DIRNAME
+    p.mkdir(parents=True, exist_ok=True)
+    return p / _IR_MANIFEST_FILENAME
+
+
+async def _resolve_fandom_id(
+    db: AsyncSession, fandom_name: str | None, dry_run: bool,
+) -> int | None:
+    """Find-or-create a fandom-kind Tag by name."""
+    if not fandom_name:
+        return None
+    existing = (await db.execute(
+        select(Tag).where(Tag.name == fandom_name, Tag.kind == "fandom")
+    )).scalar_one_or_none()
+    if existing is not None:
+        return existing.id
+    if dry_run:
+        return None
+    t = Tag(name=fandom_name, kind=TagKind.fandom)
+    db.add(t)
+    await db.flush()
+    return t.id
+
+
+async def migrate_async(
+    db: AsyncSession,
+    *,
+    data: dict,
+    images_root: Path | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Ingest a parsed imagerepo-export-v1.json dict.
+
+    Creates Tag rows + writes the IR tag manifest file. Tag-to-image
+    binding happens later in tag_apply.py (after FC's filesystem scan
+    populates image_record.sha256 → id).
+    """
+    if data.get("source_app") != "imagerepo":
+        raise ValueError("export source_app must be 'imagerepo'")
+    if data.get("schema_version") != 1:
+        raise ValueError(f"unsupported schema_version: {data.get('schema_version')}")
+
+    counts = _zero_counts()
+
+    # Phase 1: tags (skip artist + post kinds; resolve fandom_name → fandom_id).
+    # First pass: create all fandom-kind tags so they're available for FK resolution.
+    for tag in data.get("tags", []):
+        kind = tag.get("kind") or "general"
+        if kind != "fandom":
+            continue
+        counts["rows_processed"] += 1
+        existing = (await db.execute(
+            select(Tag).where(Tag.name == tag["name"], Tag.kind == "fandom")
+        )).scalar_one_or_none()
+        if existing is not None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        db.add(Tag(name=tag["name"], kind=TagKind.fandom))
+        counts["rows_inserted"] += 1
+    if not dry_run:
+        await db.flush()
+
+    # Second pass: every other kind.
+    for tag in data.get("tags", []):
+        kind_str = tag.get("kind") or "general"
+        if kind_str in _SKIP_KINDS:
+            counts["rows_skipped"] += 1
+            continue
+        if kind_str == "fandom":
+            continue  # handled above
+        counts["rows_processed"] += 1
+        try:
+            kind = TagKind(kind_str)
+        except ValueError:
+            kind = TagKind.general
+        fandom_id = await _resolve_fandom_id(db, tag.get("fandom_name"), dry_run)
+        existing = (await db.execute(
+            select(Tag).where(Tag.name == tag["name"], Tag.kind == kind)
+        )).scalar_one_or_none()
+        if existing is not None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        db.add(Tag(name=tag["name"], kind=kind, fandom_id=fandom_id))
+        counts["rows_inserted"] += 1
+
+    if not dry_run:
+        await db.commit()
+
+    # Phase 2: write the per-image manifest for tag_apply.py to consume later.
+    manifest = {
+        "schema_version": 1,
+        "image_artist_assignments": data.get("image_artist_assignments", []),
+        "image_tag_associations": data.get("image_tag_associations", []),
+        "series_pages": data.get("series_pages", []),
+    }
+    counts["rows_processed"] += len(manifest["image_artist_assignments"])
+    counts["rows_processed"] += len(manifest["image_tag_associations"])
+    counts["rows_processed"] += len(manifest["series_pages"])
+    if not dry_run:
+        manifest_path(images_root).write_text(json.dumps(manifest, indent=2))
+
+    return counts