fc5: tag_apply — joins manifest entries to ImageRecord by sha256 after filesystem scan

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

backend/app/services/migrators/tag_apply.py

@@ -0,0 +1,172 @@
+"""Apply the IR tag manifest after FC's filesystem scan.
+
+Reads /images/_migration_state/ir_tag_manifest.json and joins each entry
+to an ImageRecord row by sha256 (which exists after the operator runs
+FC's filesystem scan over the mounted IR images dir).
+
+- image_artist_assignments → ImageRecord.artist_id (find_or_create Artist by slug).
+- image_tag_associations  → image_tag insert (idempotent).
+- series_pages            → series_page insert (idempotent on image_id unique).
+
+Unmatched sha256s are logged into the result's `unmatched` list so the
+Celery task can drop them into MigrationRun.metadata for the operator
+to inspect.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ...models import Artist, ImageRecord, SeriesPage, Tag, TagKind, image_tag
+from ...utils.slug import slugify
+from .ir_ingest import manifest_path
+
+
+def _zero_counts() -> dict:
+    return {
+        "rows_processed": 0, "rows_inserted": 0, "rows_skipped": 0,
+        "files_copied": 0, "bytes_copied": 0, "conflicts": 0,
+    }
+
+
+async def _ensure_artist_id(
+    db: AsyncSession, artist_name: str, dry_run: bool,
+) -> int | None:
+    if not artist_name or not artist_name.strip():
+        return None
+    slug = slugify(artist_name)
+    existing = (await db.execute(
+        select(Artist).where(Artist.slug == slug)
+    )).scalar_one_or_none()
+    if existing is not None:
+        return existing.id
+    if dry_run:
+        return None
+    a = Artist(name=artist_name, slug=slug, is_subscription=False)
+    db.add(a)
+    await db.flush()
+    return a.id
+
+
+async def _resolve_tag_id(
+    db: AsyncSession, tag_name: str, tag_kind: str,
+) -> int | None:
+    try:
+        kind = TagKind(tag_kind)
+    except ValueError:
+        kind = TagKind.general
+    row = (await db.execute(
+        select(Tag.id).where(Tag.name == tag_name, Tag.kind == kind)
+    )).scalar_one_or_none()
+    return row
+
+
+async def _sha_to_image_id(db: AsyncSession, sha: str) -> int | None:
+    return (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one_or_none()
+
+
+async def apply_async(
+    db: AsyncSession,
+    *,
+    images_root: Path | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Apply the manifest. Returns counts + an `unmatched` list of sha256s."""
+    mf_path = manifest_path(images_root)
+    if not mf_path.exists():
+        raise FileNotFoundError(f"no IR tag manifest at {mf_path}")
+
+    manifest = json.loads(mf_path.read_text())
+    counts = _zero_counts()
+    unmatched: list[dict] = []
+
+    # 1. Artist assignments.
+    for entry in manifest.get("image_artist_assignments", []):
+        counts["rows_processed"] += 1
+        img_id = await _sha_to_image_id(db, entry["sha256"])
+        if img_id is None:
+            unmatched.append({"kind": "artist", **entry})
+            counts["rows_skipped"] += 1
+            continue
+        aid = await _ensure_artist_id(db, entry["artist_name"], dry_run)
+        if aid is None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        img = await db.get(ImageRecord, img_id)
+        if img is not None and img.artist_id != aid:
+            img.artist_id = aid
+            counts["rows_inserted"] += 1
+        else:
+            counts["rows_skipped"] += 1
+
+    # 2. Tag associations.
+    for entry in manifest.get("image_tag_associations", []):
+        counts["rows_processed"] += 1
+        img_id = await _sha_to_image_id(db, entry["sha256"])
+        if img_id is None:
+            unmatched.append({"kind": "tag", **entry})
+            counts["rows_skipped"] += 1
+            continue
+        tag_id = await _resolve_tag_id(
+            db, entry["tag_name"], entry.get("tag_kind") or "general",
+        )
+        if tag_id is None:
+            counts["rows_skipped"] += 1
+            continue
+        # Skip if association already exists.
+        already = (await db.execute(
+            select(image_tag.c.image_record_id).where(
+                image_tag.c.image_record_id == img_id,
+                image_tag.c.tag_id == tag_id,
+            )
+        )).first()
+        if already is not None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        await db.execute(image_tag.insert().values(
+            image_record_id=img_id, tag_id=tag_id, source="manual",
+        ))
+        counts["rows_inserted"] += 1
+
+    # 3. Series pages.
+    for entry in manifest.get("series_pages", []):
+        counts["rows_processed"] += 1
+        img_id = await _sha_to_image_id(db, entry["sha256"])
+        if img_id is None:
+            unmatched.append({"kind": "series", **entry})
+            counts["rows_skipped"] += 1
+            continue
+        series_tag_id = await _resolve_tag_id(db, entry["series_tag_name"], "series")
+        if series_tag_id is None:
+            counts["rows_skipped"] += 1
+            continue
+        existing = (await db.execute(
+            select(SeriesPage).where(SeriesPage.image_id == img_id)
+        )).scalar_one_or_none()
+        if existing is not None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        db.add(SeriesPage(
+            series_tag_id=series_tag_id,
+            image_id=img_id,
+            page_number=entry["page_number"],
+        ))
+        counts["rows_inserted"] += 1
+
+    if not dry_run:
+        await db.commit()
+    return {"counts": counts, "unmatched": unmatched}

