imagerepo/scripts/export_for_fabledcurator.py

#!/usr/bin/env python
"""Export IR's config (tags, artist assignments, tag associations, series pages)
as a single JSON file that FabledCurator's FC-5 ingest can read.

One-shot tool. Excludes:
  - kind='artist' tags (FC stores artist as a row, not a tag — these become
    image_artist_assignments keyed by sha256).
  - kind='post' tags (FC has Post rows; scheduler refills going forward).
  - archive_record, post_metadata, ML predictions/embeddings, suggestion data
    (FC handles all of these via its own mechanisms — see FC-5 design doc).

All foreign keys collapsed to natural keys so the output is portable:
image_record.id -> image_record.hash (sha256); tag.id -> (tag.name, tag.kind).

Usage:
    python scripts/export_for_fabledcurator.py > ir-export.json
"""
from __future__ import annotations

import json
import sys
from datetime import UTC, datetime

# Bootstrap IR's app context so we can use its SQLAlchemy session.
sys.path.insert(0, ".")
from app import create_app, db  # noqa: E402
from app.models import ImageRecord, SeriesPage, Tag, image_tags  # noqa: E402

_SKIP_KINDS = frozenset({"artist", "post"})


def _export_tags():
    """All tags except artist + post kinds, with fandom_name resolved."""
    out = []
    for tag in Tag.query.all():
        if tag.kind in _SKIP_KINDS:
            continue
        fandom_name = None
        if tag.fandom_id:
            f = Tag.query.get(tag.fandom_id)
            if f is not None:
                fandom_name = f.name
        out.append({
            "name": tag.name,
            "kind": tag.kind or "general",
            "fandom_name": fandom_name,
        })
    return out


def _export_image_artist_assignments():
    """For each image, find its artist:Name tag (if any) and emit
    {sha256, artist_name}."""
    out = []
    artist_tag_ids = {
        t.id: t.name for t in Tag.query.filter_by(kind="artist").all()
    }
    if not artist_tag_ids:
        return out
    # Join image_record + image_tags to find which images have artist tags.
    q = (
        db.session.query(ImageRecord.hash, Tag.name)
        .join(image_tags, image_tags.c.image_id == ImageRecord.id)
        .join(Tag, Tag.id == image_tags.c.tag_id)
        .filter(Tag.kind == "artist")
    )
    for sha, artist_name in q.all():
        out.append({"sha256": sha, "artist_name": artist_name})
    return out


def _export_image_tag_associations():
    """For each (image, tag) where tag.kind not in _SKIP_KINDS, emit
    {sha256, tag_name, tag_kind, fandom_name}."""
    out = []
    q = (
        db.session.query(ImageRecord.hash, Tag.name, Tag.kind, Tag.fandom_id)
        .join(image_tags, image_tags.c.image_id == ImageRecord.id)
        .join(Tag, Tag.id == image_tags.c.tag_id)
        .filter(~Tag.kind.in_(_SKIP_KINDS))
    )
    fandom_name_cache: dict[int, str] = {}
    for sha, tag_name, tag_kind, fandom_id in q.all():
        fandom_name = None
        if fandom_id:
            if fandom_id not in fandom_name_cache:
                f = Tag.query.get(fandom_id)
                fandom_name_cache[fandom_id] = f.name if f else None
            fandom_name = fandom_name_cache[fandom_id]
        out.append({
            "sha256": sha,
            "tag_name": tag_name,
            "tag_kind": tag_kind or "general",
            "fandom_name": fandom_name,
        })
    return out


def _export_series_pages():
    """For each SeriesPage, emit {sha256, series_tag_name, page_number}."""
    out = []
    q = (
        db.session.query(ImageRecord.hash, Tag.name, SeriesPage.page_number)
        .join(SeriesPage, SeriesPage.image_id == ImageRecord.id)
        .join(Tag, Tag.id == SeriesPage.series_tag_id)
    )
    for sha, series_tag_name, page_number in q.all():
        out.append({
            "sha256": sha,
            "series_tag_name": series_tag_name,
            "page_number": page_number,
        })
    return out


def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output",
        default=None,
        help="Write JSON to this path instead of stdout (use this when "
             "running inside a Swarm service that writes to a bind-mount).",
    )
    args = parser.parse_args()

    app = create_app()
    with app.app_context():
        export = {
            "source_app": "imagerepo",
            "schema_version": 1,
            "exported_at": datetime.now(UTC).isoformat(),
            "tags": _export_tags(),
            "image_artist_assignments": _export_image_artist_assignments(),
            "image_tag_associations": _export_image_tag_associations(),
            "series_pages": _export_series_pages(),
        }

    if args.output:
        from pathlib import Path
        out_path = Path(args.output)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(json.dumps(export, indent=2) + "\n")
        print(f"wrote {out_path} ({len(export['tags'])} tags, "
              f"{len(export['image_tag_associations'])} associations)",
              file=sys.stderr)
    else:
        json.dump(export, sys.stdout, indent=2)
        sys.stdout.write("\n")


if __name__ == "__main__":
    main()