fc5: gs_ingest — parsed-JSON ingest, Artist/Source/Credential, re-encrypts with FC key

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

backend/app/services/migrators/gs_ingest.py

@@ -0,0 +1,126 @@
+"""GallerySubscriber export → FabledCurator ingest.
+
+Reads a parsed gallerysubscriber-export-v1.json dict (no DB connection
+to GS). Creates Artist (from subscriptions) + Source (nested under each
+subscription) + Credential (re-encrypted with FC's key). Idempotent on
+natural keys: Artist.slug, (artist_id, platform, url), Credential.platform.
+
+Credentials arrive plaintext in the export — GS's export script
+decrypts using GS's Fernet key in GS's own process. FC re-encrypts
+with FC's CredentialCrypto.
+"""
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ...models import Artist, Credential, Source
+from ...utils.slug import slugify
+from ..credential_crypto import CredentialCrypto
+
+
+def _zero_counts() -> dict:
+    return {
+        "rows_processed": 0, "rows_inserted": 0, "rows_skipped": 0,
+        "files_copied": 0, "bytes_copied": 0, "conflicts": 0,
+    }
+
+
+async def migrate_async(
+    db: AsyncSession,
+    *,
+    data: dict,
+    fc_crypto: CredentialCrypto | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """Ingest a parsed gallerysubscriber-export-v1.json dict."""
+    if data.get("source_app") != "gallerysubscriber":
+        raise ValueError("export source_app must be 'gallerysubscriber'")
+    if data.get("schema_version") != 1:
+        raise ValueError(f"unsupported schema_version: {data.get('schema_version')}")
+
+    counts = _zero_counts()
+
+    # Phase 1: subscriptions → Artist; nested sources within each.
+    for sub in data.get("subscriptions", []):
+        counts["rows_processed"] += 1
+        slug = slugify(sub["name"])
+        artist = (await db.execute(
+            select(Artist).where(Artist.slug == slug)
+        )).scalar_one_or_none()
+        if artist is None:
+            if dry_run:
+                counts["rows_inserted"] += 1
+                # Continue to nested sources, but they can't link without an artist row.
+                continue
+            notes = json.dumps(sub.get("metadata"), indent=2) if sub.get("metadata") else None
+            artist = Artist(
+                name=sub["name"], slug=slug,
+                is_subscription=True,
+                auto_check=bool(sub.get("enabled", True)),
+                notes=notes,
+            )
+            db.add(artist)
+            await db.flush()
+            counts["rows_inserted"] += 1
+        else:
+            counts["rows_skipped"] += 1
+
+        # Nested sources under this subscription.
+        for src in sub.get("sources", []):
+            counts["rows_processed"] += 1
+            existing = (await db.execute(
+                select(Source).where(
+                    Source.artist_id == artist.id,
+                    Source.platform == src["platform"],
+                    Source.url == src["url"],
+                )
+            )).scalar_one_or_none()
+            if existing is not None:
+                counts["rows_skipped"] += 1
+                continue
+            if dry_run:
+                counts["rows_inserted"] += 1
+                continue
+            db.add(Source(
+                artist_id=artist.id,
+                platform=src["platform"],
+                url=src["url"],
+                enabled=bool(src.get("enabled", True)),
+                check_interval_override=src.get("check_interval"),
+                config_overrides=src.get("metadata") or {},
+            ))
+            counts["rows_inserted"] += 1
+
+    # Phase 2: credentials.
+    for cred in data.get("credentials", []):
+        counts["rows_processed"] += 1
+        existing = (await db.execute(
+            select(Credential).where(Credential.platform == cred["platform"])
+        )).scalar_one_or_none()
+        if existing is not None:
+            counts["rows_skipped"] += 1
+            continue
+        if dry_run:
+            counts["rows_inserted"] += 1
+            continue
+        if fc_crypto is None:
+            # Without a crypto helper we can't encrypt — skip rather than
+            # store plaintext.
+            counts["rows_skipped"] += 1
+            counts["conflicts"] += 1
+            continue
+        encrypted = fc_crypto.encrypt(cred["plaintext"])
+        db.add(Credential(
+            platform=cred["platform"],
+            kind=cred.get("credential_type") or "cookies",
+            encrypted_blob=encrypted,
+            expires_at=cred.get("expires_at"),
+        ))
+        counts["rows_inserted"] += 1
+
+    if not dry_run:
+        await db.commit()
+    return counts

tests/test_gs_ingest.py

@@ -0,0 +1,122 @@
+"""FC-5: gs_ingest unit tests."""
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist, Credential, Source
+from backend.app.services.credential_crypto import CredentialCrypto
+from backend.app.services.migrators import gs_ingest
+
+pytestmark = pytest.mark.integration
+
+
+def _gs_export(subscriptions=None, credentials=None) -> dict:
+    return {
+        "source_app": "gallerysubscriber",
+        "schema_version": 1,
+        "exported_at": datetime.now(UTC).isoformat(),
+        "subscriptions": subscriptions or [],
+        "credentials": credentials or [],
+    }
+
+
+@pytest.mark.asyncio
+async def test_subscription_creates_artist_with_slug(db):
+    data = _gs_export(subscriptions=[
+        {"name": "Alice Author", "enabled": True, "metadata": {}, "sources": []},
+    ])
+    counts = await gs_ingest.migrate_async(db, data=data, dry_run=False)
+    assert counts["rows_inserted"] >= 1
+
+    artist = (await db.execute(
+        select(Artist).where(Artist.name == "Alice Author")
+    )).scalar_one()
+    assert artist.slug == "alice-author"
+    assert artist.is_subscription is True
+
+
+@pytest.mark.asyncio
+async def test_nested_source_attached_to_artist(db):
+    data = _gs_export(subscriptions=[{
+        "name": "bob-author",
+        "enabled": True,
+        "metadata": {},
+        "sources": [{
+            "platform": "patreon",
+            "url": "https://patreon.com/bob",
+            "enabled": True,
+            "check_interval": 28800,
+            "metadata": {"k": "v"},
+        }],
+    }])
+    await gs_ingest.migrate_async(db, data=data, dry_run=False)
+
+    src = (await db.execute(
+        select(Source).where(Source.url == "https://patreon.com/bob")
+    )).scalar_one()
+    artist = (await db.execute(
+        select(Artist).where(Artist.id == src.artist_id)
+    )).scalar_one()
+    assert artist.name == "bob-author"
+    assert src.platform == "patreon"
+    assert src.check_interval_override == 28800
+    assert src.config_overrides == {"k": "v"}
+
+
+@pytest.mark.asyncio
+async def test_credential_re_encrypted_with_fc_key(db, tmp_path):
+    fc_key_path = tmp_path / "credential_key.b64"
+    fc_crypto = CredentialCrypto(fc_key_path)
+
+    data = _gs_export(credentials=[{
+        "platform": "patreon",
+        "credential_type": "cookies",
+        "plaintext": "session_cookie=abcdef",
+        "expires_at": None,
+    }])
+    await gs_ingest.migrate_async(db, data=data, fc_crypto=fc_crypto, dry_run=False)
+
+    cred = (await db.execute(
+        select(Credential).where(Credential.platform == "patreon")
+    )).scalar_one()
+    assert fc_crypto.decrypt(cred.encrypted_blob) == "session_cookie=abcdef"
+    assert cred.kind == "cookies"
+
+
+@pytest.mark.asyncio
+async def test_dry_run_makes_no_changes(db):
+    data = _gs_export(subscriptions=[
+        {"name": "dave", "enabled": True, "metadata": {}, "sources": []},
+    ])
+    counts = await gs_ingest.migrate_async(db, data=data, dry_run=True)
+    assert counts["rows_processed"] >= 1
+
+    after = (await db.execute(
+        select(Artist).where(Artist.name == "dave")
+    )).scalar_one_or_none()
+    assert after is None
+
+
+@pytest.mark.asyncio
+async def test_idempotent_on_rerun(db):
+    data = _gs_export(subscriptions=[
+        {"name": "eve", "enabled": True, "metadata": {}, "sources": []},
+    ])
+    await gs_ingest.migrate_async(db, data=data, dry_run=False)
+    counts2 = await gs_ingest.migrate_async(db, data=data, dry_run=False)
+    assert counts2["rows_skipped"] >= 1
+
+
+@pytest.mark.asyncio
+async def test_rejects_wrong_source_app(db):
+    bad = {"source_app": "wrong", "schema_version": 1, "subscriptions": [], "credentials": []}
+    with pytest.raises(ValueError):
+        await gs_ingest.migrate_async(db, data=bad, dry_run=False)
+
+
+@pytest.mark.asyncio
+async def test_rejects_unsupported_schema_version(db):
+    bad = {"source_app": "gallerysubscriber", "schema_version": 99, "subscriptions": [], "credentials": []}
+    with pytest.raises(ValueError):
+        await gs_ingest.migrate_async(db, data=bad, dry_run=False)