tests/test_tag_apply.py

@@ -0,0 +1,156 @@
+"""FC-5: tag_apply unit tests."""
+import json
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist, ImageRecord, SeriesPage, Tag, TagKind, image_tag
+from backend.app.services.migrators import tag_apply
+
+pytestmark = pytest.mark.integration
+
+
+async def _seed_image(db, sha: str, *, suffix="x"):
+    img = ImageRecord(
+        path=f"/images/imported/aa/{suffix}.jpg",
+        sha256=sha, size_bytes=10, mime="image/jpeg",
+        width=10, height=10, origin="imported_filesystem",
+    )
+    db.add(img)
+    await db.flush()
+    return img
+
+
+def _write_manifest(tmp_path, *, artist_assignments=None, tag_associations=None, series_pages=None):
+    d = tmp_path / "_migration_state"
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "ir_tag_manifest.json").write_text(json.dumps({
+        "schema_version": 1,
+        "image_artist_assignments": artist_assignments or [],
+        "image_tag_associations": tag_associations or [],
+        "series_pages": series_pages or [],
+    }))
+
+
+@pytest.mark.asyncio
+async def test_artist_assignment_sets_image_artist_id(db, tmp_path):
+    sha = "a" * 64
+    await _seed_image(db, sha)
+    await db.commit()
+
+    _write_manifest(tmp_path, artist_assignments=[
+        {"sha256": sha, "artist_name": "BobAuthor"},
+    ])
+
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert result["counts"]["rows_inserted"] >= 1
+
+    img = (await db.execute(
+        select(ImageRecord).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    artist = (await db.execute(
+        select(Artist).where(Artist.id == img.artist_id)
+    )).scalar_one()
+    assert artist.name == "BobAuthor"
+
+
+@pytest.mark.asyncio
+async def test_tag_association_creates_image_tag(db, tmp_path):
+    sha = "b" * 64
+    await _seed_image(db, sha, suffix="b")
+    # Seed the tag itself (ir_ingest would have created it).
+    tag = Tag(name="blue_eyes", kind=TagKind.general)
+    db.add(tag)
+    await db.commit()
+
+    _write_manifest(tmp_path, tag_associations=[
+        {"sha256": sha, "tag_name": "blue_eyes", "tag_kind": "general", "fandom_name": None},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+
+    img_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    row = (await db.execute(
+        select(image_tag.c.tag_id).where(image_tag.c.image_record_id == img_id)
+    )).all()
+    assert len(row) == 1
+
+
+@pytest.mark.asyncio
+async def test_unmatched_sha_appears_in_unmatched_list(db, tmp_path):
+    _write_manifest(tmp_path, tag_associations=[
+        {"sha256": "z" * 64, "tag_name": "missing_image", "tag_kind": "general", "fandom_name": None},
+    ])
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert any(e["sha256"] == "z" * 64 for e in result["unmatched"])
+
+
+@pytest.mark.asyncio
+async def test_series_page_created(db, tmp_path):
+    sha = "c" * 64
+    await _seed_image(db, sha, suffix="c")
+    series_tag = Tag(name="MySeries", kind=TagKind.series)
+    db.add(series_tag)
+    await db.commit()
+
+    _write_manifest(tmp_path, series_pages=[
+        {"sha256": sha, "series_tag_name": "MySeries", "page_number": 1},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+
+    img_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    sp = (await db.execute(
+        select(SeriesPage).where(SeriesPage.image_id == img_id)
+    )).scalar_one()
+    assert sp.page_number == 1
+
+
+@pytest.mark.asyncio
+async def test_idempotent_on_rerun(db, tmp_path):
+    sha = "d" * 64
+    await _seed_image(db, sha, suffix="d")
+    tag = Tag(name="rerun_tag", kind=TagKind.general)
+    db.add(tag)
+    await db.commit()
+
+    _write_manifest(tmp_path, tag_associations=[
+        {"sha256": sha, "tag_name": "rerun_tag", "tag_kind": "general", "fandom_name": None},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    result2 = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert result2["counts"]["rows_skipped"] >= 1
+
+
+@pytest.mark.asyncio
+async def test_missing_manifest_raises(db, tmp_path):
+    with pytest.raises(FileNotFoundError):
+        await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+
+
+@pytest.mark.asyncio
+async def test_dry_run_makes_no_changes(db, tmp_path):
+    sha = "e" * 64
+    await _seed_image(db, sha, suffix="e")
+    tag = Tag(name="dry_tag", kind=TagKind.general)
+    db.add(tag)
+    await db.commit()
+
+    _write_manifest(tmp_path, tag_associations=[
+        {"sha256": sha, "tag_name": "dry_tag", "tag_kind": "general", "fandom_name": None},
+    ])
+
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=True)
+
+    img_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    rows = (await db.execute(
+        select(image_tag.c.tag_id).where(image_tag.c.image_record_id == img_id)
+    )).all()
+    assert len(rows) == 0