refactor(I5): remove one-and-done GS/IR migration tooling
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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.
|
||||
"""
|
||||
@@ -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/<bucket>/...`, 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,
|
||||
},
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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}
|
||||
@@ -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,
|
||||
}
|
||||
Reference in New Issue
Block a user