fc5: export script for FabledCurator migration — subscriptions/sources + decrypted credentials

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

scripts/export_for_fabledcurator.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+"""Export GS's config (subscriptions, sources, credentials) as a single JSON
+file that FabledCurator's FC-5 ingest can read.
+
+One-shot tool. Excludes:
+  - downloads (history not migrated; FC's scheduler refills going forward).
+  - content_items (same).
+
+Credentials are decrypted in this script's process using GS's secret_key
+(passed via --secret-key or the SECRET_KEY env var); FC re-encrypts on
+ingest with its own Fernet key. The output JSON contains plaintext
+credentials — delete it after the FC ingest completes.
+
+Usage:
+    python scripts/export_for_fabledcurator.py --secret-key "$SECRET_KEY" > gs-export.json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from datetime import UTC, datetime
+
+# Bootstrap GS's app context.
+sys.path.insert(0, "backend")
+from app import create_app  # noqa: E402
+from app.database import get_session  # noqa: E402
+from app.models.credential import Credential  # noqa: E402
+from app.models.source import Source  # noqa: E402
+from app.models.subscription import Subscription  # noqa: E402
+from app.utils.encryption import decrypt_data  # noqa: E402
+
+
+def _export_subscriptions(session) -> list[dict]:
+    """Subscriptions with nested sources."""
+    out: list[dict] = []
+    for sub in session.query(Subscription).all():
+        sources_out: list[dict] = []
+        for src in sub.sources or []:
+            sources_out.append({
+                "platform": src.platform,
+                "url": src.url,
+                "enabled": bool(src.enabled),
+                "check_interval": int(src.check_interval or 28800),
+                "metadata": src.metadata_ or {},
+            })
+        out.append({
+            "name": sub.name,
+            "enabled": bool(sub.enabled),
+            "metadata": sub.metadata_ or {},
+            "sources": sources_out,
+        })
+    return out
+
+
+def _export_credentials(session, secret_key: str) -> list[dict]:
+    """Credentials decrypted to plaintext (FC re-encrypts on ingest)."""
+    out: list[dict] = []
+    for cred in session.query(Credential).all():
+        try:
+            plaintext = decrypt_data(cred.data, secret_key)
+        except Exception as exc:
+            print(
+                f"WARNING: failed to decrypt credential for "
+                f"platform={cred.platform!r}: {exc}",
+                file=sys.stderr,
+            )
+            continue
+        out.append({
+            "platform": cred.platform,
+            "credential_type": cred.credential_type,
+            "plaintext": plaintext,
+            "expires_at": (
+                cred.expires_at.isoformat() if cred.expires_at else None
+            ),
+        })
+    return out
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--secret-key",
+        default=os.environ.get("SECRET_KEY"),
+        help="GS SECRET_KEY (defaults to $SECRET_KEY env var).",
+    )
+    args = parser.parse_args()
+    if not args.secret_key:
+        print(
+            "ERROR: --secret-key required (or set SECRET_KEY env var)",
+            file=sys.stderr,
+        )
+        sys.exit(2)
+
+    app = create_app()
+    with app.app_context():
+        with get_session() as session:
+            export = {
+                "source_app": "gallerysubscriber",
+                "schema_version": 1,
+                "exported_at": datetime.now(UTC).isoformat(),
+                "subscriptions": _export_subscriptions(session),
+                "credentials": _export_credentials(session, args.secret_key),
+            }
+    json.dump(export, sys.stdout, indent=2)
+    sys.stdout.write("\n")
+
+
+if __name__ == "__main__":
+    main()