From 8649a131189a95bd8ac9875b82b2cbc11204fc1a Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 29 May 2026 14:38:59 -0400 Subject: [PATCH] refactor(I5): remove one-and-done GS/IR migration tooling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The GS/IR migration cutover is complete, so the runbook tooling is dead weight. Removed: - services/migrators/ (gs_ingest, ir_ingest, tag_apply, ml_queue, verify, cleanup), tasks/migration.py, api/migrate.py (+ blueprint registration) - MigrationRun model; alembic 0027 drops the migration_run table - frontend LegacyMigrationCard + migration store (+ MaintenancePanel ref) - celery include + task route + celery_signals queue mapping for migration.* - the 1 GB MAX_CONTENT_LENGTH / MAX_FORM_MEMORY override (added solely for the ir_ingest upload) - migration-surface tests (test_api_migrate, test_migration_verify, test_ir_ingest, test_gs_ingest, test_tag_apply) Kept: the alembic schema-migration tests (test_migration_00XX — unrelated) and cleanup_service.py (the permanent artist-cascade/unlink home). Co-Authored-By: Claude Opus 4.7 (1M context) --- alembic/versions/0027_drop_migration_run.py | 50 +++ backend/app/__init__.py | 6 - backend/app/api/__init__.py | 2 - backend/app/api/migrate.py | 107 ----- backend/app/celery_app.py | 2 - backend/app/celery_signals.py | 5 +- backend/app/models/__init__.py | 2 - backend/app/models/migration_run.py | 37 -- backend/app/services/cleanup_service.py | 5 +- backend/app/services/migrators/__init__.py | 10 - backend/app/services/migrators/cleanup.py | 182 -------- backend/app/services/migrators/gs_ingest.py | 126 ------ backend/app/services/migrators/ir_ingest.py | 146 ------- backend/app/services/migrators/ml_queue.py | 21 - backend/app/services/migrators/tag_apply.py | 368 ---------------- backend/app/services/migrators/verify.py | 80 ---- backend/app/tasks/migration.py | 169 -------- .../settings/LegacyMigrationCard.vue | 245 ----------- .../components/settings/MaintenancePanel.vue | 5 +- frontend/src/stores/migration.js | 83 ---- tests/test_api_migrate.py | 98 ----- tests/test_gs_ingest.py | 122 ------ tests/test_ir_ingest.py | 129 ------ tests/test_migration_verify.py | 72 ---- tests/test_tag_apply.py | 395 ------------------ 25 files changed, 55 insertions(+), 2412 deletions(-) create mode 100644 alembic/versions/0027_drop_migration_run.py delete mode 100644 backend/app/api/migrate.py delete mode 100644 backend/app/models/migration_run.py delete mode 100644 backend/app/services/migrators/__init__.py delete mode 100644 backend/app/services/migrators/cleanup.py delete mode 100644 backend/app/services/migrators/gs_ingest.py delete mode 100644 backend/app/services/migrators/ir_ingest.py delete mode 100644 backend/app/services/migrators/ml_queue.py delete mode 100644 backend/app/services/migrators/tag_apply.py delete mode 100644 backend/app/services/migrators/verify.py delete mode 100644 backend/app/tasks/migration.py delete mode 100644 frontend/src/components/settings/LegacyMigrationCard.vue delete mode 100644 frontend/src/stores/migration.js delete mode 100644 tests/test_api_migrate.py delete mode 100644 tests/test_gs_ingest.py delete mode 100644 tests/test_ir_ingest.py delete mode 100644 tests/test_migration_verify.py delete mode 100644 tests/test_tag_apply.py diff --git a/alembic/versions/0027_drop_migration_run.py b/alembic/versions/0027_drop_migration_run.py new file mode 100644 index 0000000..454481b --- /dev/null +++ b/alembic/versions/0027_drop_migration_run.py @@ -0,0 +1,50 @@ +"""drop migration_run — one-and-done GS/IR migration tooling removed + +Revision ID: 0027 +Revises: 0026 +Create Date: 2026-05-29 + +The GS/IR migration tooling (services/migrators, /api/migrate, the +run_migration task, LegacyMigrationCard, and the MigrationRun model) was +removed after the migration cutover completed. This drops its now-orphaned +run-log table. Downgrade recreates the table (mirrors the old model) so the +migration is reversible. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +revision: str = "0027" +down_revision: Union[str, None] = "0026" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.drop_table("migration_run") + + +def downgrade() -> None: + op.create_table( + "migration_run", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("kind", sa.String(length=32), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("dry_run", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column( + "started_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now(), + ), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "counts", JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb"), + ), + sa.Column("error", sa.Text(), nullable=True), + sa.Column( + "metadata", JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb"), + ), + ) + op.create_index("ix_migration_run_kind", "migration_run", ["kind"]) + op.create_index("ix_migration_run_status", "migration_run", ["status"]) diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 70efbda..75ace0d 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -33,12 +33,6 @@ def create_app() -> Quart: app = Quart(__name__) app.secret_key = cfg.secret_key - # FC-5: legacy IR ingest JSON can run to tens of MB (hundreds of - # thousands of image_tag_associations). Werkzeug's default form - # memory cap is 500KB; raise both ceilings so the multipart upload - # for /api/migrate/ir_ingest doesn't 413. - app.config["MAX_CONTENT_LENGTH"] = 1024 * 1024 * 1024 # 1 GB - app.config["MAX_FORM_MEMORY_SIZE"] = 1024 * 1024 * 1024 # 1 GB for bp in all_blueprints(): app.register_blueprint(bp) diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index b8dc7c4..f97b72a 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -26,7 +26,6 @@ def all_blueprints() -> list[Blueprint]: from .extension import extension_bp from .gallery import gallery_bp from .import_admin import import_admin_bp - from .migrate import migrate_bp from .ml_admin import ml_admin_bp from .platforms import platforms_bp from .posts import posts_bp @@ -54,7 +53,6 @@ def all_blueprints() -> list[Blueprint]: admin_bp, cleanup_bp, import_admin_bp, - migrate_bp, suggestions_bp, allowlist_bp, aliases_bp, diff --git a/backend/app/api/migrate.py b/backend/app/api/migrate.py deleted file mode 100644 index d48e9f4..0000000 --- a/backend/app/api/migrate.py +++ /dev/null @@ -1,107 +0,0 @@ -"""FC-5: /api/migrate — trigger and poll migration runs. - -Ingest kinds (gs_ingest, ir_ingest) accept multipart/form-data with an -`export_file` field. All other kinds accept JSON. Backup + rollback -were retired in FC-3h (2026-05-24); use /api/system/backup/* instead. -""" - -import json - -from quart import Blueprint, jsonify, request -from sqlalchemy import select - -from ..extensions import get_session -from ..models import MigrationRun -from ..tasks.migration import run_migration -from ._responses import error_response as _bad - -migrate_bp = Blueprint("migrate", __name__, url_prefix="/api/migrate") - -# 'backup' + 'rollback' retired 2026-05-24 (FC-3h); see /api/system/backup/*. -_VALID_KINDS = frozenset({ - "gs_ingest", "ir_ingest", "tag_apply", - "ml_queue", "verify", "cleanup", -}) -_INGEST_KINDS = frozenset({"gs_ingest", "ir_ingest"}) - - -def _run_to_dict(run: MigrationRun) -> dict: - return { - "id": run.id, - "kind": run.kind, - "status": run.status, - "dry_run": run.dry_run, - "started_at": run.started_at.isoformat(), - "finished_at": run.finished_at.isoformat() if run.finished_at else None, - "counts": run.counts or {}, - "error": run.error, - "metadata": run.metadata_ or {}, - } - - -@migrate_bp.route("/", methods=["POST"]) -async def create_run(kind: str): - if kind not in _VALID_KINDS: - return _bad("unknown_kind", detail=f"kind must be one of {sorted(_VALID_KINDS)}") - - # Ingest kinds accept multipart/form-data; everything else takes JSON. - if kind in _INGEST_KINDS: - form = await request.form - files = await request.files - if "export_file" not in files: - return _bad("missing_export_file", detail="multipart export_file required") - export_file = files["export_file"] - try: - raw = export_file.read() - data = json.loads(raw.decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - return _bad("invalid_export_file", detail=str(exc)) - dry_run = str(form.get("dry_run", "false")).lower() in ("true", "1", "yes") - params: dict = {"data": data, "dry_run": dry_run} - else: - body = await request.get_json() - if body is None: - body = {} - if not isinstance(body, dict): - return _bad("invalid_body") - dry_run = bool(body.get("dry_run", False)) - params = dict(body) - - async with get_session() as session: - run = MigrationRun(kind=kind, status="pending", dry_run=dry_run) - session.add(run) - await session.commit() - await session.refresh(run) - run_id = run.id - - run_migration.delay(run_id, kind, params) - return jsonify({"run_id": run_id, "status": "pending"}), 202 - - -@migrate_bp.route("/runs/", methods=["GET"]) -async def get_run(run_id: int): - async with get_session() as session: - run = (await session.execute( - select(MigrationRun).where(MigrationRun.id == run_id) - )).scalar_one_or_none() - if run is None: - return _bad("not_found", status=404) - return jsonify(_run_to_dict(run)) - - -@migrate_bp.route("/runs", methods=["GET"]) -async def list_runs(): - try: - limit = int(request.args.get("limit", "10")) - except ValueError: - return _bad("invalid_limit") - if limit < 1 or limit > 100: - return _bad("invalid_limit") - - async with get_session() as session: - rows = (await session.execute( - select(MigrationRun) - .order_by(MigrationRun.id.desc()) - .limit(limit) - )).scalars().all() - return jsonify([_run_to_dict(r) for r in rows]) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 3068ca2..42f72ec 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -28,7 +28,6 @@ def make_celery() -> Celery: "backend.app.tasks.import_file", "backend.app.tasks.thumbnail", "backend.app.tasks.maintenance", - "backend.app.tasks.migration", "backend.app.tasks.ml", "backend.app.tasks.download", "backend.app.tasks.backup", @@ -45,7 +44,6 @@ def make_celery() -> Celery: "backend.app.tasks.download.*": {"queue": "download"}, "backend.app.tasks.scan.*": {"queue": "scan"}, "backend.app.tasks.maintenance.*": {"queue": "maintenance"}, - "backend.app.tasks.migration.*": {"queue": "maintenance"}, "backend.app.tasks.backup.*": {"queue": "maintenance"}, "backend.app.tasks.admin.*": {"queue": "maintenance"}, "backend.app.tasks.library_audit.*": {"queue": "maintenance"}, diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index 307d5ea..37bd482 100644 --- a/backend/app/celery_signals.py +++ b/backend/app/celery_signals.py @@ -66,10 +66,7 @@ def _queue_for(task) -> str: return "download" if name.startswith("backend.app.tasks.scan."): return "scan" - if name.startswith(( - "backend.app.tasks.maintenance.", - "backend.app.tasks.migration.", - )): + if name.startswith("backend.app.tasks.maintenance."): return "maintenance" return "default" diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 00a7c80..eec9da0 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -12,7 +12,6 @@ from .import_batch import ImportBatch from .import_settings import ImportSettings from .import_task import ImportTask from .library_audit_run import LibraryAuditRun -from .migration_run import MigrationRun from .ml_settings import MLSettings from .post import Post from .post_attachment import PostAttachment @@ -46,7 +45,6 @@ __all__ = [ "ImportSettings", "LibraryAuditRun", "MLSettings", - "MigrationRun", "TagAlias", "TagAllowlist", "TagReferenceEmbedding", diff --git a/backend/app/models/migration_run.py b/backend/app/models/migration_run.py deleted file mode 100644 index bdb5fed..0000000 --- a/backend/app/models/migration_run.py +++ /dev/null @@ -1,37 +0,0 @@ -"""MigrationRun — tracks each FC-5 migration invocation (backup/gs/ir/etc). - -kind/status are String(32) not Postgres ENUM so adding kinds later -doesn't need a schema migration. The API layer validates values. -""" - -from datetime import datetime - -import sqlalchemy as sa -from sqlalchemy import Boolean, DateTime, Integer, String, Text, func -from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.orm import Mapped, mapped_column - -from .base import Base - - -class MigrationRun(Base): - __tablename__ = "migration_run" - - id: Mapped[int] = mapped_column(Integer, primary_key=True) - kind: Mapped[str] = mapped_column(String(32), nullable=False, index=True) - status: Mapped[str] = mapped_column(String(32), nullable=False, index=True) - dry_run: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) - started_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, server_default=func.now(), - ) - finished_at: Mapped[datetime | None] = mapped_column( - DateTime(timezone=True), nullable=True, - ) - counts: Mapped[dict] = mapped_column( - JSONB, nullable=False, default=dict, server_default=sa.text("'{}'::jsonb"), - ) - error: Mapped[str | None] = mapped_column(Text, nullable=True) - metadata_: Mapped[dict] = mapped_column( - "metadata", JSONB, nullable=False, default=dict, - server_default=sa.text("'{}'::jsonb"), - ) diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py index dba9323..0e817c4 100644 --- a/backend/app/services/cleanup_service.py +++ b/backend/app/services/cleanup_service.py @@ -6,9 +6,8 @@ HTTP handlers (small ops) and from Celery tasks in backend.app.tasks.admin (long ops). This module is the PERMANENT home of artist-cascade + image-unlink -logic. The legacy copy at backend/app/services/migrators/cleanup.py -stays in place until FC-3j; FC-3j will replace its body with thin -re-exports from this module and then delete the wrapper. +logic. (The legacy migrators/cleanup.py copy was removed with the rest of +the one-and-done GS/IR migration tooling.) """ from __future__ import annotations diff --git a/backend/app/services/migrators/__init__.py b/backend/app/services/migrators/__init__.py deleted file mode 100644 index b7c0908..0000000 --- a/backend/app/services/migrators/__init__.py +++ /dev/null @@ -1,10 +0,0 @@ -"""FC-5 migration tooling. - -One module per concern (gs/ir/overlap/ml_queue/verify/cleanup). -Each migrator returns a counts dict; the run_migration task wires -that dict into MigrationRun.counts so the UI polling shows progress. - -backup + rollback were retired in FC-3h (2026-05-24); first-class -backup lives at backend/app/services/backup_service.py and exposes -its own /api/system/backup/* surface. -""" diff --git a/backend/app/services/migrators/cleanup.py b/backend/app/services/migrators/cleanup.py deleted file mode 100644 index 9f8da2c..0000000 --- a/backend/app/services/migrators/cleanup.py +++ /dev/null @@ -1,182 +0,0 @@ -"""Targeted cleanup migrator: delete every image attributed to one Artist. - -Built for the IR-migration rescue case where the filesystem scan derived -a bogus 'imagerepo' artist from a mismatched bind-mount layout. Every -image attributed to that artist (40k+ rows) needs to be removed — DB -rows, original files under `/images//...`, and thumbnails under -`/images/thumbs/...` — before the operator remounts and re-scans. - -CASCADE handles image_tag, image_provenance, series_page, and -tag_suggestion_rejection child rows; import_task.result_image_id is -SET NULL by FK. We also delete ImportTask rows whose source_path starts -with the (still-existing) IR scan prefix so the next scan isn't fooled -by them. -""" -from __future__ import annotations - -import logging -from pathlib import Path - -from sqlalchemy import delete, func, select -from sqlalchemy.ext.asyncio import AsyncSession - -from ...models import Artist, ImageRecord, ImportBatch, ImportTask - -log = logging.getLogger(__name__) - -_BATCH_SIZE = 500 - - -def _zero_counts() -> dict: - return { - "rows_processed": 0, "rows_inserted": 0, "rows_skipped": 0, - "files_copied": 0, "bytes_copied": 0, "conflicts": 0, - } - - -def _thumb_path(images_root: Path, sha256_hex: str) -> tuple[Path, Path]: - """Return both possible thumbnail paths (.jpg and .png). We try both - because the extension is chosen at generate-time based on the source - image's mode (alpha → .png, otherwise → .jpg).""" - bucket = sha256_hex[:3] - base = images_root / "thumbs" / bucket / sha256_hex - return base.with_suffix(".jpg"), base.with_suffix(".png") - - -def _delete_file(path: Path) -> bool: - """Best-effort unlink; True if the file was actually removed.""" - try: - path.unlink(missing_ok=True) - return True - except OSError as exc: - log.warning("cleanup: failed to unlink %s: %s", path, exc) - return False - - -async def cleanup_artist_async( - db: AsyncSession, - *, - slug: str, - images_root: Path | None = None, - dry_run: bool = False, - source_path_prefix: str | None = None, -) -> dict: - """Delete every image attributed to the Artist with this slug, - along with the artist row itself and any associated import tasks. - - Args: - slug: artist.slug to target (e.g. 'imagerepo'). - images_root: defaults to /images. - dry_run: skip filesystem + DB writes; still walk rows for counts. - source_path_prefix: if set, ImportTask rows whose source_path - starts with this string are deleted too (use the IR scan - mount prefix, e.g. '/import/imagerepo'). - """ - root = images_root if images_root is not None else Path("/images") - - artist = (await db.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one_or_none() - if artist is None: - raise ValueError(f"no Artist with slug={slug!r}") - - artist_id = artist.id - artist_name = artist.name - - total_images = (await db.execute( - select(func.count(ImageRecord.id)).where(ImageRecord.artist_id == artist_id) - )).scalar_one() - - counts = _zero_counts() - files_deleted = 0 - thumbs_deleted = 0 - images_deleted = 0 - - # Batched delete loop. CASCADE handles image_tag, image_provenance, - # series_page, tag_suggestion_rejection. import_task.result_image_id - # is SET NULL by FK. - while True: - rows = (await db.execute( - select(ImageRecord.id, ImageRecord.path, ImageRecord.sha256) - .where(ImageRecord.artist_id == artist_id) - .limit(_BATCH_SIZE) - )).all() - if not rows: - break - - ids = [r.id for r in rows] - counts["rows_processed"] += len(ids) - - if not dry_run: - for r in rows: - if r.path: - if _delete_file(Path(r.path)): - files_deleted += 1 - if r.sha256: - jpg, png = _thumb_path(root, r.sha256) - if _delete_file(jpg): - thumbs_deleted += 1 - if _delete_file(png): - thumbs_deleted += 1 - - await db.execute( - delete(ImageRecord).where(ImageRecord.id.in_(ids)) - ) - await db.commit() - - images_deleted += len(ids) - - if dry_run: - # Nothing was actually deleted from the DB; bail after one - # pass so we don't loop forever. - break - - import_tasks_deleted = 0 - if source_path_prefix and not dry_run: - # Delete ImportTask rows whose source_path is under the bad mount - # prefix. These are mostly orphaned now (result_image_id was set - # NULL by CASCADE) but their presence still blocks the - # idempotency check in scan_directory if the operator remounts - # the same prefix. - like_pattern = source_path_prefix.rstrip("/") + "/%" - result = await db.execute( - delete(ImportTask).where(ImportTask.source_path.like(like_pattern)) - ) - import_tasks_deleted = result.rowcount or 0 - await db.commit() - - # Sweep ImportBatch rows that are now empty. - empty_batches_deleted = 0 - if not dry_run: - empty_batch_ids = (await db.execute( - select(ImportBatch.id).where( - ~select(ImportTask.id) - .where(ImportTask.batch_id == ImportBatch.id) - .exists() - ) - )).scalars().all() - if empty_batch_ids: - result = await db.execute( - delete(ImportBatch).where(ImportBatch.id.in_(empty_batch_ids)) - ) - empty_batches_deleted = result.rowcount or 0 - await db.commit() - - # Finally, the artist row. - if not dry_run: - await db.execute(delete(Artist).where(Artist.id == artist_id)) - await db.commit() - - return { - "counts": counts, - "artist": {"id": artist_id, "name": artist_name, "slug": slug}, - "summary": { - "images_targeted": total_images, - "images_deleted": images_deleted, - "files_deleted": files_deleted, - "thumbs_deleted": thumbs_deleted, - "import_tasks_deleted": import_tasks_deleted, - "empty_batches_deleted": empty_batches_deleted, - "dry_run": dry_run, - }, - } diff --git a/backend/app/services/migrators/gs_ingest.py b/backend/app/services/migrators/gs_ingest.py deleted file mode 100644 index 18b832a..0000000 --- a/backend/app/services/migrators/gs_ingest.py +++ /dev/null @@ -1,126 +0,0 @@ -"""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"], - credential_type=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 diff --git a/backend/app/services/migrators/ir_ingest.py b/backend/app/services/migrators/ir_ingest.py deleted file mode 100644 index b501583..0000000 --- a/backend/app/services/migrators/ir_ingest.py +++ /dev/null @@ -1,146 +0,0 @@ -"""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") not in (1, 2): - 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. - # schema_version 2 (added 2026-05-24) carries `image_posts` for - # Post + Source + ImageProvenance restore; schema 1 manifests - # without it stay valid (tag_apply treats the missing field as []). - manifest = { - "schema_version": data.get("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", []), - "image_posts": data.get("image_posts", []), - } - counts["rows_processed"] += len(manifest["image_artist_assignments"]) - counts["rows_processed"] += len(manifest["image_tag_associations"]) - counts["rows_processed"] += len(manifest["series_pages"]) - counts["rows_processed"] += len(manifest["image_posts"]) - if not dry_run: - manifest_path(images_root).write_text(json.dumps(manifest, indent=2)) - - return counts diff --git a/backend/app/services/migrators/ml_queue.py b/backend/app/services/migrators/ml_queue.py deleted file mode 100644 index 6d9e92f..0000000 --- a/backend/app/services/migrators/ml_queue.py +++ /dev/null @@ -1,21 +0,0 @@ -"""Queue every migrated image_record with no embedding for ML re-processing.""" -from __future__ import annotations - -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from ...models import ImageRecord - - -async def queue_all_unprocessed_async(db: AsyncSession) -> int: - """Find every ImageRecord with siglip_embedding IS NULL, fire - tag_and_embed.delay(id) for each. Returns count queued. - """ - from ...tasks.ml import tag_and_embed - - rows = (await db.execute( - select(ImageRecord.id).where(ImageRecord.siglip_embedding.is_(None)) - )).scalars().all() - for image_id in rows: - tag_and_embed.delay(image_id) - return len(rows) diff --git a/backend/app/services/migrators/tag_apply.py b/backend/app/services/migrators/tag_apply.py deleted file mode 100644 index 3f15c88..0000000 --- a/backend/app/services/migrators/tag_apply.py +++ /dev/null @@ -1,368 +0,0 @@ -"""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). -- image_posts (schema v2) → Source + Post + ImageProvenance restore. - -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 datetime import datetime -from pathlib import Path - -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from ...models import ( - Artist, - ImageProvenance, - ImageRecord, - Post, - SeriesPage, - Source, - Tag, - TagKind, - image_tag, -) -from ...utils.slug import slugify -from .ir_ingest import manifest_path - -# Per-platform artist-profile URL — used as Source.url when restoring -# IR PostMetadata into FC. Must cover every platform that -# backend/app/services/extension_service.py:_PLATFORM_PATTERNS -# recognizes; an entry missing here silently drops ALL PostMetadata for -# that platform during phase 4 (operator hit this 2026-05-25: -# DeviantArt + Pixiv posts in the IR migration produced empty -# ImageProvenance because they fell through this table). -# -# Pixiv caveat: the real profile URL takes a numeric user_id -# (https://www.pixiv.net/users/12345), but IR's PostMetadata.artist -# stores the display name not the id. We use the slugified name here -# so we preserve the artist→post→image linkage; the resulting Source.url -# won't resolve in a browser and the operator may want to manually fix -# it via Settings → Subscriptions once the migration lands. -_PLATFORM_PROFILE_URL = { - "patreon": "https://www.patreon.com/{slug}", - "subscribestar": "https://www.subscribestar.com/{slug}", - "hentaifoundry": "https://www.hentai-foundry.com/user/{slug}", - "deviantart": "https://www.deviantart.com/{slug}", - "pixiv": "https://www.pixiv.net/users/{slug}", -} - - -def _profile_url(platform: str, artist_slug: str) -> str | None: - fmt = _PLATFORM_PROFILE_URL.get(platform) - return fmt.format(slug=artist_slug) if fmt else None - - -async def _find_or_create_source( - db: AsyncSession, *, artist_id: int, platform: str, url: str, dry_run: bool, -) -> int | None: - existing = (await db.execute( - select(Source.id).where( - Source.artist_id == artist_id, - Source.platform == platform, - Source.url == url, - ) - )).scalar_one_or_none() - if existing is not None: - return existing - if dry_run: - return None - s = Source(artist_id=artist_id, platform=platform, url=url, enabled=False) - db.add(s) - await db.flush() - return s.id - - -async def _find_or_create_post( - db: AsyncSession, *, - source_id: int, external_post_id: str, - title: str | None, description: str | None, post_url: str | None, - post_date_iso: str | None, attachment_count: int, dry_run: bool, -) -> int | None: - existing = (await db.execute( - select(Post.id).where( - Post.source_id == source_id, - Post.external_post_id == external_post_id, - ) - )).scalar_one_or_none() - if existing is not None: - return existing - if dry_run: - return None - post_date = None - if post_date_iso: - post_date = datetime.fromisoformat(post_date_iso) - p = Post( - source_id=source_id, - external_post_id=external_post_id, - post_title=title, - description=description, - post_url=post_url, - post_date=post_date, - attachment_count=attachment_count, - raw_metadata={"migrated_from": "imagerepo"}, - ) - db.add(p) - await db.flush() - return p.id - - -async def _ensure_provenance( - db: AsyncSession, *, - image_id: int, post_id: int, source_id: int, dry_run: bool, -) -> bool: - """Returns True if a new ImageProvenance row was inserted. - - Also sets ImageRecord.primary_post_id to this post if the image - doesn't already have one — preserves any primary_post_id already - assigned at download time by the importer (don't clobber). This is - the linkage gallery_service.py uses to surface Post.post_date as - the image's effective date for sort/group/jump/neighbor nav. - """ - existing = (await db.execute( - select(ImageProvenance.id).where( - ImageProvenance.image_record_id == image_id, - ImageProvenance.post_id == post_id, - ImageProvenance.source_id == source_id, - ) - )).scalar_one_or_none() - - # Whether-or-not the provenance row already exists, ensure the - # image's primary_post_id is set so the gallery date-coalesce works. - # Idempotent: only writes when currently NULL. - if not dry_run: - await db.execute( - ImageRecord.__table__.update() - .where(ImageRecord.id == image_id) - .where(ImageRecord.primary_post_id.is_(None)) - .values(primary_post_id=post_id) - ) - - if existing is not None: - return False - if dry_run: - return True - db.add(ImageProvenance( - image_record_id=image_id, post_id=post_id, source_id=source_id, - )) - await db.flush() - return True - - -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 - - # 4. Image posts (schema v2) → Source + Post + ImageProvenance. - # Restores IR PostMetadata as FC's downloader-track provenance, - # so the modal's ProvenancePanel surfaces title/description/ - # source URL/publish date the same way it does for live - # gallery-dl downloads. - for entry in manifest.get("image_posts", []): - counts["rows_processed"] += 1 - platform = entry.get("platform") - artist_name = entry.get("artist") - if not platform or not artist_name: - counts["rows_skipped"] += 1 - continue - - aid = await _ensure_artist_id(db, artist_name, dry_run) - if aid is None: - counts["rows_skipped"] += 1 - continue - - url = _profile_url(platform, slugify(artist_name)) - if url is None: - counts["rows_skipped"] += 1 - continue - - source_id = await _find_or_create_source( - db, artist_id=aid, platform=platform, url=url, dry_run=dry_run, - ) - if source_id is None: - counts["rows_skipped"] += 1 - continue - - post_id = await _find_or_create_post( - db, source_id=source_id, - external_post_id=entry.get("post_id") or "", - title=entry.get("title"), - description=entry.get("description"), - post_url=entry.get("source_url"), - post_date_iso=entry.get("published_at"), - attachment_count=entry.get("attachment_count") or 0, - dry_run=dry_run, - ) - if post_id is None: - counts["rows_skipped"] += 1 - continue - - for sha in entry.get("image_sha256s", []): - img_id = await _sha_to_image_id(db, sha) - if img_id is None: - unmatched.append({ - "kind": "post", "sha256": sha, - "post_id": entry.get("post_id"), - }) - continue - inserted = await _ensure_provenance( - db, image_id=img_id, post_id=post_id, - source_id=source_id, dry_run=dry_run, - ) - if inserted: - counts["rows_inserted"] += 1 - else: - counts["rows_skipped"] += 1 - - if not dry_run: - await db.commit() - return {"counts": counts, "unmatched": unmatched} diff --git a/backend/app/services/migrators/verify.py b/backend/app/services/migrators/verify.py deleted file mode 100644 index 52bf37c..0000000 --- a/backend/app/services/migrators/verify.py +++ /dev/null @@ -1,80 +0,0 @@ -"""Post-migration verification: row counts + sha256 sampling.""" -from __future__ import annotations - -import hashlib -from pathlib import Path - -from sqlalchemy import func, select -from sqlalchemy.ext.asyncio import AsyncSession - -from ...models import Artist, Credential, ImageRecord, Source, Tag - - -async def verify_async(db: AsyncSession, *, expected: dict | None = None) -> dict: - """Return per-check status dicts. `expected` is optional row-count - assertions; checks default to status='ok' when no expected provided.""" - expected = expected or {} - results: dict[str, dict] = {} - - checks = { - "artist_subscriptions": ( - select(func.count(Artist.id)).where(Artist.is_subscription.is_(True)) - ), - "source_count": select(func.count(Source.id)), - "credential_count": select(func.count(Credential.id)), - "tag_count": select(func.count(Tag.id)), - "image_record_imported_or_downloaded": ( - select(func.count(ImageRecord.id)) - .where(ImageRecord.origin.in_(["imported_filesystem", "downloaded"])) - ), - } - for name, stmt in checks.items(): - actual = (await db.execute(stmt)).scalar_one() - exp = expected.get(name) - status = "ok" if exp is None or exp == actual else "mismatch" - results[name] = {"status": status, "actual": int(actual), "expected": exp} - return results - - -async def verify_sha256_sample( - db: AsyncSession, *, sample_size: int = 20, -) -> dict: - """Sample N image_records; verify file exists + sha256 matches.""" - rows = (await db.execute( - select(ImageRecord.id, ImageRecord.path, ImageRecord.sha256) - .order_by(func.random()).limit(sample_size) - )).all() - - matched = 0 - mismatched = 0 - missing = 0 - samples: list[dict] = [] - for img_id, path, expected_sha in rows: - p = Path(path) - # Sync stdlib filesystem ops are intentional: this verify pass runs - # inside a Celery task under asyncio.run; no other awaitables compete - # for the loop. Same pattern as download_service.py. - if not p.exists(): # noqa: ASYNC240 - missing += 1 - samples.append({"id": img_id, "path": path, "result": "missing"}) - continue - h = hashlib.sha256() - with p.open("rb") as f: # noqa: ASYNC230 - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) - if h.hexdigest() == expected_sha: - matched += 1 - samples.append({"id": img_id, "result": "ok"}) - else: - mismatched += 1 - samples.append({ - "id": img_id, "path": path, "result": "mismatch", - "expected_sha": expected_sha, "actual_sha": h.hexdigest(), - }) - return { - "sample_size": len(rows), - "matched": matched, - "mismatched": mismatched, - "missing": missing, - "samples": samples, - } diff --git a/backend/app/tasks/migration.py b/backend/app/tasks/migration.py deleted file mode 100644 index aec5102..0000000 --- a/backend/app/tasks/migration.py +++ /dev/null @@ -1,169 +0,0 @@ -"""FC-5 run_migration Celery task. - -Dispatches to the right migrator based on `kind`. Updates MigrationRun -row's status/counts/finished_at as it runs. Failures set status='error' -with the error message preserved. - -kinds: gs_ingest, ir_ingest, tag_apply, ml_queue, verify, cleanup -(backup + rollback retired 2026-05-24 → see /api/system/backup/*) -""" -from __future__ import annotations - -import asyncio -import logging -from datetime import UTC, datetime -from pathlib import Path - -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from ..celery_app import celery -from ..models import MigrationRun -from ..services.credential_crypto import CredentialCrypto -from ..services.migrators import cleanup as cleanup_mod -from ..services.migrators import gs_ingest, ir_ingest, ml_queue, tag_apply, verify -from ._async_session import async_session_factory - -log = logging.getLogger(__name__) - -IMAGES_ROOT = Path("/images") -_KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64" - - -async def _update_run( - db: AsyncSession, run_id: int, *, - status: str | None = None, counts: dict | None = None, - error: str | None = None, finished_at: datetime | None = None, - metadata_patch: dict | None = None, -) -> None: - run = (await db.execute( - select(MigrationRun).where(MigrationRun.id == run_id) - )).scalar_one() - if status is not None: - run.status = status - if counts is not None: - run.counts = counts - if error is not None: - run.error = error - if finished_at is not None: - run.finished_at = finished_at - if metadata_patch: - run.metadata_ = {**(run.metadata_ or {}), **metadata_patch} - await db.commit() - - -async def _run_async(run_id: int, kind: str, params: dict) -> dict: - factory, engine = async_session_factory() - try: - async with factory() as db: - await _update_run(db, run_id, status="running") - try: - if kind in ("backup", "rollback"): - raise ValueError( - f"kind {kind!r} retired in FC-3h; " - "use /api/system/backup/* instead" - ) - - elif kind == "gs_ingest": - fc_crypto = CredentialCrypto(_KEY_PATH) - counts = await gs_ingest.migrate_async( - db, data=params["data"], - fc_crypto=fc_crypto, - dry_run=params.get("dry_run", False), - ) - await _update_run( - db, run_id, status="ok", counts=counts, - finished_at=datetime.now(UTC), - ) - return counts - - elif kind == "ir_ingest": - counts = await ir_ingest.migrate_async( - db, data=params["data"], - images_root=IMAGES_ROOT, - dry_run=params.get("dry_run", False), - ) - await _update_run( - db, run_id, status="ok", counts=counts, - finished_at=datetime.now(UTC), - ) - return counts - - elif kind == "tag_apply": - result = await tag_apply.apply_async( - db, images_root=IMAGES_ROOT, - dry_run=params.get("dry_run", False), - ) - await _update_run( - db, run_id, status="ok", - counts=result["counts"], - finished_at=datetime.now(UTC), - metadata_patch={"unmatched": result["unmatched"]}, - ) - return result - - elif kind == "ml_queue": - count = await ml_queue.queue_all_unprocessed_async(db) - await _update_run( - db, run_id, status="ok", - counts={"rows_processed": count, "rows_inserted": 0, - "rows_skipped": 0, "files_copied": 0, - "bytes_copied": 0, "conflicts": 0}, - finished_at=datetime.now(UTC), - ) - return {"queued": count} - - elif kind == "verify": - checks = await verify.verify_async(db, expected=params.get("expected")) - sample = await verify.verify_sha256_sample( - db, sample_size=params.get("sample_size", 20), - ) - await _update_run( - db, run_id, status="ok", - counts={"rows_processed": sample["sample_size"], - "rows_inserted": 0, "rows_skipped": 0, - "files_copied": 0, "bytes_copied": 0, - "conflicts": sample["mismatched"] + sample["missing"]}, - finished_at=datetime.now(UTC), - metadata_patch={"checks": checks, "sample": sample}, - ) - return {"checks": checks, "sample": sample} - - elif kind == "cleanup": - slug = params.get("slug") - if not slug: - raise ValueError("cleanup requires params.slug") - result = await cleanup_mod.cleanup_artist_async( - db, slug=slug, images_root=IMAGES_ROOT, - dry_run=params.get("dry_run", False), - source_path_prefix=params.get("source_path_prefix"), - ) - await _update_run( - db, run_id, status="ok", - counts=result["counts"], - finished_at=datetime.now(UTC), - metadata_patch={ - "artist": result["artist"], - "summary": result["summary"], - }, - ) - return result - - else: - raise ValueError(f"unknown kind: {kind}") - - except Exception as exc: - log.exception("migration kind=%s failed", kind) - await _update_run( - db, run_id, status="error", error=str(exc), - finished_at=datetime.now(UTC), - ) - raise - finally: - await engine.dispose() - - -@celery.task(name="backend.app.tasks.migration.run_migration", bind=True, acks_late=True) -def run_migration(self, run_id: int, kind: str, params: dict) -> dict: - """FC-5: dispatch a migration kind. Updates MigrationRun row as it goes.""" - return asyncio.run(_run_async(run_id, kind, params)) diff --git a/frontend/src/components/settings/LegacyMigrationCard.vue b/frontend/src/components/settings/LegacyMigrationCard.vue deleted file mode 100644 index be2dfd1..0000000 --- a/frontend/src/components/settings/LegacyMigrationCard.vue +++ /dev/null @@ -1,245 +0,0 @@ - - - - - diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 0258bc1..e332e1c 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -17,8 +17,8 @@ - + theme, and clusters with the other audit cards. LegacyMigrationCard + removed once the one-and-done GS/IR migration cutover completed. --> @@ -32,7 +32,6 @@ import MLThresholdSliders from './MLThresholdSliders.vue' import AllowlistTable from './AllowlistTable.vue' import AliasTable from './AliasTable.vue' import BackupCard from './BackupCard.vue' -import LegacyMigrationCard from './LegacyMigrationCard.vue' import { useSystemActivityStore } from '../../stores/systemActivity.js' // Poll queue depths so each card's QueueStatusBar shows live pending diff --git a/frontend/src/stores/migration.js b/frontend/src/stores/migration.js deleted file mode 100644 index 5bdebd1..0000000 --- a/frontend/src/stores/migration.js +++ /dev/null @@ -1,83 +0,0 @@ -import { defineStore } from 'pinia' -import { ref, computed } from 'vue' -import { useApi } from '../composables/useApi.js' - -const POLL_MS = 2000 - -export const useMigrationStore = defineStore('migration', () => { - const api = useApi() - - const activeRun = ref(null) - const recentRuns = ref([]) - const error = ref(null) - const loading = ref(false) - - let pollHandle = null - - async function trigger(kind, params, file = null) { - loading.value = true - error.value = null - try { - let body - if (file) { - // Multipart upload for ingest kinds. - const form = new FormData() - form.append('export_file', file) - if (params && params.dry_run) form.append('dry_run', 'true') - if (params && params.force) form.append('force', 'true') - body = await api.post(`/api/migrate/${kind}`, { body: form }) - } else { - body = await api.post(`/api/migrate/${kind}`, { body: params || {} }) - } - await loadRun(body.run_id) - pollActive(body.run_id) - return body - } catch (e) { - error.value = e - throw e - } finally { - loading.value = false - } - } - - async function loadRun(runId) { - activeRun.value = await api.get(`/api/migrate/runs/${runId}`) - } - - async function loadRecent() { - recentRuns.value = await api.get('/api/migrate/runs', { params: { limit: 10 } }) - } - - function pollActive(runId) { - stopPolling() - pollHandle = setInterval(async () => { - try { - const run = await api.get(`/api/migrate/runs/${runId}`) - activeRun.value = run - if (run.status === 'ok' || run.status === 'error') { - stopPolling() - await loadRecent() - } - } catch (e) { - error.value = e - stopPolling() - } - }, POLL_MS) - } - - function stopPolling() { - if (pollHandle) { - clearInterval(pollHandle) - pollHandle = null - } - } - - const isRunning = computed( - () => activeRun.value && activeRun.value.status === 'running', - ) - - return { - activeRun, recentRuns, error, loading, isRunning, - trigger, loadRun, loadRecent, pollActive, stopPolling, - } -}) diff --git a/tests/test_api_migrate.py b/tests/test_api_migrate.py deleted file mode 100644 index 9febf96..0000000 --- a/tests/test_api_migrate.py +++ /dev/null @@ -1,98 +0,0 @@ -"""FC-5: /api/migrate API tests.""" -import io -import json - -import pytest -from werkzeug.datastructures import FileStorage - -import backend.app.tasks.migration # noqa: F401 — register celery task -from backend.app.celery_app import celery - -pytestmark = pytest.mark.integration - - -def _file_storage(payload: dict, filename: str = "export.json") -> FileStorage: - return FileStorage( - stream=io.BytesIO(json.dumps(payload).encode("utf-8")), - filename=filename, - content_type="application/json", - ) - - -# Retired 2026-05-24 (FC-3h): `test_post_backup_returns_202_and_id` -# asserted the /api/migrate/backup endpoint, which was retired in FC-3h. -# Backup is now a first-class feature at /api/system/backup/*; -# coverage lives in tests/test_api_system_backup.py. - - -@pytest.mark.asyncio -async def test_post_rejects_unknown_kind(client): - resp = await client.post("/api/migrate/destroy", json={}) - assert resp.status_code == 400 - body = await resp.get_json() - assert body["error"] == "unknown_kind" - - -@pytest.mark.asyncio -async def test_post_ingest_rejects_missing_file(client): - resp = await client.post("/api/migrate/gs_ingest", form={}) - assert resp.status_code == 400 - body = await resp.get_json() - assert body["error"] == "missing_export_file" - - -@pytest.mark.asyncio -async def test_post_ingest_accepts_multipart_file(client, monkeypatch): - monkeypatch.setattr( - "backend.app.api.migrate.run_migration", - type("F", (), {"delay": lambda self, *a, **k: None})(), - ) - - export = { - "source_app": "gallerysubscriber", "schema_version": 1, - "subscriptions": [], "credentials": [], - } - resp = await client.post( - "/api/migrate/gs_ingest", - form={"dry_run": "false"}, - files={"export_file": _file_storage(export, "gs-export.json")}, - ) - assert resp.status_code == 202 - - -# Retired 2026-05-24: `test_post_apply_without_backup_rejected` asserted -# the migration backup-gate rejected applies without a recent backup. -# That gate was removed in commit 5535677 (_APPLY_KINDS is now empty, -# FC-3h will own backup as a first-class feature) so the test was -# testing dead behavior. Removed rather than rewritten. - - -@pytest.mark.asyncio -async def test_post_dry_run_allowed_without_backup(client, monkeypatch): - # FC-3h: the _has_recent_pre_migration_backup gate was retired; dry-run - # ingests were always allowed and now non-dry-run ingests are too. - # This test stays as regression coverage that dry_run ingest still works. - monkeypatch.setattr( - "backend.app.api.migrate.run_migration", - type("F", (), {"delay": lambda self, *a, **k: None})(), - ) - export = { - "source_app": "gallerysubscriber", "schema_version": 1, - "subscriptions": [], "credentials": [], - } - resp = await client.post( - "/api/migrate/gs_ingest", - form={"dry_run": "true"}, - files={"export_file": _file_storage(export, "gs-export.json")}, - ) - assert resp.status_code == 202 - - -@pytest.mark.asyncio -async def test_get_run_404_unknown(client): - resp = await client.get("/api/migrate/runs/9999999") - assert resp.status_code == 404 - - -def test_run_migration_task_registered(): - assert "backend.app.tasks.migration.run_migration" in celery.tasks diff --git a/tests/test_gs_ingest.py b/tests/test_gs_ingest.py deleted file mode 100644 index 189bbef..0000000 --- a/tests/test_gs_ingest.py +++ /dev/null @@ -1,122 +0,0 @@ -"""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.credential_type == "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) diff --git a/tests/test_ir_ingest.py b/tests/test_ir_ingest.py deleted file mode 100644 index 2af8ac8..0000000 --- a/tests/test_ir_ingest.py +++ /dev/null @@ -1,129 +0,0 @@ -"""FC-5: ir_ingest unit tests.""" -import json -from datetime import UTC, datetime - -import pytest -from sqlalchemy import select - -from backend.app.models import Tag -from backend.app.services.migrators import ir_ingest - -pytestmark = pytest.mark.integration - - -def _ir_export(tags=None, artist_assignments=None, tag_associations=None, series_pages=None): - return { - "source_app": "imagerepo", - "schema_version": 1, - "exported_at": datetime.now(UTC).isoformat(), - "tags": tags or [], - "image_artist_assignments": artist_assignments or [], - "image_tag_associations": tag_associations or [], - "series_pages": series_pages or [], - } - - -@pytest.mark.asyncio -async def test_general_tag_created(db, tmp_path): - data = _ir_export(tags=[ - {"name": "blue_eyes", "kind": "general", "fandom_name": None}, - ]) - await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - - tag = (await db.execute( - select(Tag).where(Tag.name == "blue_eyes", Tag.kind == "general") - )).scalar_one() - assert tag.fandom_id is None - - -@pytest.mark.asyncio -async def test_character_tag_resolves_fandom_name(db, tmp_path): - data = _ir_export(tags=[ - {"name": "Wonderland", "kind": "fandom", "fandom_name": None}, - {"name": "Alice", "kind": "character", "fandom_name": "Wonderland"}, - ]) - await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - - fandom = (await db.execute( - select(Tag).where(Tag.name == "Wonderland", Tag.kind == "fandom") - )).scalar_one() - alice = (await db.execute( - select(Tag).where(Tag.name == "Alice", Tag.kind == "character") - )).scalar_one() - assert alice.fandom_id == fandom.id - - -@pytest.mark.asyncio -async def test_artist_and_post_kinds_skipped(db, tmp_path): - data = _ir_export(tags=[ - {"name": "BobArtist", "kind": "artist", "fandom_name": None}, - {"name": "PostXYZ", "kind": "post", "fandom_name": None}, - {"name": "general_one", "kind": "general", "fandom_name": None}, - ]) - counts = await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - assert counts["rows_skipped"] >= 2 - - bob = (await db.execute( - select(Tag).where(Tag.name == "BobArtist") - )).scalar_one_or_none() - post = (await db.execute( - select(Tag).where(Tag.name == "PostXYZ") - )).scalar_one_or_none() - assert bob is None - assert post is None - - -@pytest.mark.asyncio -async def test_manifest_written_to_disk(db, tmp_path): - data = _ir_export( - artist_assignments=[{"sha256": "a" * 64, "artist_name": "Alice"}], - tag_associations=[ - {"sha256": "a" * 64, "tag_name": "blue_eyes", "tag_kind": "general", "fandom_name": None}, - ], - series_pages=[ - {"sha256": "a" * 64, "series_tag_name": "MySeries", "page_number": 1}, - ], - ) - await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - - manifest_file = tmp_path / "_migration_state" / "ir_tag_manifest.json" - assert manifest_file.exists() - manifest = json.loads(manifest_file.read_text()) - assert manifest["schema_version"] == 1 - assert len(manifest["image_artist_assignments"]) == 1 - assert len(manifest["image_tag_associations"]) == 1 - assert len(manifest["series_pages"]) == 1 - - -@pytest.mark.asyncio -async def test_dry_run_writes_no_manifest_and_no_tags(db, tmp_path): - data = _ir_export( - tags=[{"name": "dry", "kind": "general", "fandom_name": None}], - artist_assignments=[{"sha256": "a" * 64, "artist_name": "DryArtist"}], - ) - await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=True) - - tag = (await db.execute( - select(Tag).where(Tag.name == "dry") - )).scalar_one_or_none() - assert tag is None - manifest_file = tmp_path / "_migration_state" / "ir_tag_manifest.json" - assert not manifest_file.exists() - - -@pytest.mark.asyncio -async def test_idempotent_on_rerun(db, tmp_path): - data = _ir_export(tags=[ - {"name": "rerun", "kind": "general", "fandom_name": None}, - ]) - await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - counts2 = await ir_ingest.migrate_async(db, data=data, images_root=tmp_path, dry_run=False) - assert counts2["rows_skipped"] >= 1 - - -@pytest.mark.asyncio -async def test_rejects_wrong_source_app(db, tmp_path): - bad = {"source_app": "elsewhere", "schema_version": 1, "tags": [], - "image_artist_assignments": [], "image_tag_associations": [], "series_pages": []} - with pytest.raises(ValueError): - await ir_ingest.migrate_async(db, data=bad, images_root=tmp_path, dry_run=False) diff --git a/tests/test_migration_verify.py b/tests/test_migration_verify.py deleted file mode 100644 index 549cabb..0000000 --- a/tests/test_migration_verify.py +++ /dev/null @@ -1,72 +0,0 @@ -"""FC-5: verify + ml_queue unit tests.""" -import hashlib - -import pytest - -from backend.app.models import Artist, ImageRecord -from backend.app.services.migrators import ml_queue, verify - -pytestmark = pytest.mark.integration - - -@pytest.mark.asyncio -async def test_verify_row_counts_match(db): - db.add(Artist(name="vAlpha", slug="valpha", is_subscription=True)) - db.add(Artist(name="vBeta", slug="vbeta", is_subscription=False)) - await db.commit() - result = await verify.verify_async( - db, expected={"artist_subscriptions": 1}, - ) - assert result["artist_subscriptions"]["status"] == "ok" - - -@pytest.mark.asyncio -async def test_verify_sha256_sample_matches_disk(db, tmp_path): - f = tmp_path / "sample.bin" - f.write_bytes(b"hello world") - sha = hashlib.sha256(b"hello world").hexdigest() - db.add(ImageRecord( - path=str(f), sha256=sha, size_bytes=11, mime="application/octet-stream", - origin="imported_filesystem", - )) - await db.commit() - result = await verify.verify_sha256_sample(db, sample_size=10) - assert result["matched"] >= 1 - assert result["mismatched"] == 0 - - -@pytest.mark.asyncio -async def test_verify_sha256_sample_detects_missing_file(db, tmp_path): - sha = "f" * 64 - db.add(ImageRecord( - path=str(tmp_path / "does-not-exist.bin"), - sha256=sha, size_bytes=1, mime="application/octet-stream", - origin="imported_filesystem", - )) - await db.commit() - result = await verify.verify_sha256_sample(db, sample_size=10) - assert result["missing"] >= 1 - - -@pytest.mark.asyncio -async def test_ml_queue_returns_count_for_unprocessed(db, tmp_path, monkeypatch): - queued: list[int] = [] - - class _FakeTask: - def delay(self, image_id): - queued.append(image_id) - - monkeypatch.setattr( - "backend.app.tasks.ml.tag_and_embed", _FakeTask(), - ) - - f = tmp_path / "q.bin" - f.write_bytes(b"x") - db.add(ImageRecord( - path=str(f), sha256="q" * 64, size_bytes=1, - mime="application/octet-stream", origin="imported_filesystem", - )) - await db.commit() - - count = await ml_queue.queue_all_unprocessed_async(db) - assert count >= 1 diff --git a/tests/test_tag_apply.py b/tests/test_tag_apply.py deleted file mode 100644 index d1385c1..0000000 --- a/tests/test_tag_apply.py +++ /dev/null @@ -1,395 +0,0 @@ -"""FC-5: tag_apply unit tests.""" -import json - -import pytest -from sqlalchemy import func, select - -from backend.app.models import ( - Artist, - ImageProvenance, - ImageRecord, - Post, - SeriesPage, - Source, - 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, image_posts=None, schema_version=2, -): - d = tmp_path / "_migration_state" - d.mkdir(parents=True, exist_ok=True) - (d / "ir_tag_manifest.json").write_text(json.dumps({ - "schema_version": schema_version, - "image_artist_assignments": artist_assignments or [], - "image_tag_associations": tag_associations or [], - "series_pages": series_pages or [], - "image_posts": image_posts 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 - - -# --- Phase 4: image_posts (schema v2) ----------------------------------- - - -_POST_ENTRY = { - "platform": "patreon", - "post_id": "10001", - "artist": "Maewix", - "title": "Test Post Title", - "description": "

HTML description

", - "source_url": "https://www.patreon.com/posts/test-10001", - "attachment_count": 1, - "published_at": "2026-01-15T12:00:00+00:00", -} - - -@pytest.mark.asyncio -async def test_image_posts_creates_source_post_provenance(db, tmp_path): - sha = "d" * 64 - await _seed_image(db, sha, suffix="d") - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "image_sha256s": [sha]}, - ]) - - result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False) - assert result["counts"]["rows_inserted"] >= 1 - - # Async Core-DML assertions — column selects, not ORM attribute access. - artist_count = (await db.execute( - select(func.count(Artist.id)).where(Artist.slug == "maewix") - )).scalar_one() - assert artist_count == 1 - - source_count = (await db.execute( - select(func.count(Source.id)).where( - Source.platform == "patreon", - Source.url == "https://www.patreon.com/maewix", - ) - )).scalar_one() - assert source_count == 1 - - post_count = (await db.execute( - select(func.count(Post.id)).where(Post.external_post_id == "10001") - )).scalar_one() - assert post_count == 1 - - img_id = (await db.execute( - select(ImageRecord.id).where(ImageRecord.sha256 == sha) - )).scalar_one() - prov_count = (await db.execute( - select(func.count(ImageProvenance.id)) - .where(ImageProvenance.image_record_id == img_id) - )).scalar_one() - assert prov_count == 1 - - # Phase 4 must also set ImageRecord.primary_post_id so the gallery's - # effective_date COALESCE can surface Post.post_date. Operator-flagged - # 2026-05-25: without this, IR-migrated images keep sorting by FC's - # scan date instead of the original publish date. - primary_post_id = (await db.execute( - select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id) - )).scalar_one() - canonical_post_id = (await db.execute( - select(Post.id).where(Post.external_post_id == "10001") - )).scalar_one() - assert primary_post_id == canonical_post_id - - -@pytest.mark.asyncio -async def test_image_posts_primary_post_id_not_clobbered(db, tmp_path): - """If the importer already set primary_post_id (e.g. a downloaded - image with a known provenance), phase 4 must NOT overwrite it when - re-running tag_apply against the IR migration. The existing - download-time linkage is the source of truth.""" - sha = "9" * 64 - await _seed_image(db, sha, suffix="9") - # Pre-set primary_post_id to a sentinel Post so we can detect a clobber. - img_id = (await db.execute( - select(ImageRecord.id).where(ImageRecord.sha256 == sha) - )).scalar_one() - # Build an existing Source + Post for the sentinel. - art = Artist(name="Pre-existing", slug="pre-existing") - db.add(art) - await db.flush() - src = Source( - artist_id=art.id, platform="patreon", - url="https://www.patreon.com/pre-existing", - ) - db.add(src) - await db.flush() - sentinel_post = Post( - source_id=src.id, external_post_id="sentinel-99", - post_title="Pre-existing", - ) - db.add(sentinel_post) - await db.flush() - await db.execute( - ImageRecord.__table__.update() - .where(ImageRecord.id == img_id) - .values(primary_post_id=sentinel_post.id) - ) - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "image_sha256s": [sha]}, - ]) - await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False) - - # The migration created a NEW Post (external_post_id="10001") and a - # new ImageProvenance, but primary_post_id must still point at the - # original sentinel. - primary_post_id = (await db.execute( - select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id) - )).scalar_one() - assert primary_post_id == sentinel_post.id - - -@pytest.mark.asyncio -async def test_image_posts_idempotent_on_rerun(db, tmp_path): - sha = "e" * 64 - await _seed_image(db, sha, suffix="e") - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "image_sha256s": [sha]}, - ]) - - 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) - - # Re-run: nothing should be newly inserted; all skipped. - assert result2["counts"]["rows_inserted"] == 0 - assert result2["counts"]["rows_skipped"] >= 1 - - prov_count = (await db.execute( - select(func.count(ImageProvenance.id)) - )).scalar_one() - assert prov_count == 1 - - -@pytest.mark.asyncio -async def test_image_posts_unknown_sha_unmatched(db, tmp_path): - # No image seeded; sha doesn't exist. - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "image_sha256s": ["f" * 64]}, - ]) - result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False) - unmatched_kinds = [u["kind"] for u in result["unmatched"]] - assert "post" in unmatched_kinds - - -@pytest.mark.asyncio -async def test_image_posts_unknown_platform_skipped(db, tmp_path): - sha = "1" * 64 - await _seed_image(db, sha, suffix="1") - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "platform": "tumblr", "image_sha256s": [sha]}, - ]) - result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False) - assert result["counts"]["rows_skipped"] >= 1 - - # No Post / Source / ImageProvenance should have been created. - src_count = (await db.execute( - select(func.count(Source.id)).where(Source.platform == "tumblr") - )).scalar_one() - assert src_count == 0 - - -@pytest.mark.parametrize("platform,expected_url", [ - ("deviantart", "https://www.deviantart.com/maewix"), - ("pixiv", "https://www.pixiv.net/users/maewix"), -]) -@pytest.mark.asyncio -async def test_image_posts_extended_platforms_create_source( - db, tmp_path, platform, expected_url, -): - """Regression for 2026-05-25 operator-reported bug: phase 4's - _PLATFORM_PROFILE_URL had only patreon/subscribestar/hentaifoundry, - silently dropping deviantart + pixiv PostMetadata from the IR migration.""" - sha = f"{platform[0]}" * 64 - await _seed_image(db, sha, suffix=f"_{platform}") - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "platform": platform, "image_sha256s": [sha]}, - ]) - - result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False) - assert result["counts"]["rows_inserted"] >= 1 - - src_url = (await db.execute( - select(Source.url).where(Source.platform == platform) - )).scalar_one() - assert src_url == expected_url - - # ImageProvenance row was created. - prov_count = (await db.execute( - select(func.count(ImageProvenance.id)) - .join(Source, Source.id == ImageProvenance.source_id) - .where(Source.platform == platform) - )).scalar_one() - assert prov_count == 1 - - -@pytest.mark.asyncio -async def test_image_posts_dry_run_makes_no_writes(db, tmp_path): - sha = "2" * 64 - await _seed_image(db, sha, suffix="2") - await db.commit() - - _write_manifest(tmp_path, image_posts=[ - {**_POST_ENTRY, "image_sha256s": [sha]}, - ]) - - await tag_apply.apply_async(db, images_root=tmp_path, dry_run=True) - prov_count = (await db.execute( - select(func.count(ImageProvenance.id)) - )).scalar_one() - assert prov_count == 0