feat(tags): retro-normalize existing tags to Title Case + merge case-collisions (plan #714)

Follow-up to #701: new tags are saved canonical, but the back-catalog keeps
whatever casing it was created with. This adds a maintenance action that
Title-Cases every existing tag (collapsing whitespace) and merges
case/whitespace-variant duplicates into one.

Backend:
- tag_service.normalize_existing_tags(session, *, dry_run): groups all tags by
  (kind, coalesce(fandom_id,-1), canonical_name). Per group it picks a survivor
  (prefer an already-canonical member → no rename/self-alias; else the
  best-connected tag → fewest FK repoints; else lowest id), merges the variants
  INTO it via the tested TagService._do_merge (image_tag/allowlist/embedding/
  aliases/series_page repoints + protective ML aliases), then renames the
  survivor to canonical. Losers are deleted before the rename so there's no
  transient unique-index clash; commits per group and isolates failures per
  group. Idempotent — an already-canonical lone tag is a no-op.
- normalize_tags_task (maintenance queue, asyncio.run + per-task NullPool async
  engine, soft 1800/hard 2400) — recovery/timeout/duration covered by FC-3i.
- POST /api/admin/tags/normalize: dry_run=true returns a projection inline
  (group/collision/rename counts + sample); dry_run=false enqueues the task.

Frontend: a "Standardize tag casing" section in TagMaintenanceCard (Cleanup
tab) — preview → apply (polls the activity dashboard to terminal status),
behind a back-up-first warning. admin store gains normalizeTags().

Tests: tests/test_tag_normalize.py — dry-run counts, live merge + image-tag
dedup/repoint, idempotency, same-name-different-fandom and -different-kind kept
separate, ML-known loser keeps a protective alias.

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

backend/app/api/admin.py

@@ -245,6 +245,32 @@ async def tags_reset_content():
     return jsonify(result)
 
 
+@admin_bp.route("/tags/normalize", methods=["POST"])
+async def tags_normalize():
+    """#714: retro-normalize existing tags to the #701 canonical form (Title
+    Case + collapsed whitespace) and merge case/whitespace-variant duplicates.
+
+    dry_run=true (default) returns a projection inline — group/collision/rename
+    counts + a sample of the changes — so the UI shows exactly what'll happen.
+    dry_run=false dispatches the long-running maintenance task (the merge FK
+    repoints can touch many tags); the UI tails the activity dashboard for the
+    summary. Idempotent; back up first (the merges are irreversible)."""
+    from ..services.tag_service import normalize_existing_tags
+
+    body = await request.get_json(silent=True) or {}
+    dry_run = bool(body.get("dry_run", True))
+
+    if dry_run:
+        async with get_session() as session:
+            result = await normalize_existing_tags(session, dry_run=True)
+        return jsonify(result)
+
+    from ..tasks.admin import normalize_tags_task
+
+    async_result = normalize_tags_task.delay()
+    return jsonify({"task_id": async_result.id, "status": "queued"}), 202
+
+
 @admin_bp.route("/maintenance/db-stats", methods=["GET"])
 async def db_stats():
     """Per-table bloat readout (pg_stat_user_tables) for the high-churn tables

backend/app/services/tag_service.py

@@ -1,5 +1,6 @@
 """Tag CRUD + autocomplete + image-tag association."""
 
+import logging
 from collections.abc import Sequence
 from dataclasses import dataclass
 
@@ -12,6 +13,8 @@ from ..models import Tag, TagKind, image_tag
 from ..models.tag_allowlist import TagAllowlist
 from ..models.tag_reference_embedding import TagReferenceEmbedding
 
+log = logging.getLogger(__name__)
+
 
 def normalize_tag_name(name: str) -> str:
     """Canonical tag form (#701): collapse whitespace + Title Case.
@@ -526,6 +529,21 @@ class TagService:
             update(Tag).where(Tag.fandom_id == src).values(fandom_id=tgt)
         )
 
+    async def _image_assoc_counts(self, tag_ids: list[int]) -> dict[int, int]:
+        """image_tag row counts keyed by tag_id, for the survivor heuristic
+        (the best-connected tag in a collision group survives → fewest
+        FK repoints). Tags with zero associations are absent from the map."""
+        if not tag_ids:
+            return {}
+        rows = (
+            await self.session.execute(
+                select(image_tag.c.tag_id, func.count())
+                .where(image_tag.c.tag_id.in_(tag_ids))
+                .group_by(image_tag.c.tag_id)
+            )
+        ).all()
+        return {tid: int(n) for tid, n in rows}
+
     async def _create_protective_aliases(
         self, src_name: str, src_kind: TagKind, tgt: int
     ) -> bool:
@@ -573,3 +591,161 @@ class TagService:
             if res.rowcount:
                 created = True
         return created
+
+
+# ---------------------------------------------------------------------------
+# #714: retro-normalize existing tags to the #701 canonical (Title Case +
+# collapsed whitespace) and merge case/whitespace-variant duplicates.
+# ---------------------------------------------------------------------------
+
+_NORMALIZE_SAMPLE_CAP = 50
+
+
+def _group_existing_tags(
+    rows,
+) -> dict[tuple, list[tuple[int, str]]]:
+    """Group (id, name, kind, fandom_id) rows by their post-normalization
+    identity: (kind, COALESCE(fandom_id, -1), canonical_name). Every tag that
+    would collapse to the same canonical lives in ONE group, so the canonical
+    form is unique within (kind, fandom) once each group is resolved."""
+    groups: dict[tuple, list[tuple[int, str]]] = {}
+    for tag_id, name, kind, fandom_id in rows:
+        canonical = normalize_tag_name(name)
+        key = (kind, fandom_id if fandom_id is not None else -1, canonical)
+        groups.setdefault(key, []).append((tag_id, name))
+    return groups
+
+
+def _group_needs_change(canonical: str, members: list[tuple[int, str]]) -> bool:
+    """A group is already canonical iff it's a single member whose name equals
+    the canonical form. Anything else (a collision, or a lone mis-cased tag)
+    needs work."""
+    if len(members) > 1:
+        return True
+    return members[0][1] != canonical
+
+
+def _best_connected(tag_ids: list[int], counts: dict[int, int]) -> int:
+    """The tag with the most image associations (→ fewest FK repoints when it
+    survives), tie-broken to the lowest id for determinism. Module-level so the
+    key closes over its parameter, not a loop variable (ruff B023)."""
+    return max(tag_ids, key=lambda tid: (counts.get(tid, 0), -tid))
+
+
+async def normalize_existing_tags(
+    session: AsyncSession, *, dry_run: bool = False
+) -> dict:
+    """Convert the back-catalog to the #701 canonical tag form.
+
+    For each (kind, fandom, canonical) group: pick a survivor, merge any
+    case/whitespace-variant siblings INTO it via the tested merge path
+    (TagService._do_merge — FK repoints + protective aliases), then rename the
+    survivor to the canonical form. Idempotent: a group that is already a lone
+    canonical tag is a no-op, so re-running is safe.
+
+    dry_run=True returns a projection (counts + a sample of the changes) with no
+    mutations. Live runs commit per group and isolate failures per group so one
+    bad group can't strand the rest.
+
+    Returns (dry_run):
+      {"groups": N, "collisions": M, "tags_to_merge": K, "tags_to_rename": R,
+       "total_changes": T, "sample": [{"to", "from": [...], "kind", "merge"}]}
+    Returns (live):
+      {"groups_processed", "merged", "renamed", "aliases_created", "errors",
+       "sample": [...]}
+    """
+    rows = (
+        await session.execute(
+            select(Tag.id, Tag.name, Tag.kind, Tag.fandom_id)
+        )
+    ).all()
+    groups = _group_existing_tags(rows)
+
+    # Deterministic sample/ordering: by kind then canonical name.
+    touched = sorted(
+        (
+            (key, members)
+            for key, members in groups.items()
+            if _group_needs_change(key[2], members)
+        ),
+        key=lambda km: (
+            km[0][0].value if hasattr(km[0][0], "value") else str(km[0][0]),
+            km[0][2].lower(),
+        ),
+    )
+
+    sample = [
+        {
+            "to": key[2],
+            "from": [name for _id, name in members],
+            "kind": key[0].value if hasattr(key[0], "value") else str(key[0]),
+            "merge": len(members) > 1,
+        }
+        for key, members in touched[:_NORMALIZE_SAMPLE_CAP]
+    ]
+
+    if dry_run:
+        collisions = sum(1 for key, m in touched if len(m) > 1)
+        tags_to_merge = sum(len(m) - 1 for key, m in touched if len(m) > 1)
+        # A group renames iff the canonical form isn't already one of its
+        # members' exact names (else that member is picked as survivor → no
+        # rename, the rest merge into it).
+        tags_to_rename = sum(
+            1
+            for key, m in touched
+            if key[2] not in {name for _id, name in m}
+        )
+        return {
+            "groups": len(groups),
+            "collisions": collisions,
+            "tags_to_merge": tags_to_merge,
+            "tags_to_rename": tags_to_rename,
+            "total_changes": len(touched),
+            "sample": sample,
+        }
+
+    svc = TagService(session)
+    summary = {
+        "groups_processed": 0,
+        "merged": 0,
+        "renamed": 0,
+        "aliases_created": 0,
+        "errors": 0,
+        "sample": sample,
+    }
+    for key, members in touched:
+        canonical = key[2]
+        names_by_id = {tag_id: name for tag_id, name in members}
+        # Survivor: prefer a member already named canonically (no rename, no
+        # self-alias); else the best-connected (fewest FK repoints); else
+        # lowest id for determinism.
+        survivor_id = next(
+            (tid for tid, name in members if name == canonical), None
+        )
+        if survivor_id is None:
+            counts = await svc._image_assoc_counts(list(names_by_id))
+            survivor_id = _best_connected(list(names_by_id), counts)
+        loser_ids = [tid for tid in names_by_id if tid != survivor_id]
+        try:
+            survivor = await session.get(Tag, survivor_id)
+            for loser_id in loser_ids:
+                loser = await session.get(Tag, loser_id)
+                if loser is None:
+                    continue
+                result = await svc._do_merge(loser, survivor)
+                if result.alias_created:
+                    summary["aliases_created"] += 1
+            if survivor.name != canonical:
+                survivor.name = canonical
+                await session.flush()
+                summary["renamed"] += 1
+            await session.commit()
+            summary["groups_processed"] += 1
+            summary["merged"] += len(loser_ids)
+        except Exception as exc:  # one bad group must not strand the rest
+            await session.rollback()
+            summary["errors"] += 1
+            log.warning(
+                "tag normalize failed for group %r: %s", canonical, exc
+            )
+    return summary

backend/app/tasks/admin.py

@@ -73,3 +73,32 @@ def reextract_archive_attachments_task(self) -> dict:
         return cleanup_service.reextract_archive_attachments(
             session, images_root=IMAGES_ROOT,
         )
+
+
+@celery.task(
+    name="backend.app.tasks.admin.normalize_tags_task",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=15, retry_backoff_max=180, max_retries=1,
+    soft_time_limit=1800, time_limit=2400,  # 30 min / 40 min
+)
+def normalize_tags_task(self) -> dict:
+    """Wraps tag_service.normalize_existing_tags (#714): Title-Case the
+    back-catalog and merge case/whitespace-variant duplicate tags via the
+    tested async merge path. Runs under its own asyncio loop + per-task async
+    engine (NullPool, disposed when the loop ends), mirroring download_source."""
+    import asyncio
+
+    from ..services.tag_service import normalize_existing_tags
+    from ._async_session import async_session_factory
+
+    async def _run() -> dict:
+        async_factory, async_engine = async_session_factory()
+        try:
+            async with async_factory() as session:
+                # normalize_existing_tags commits per group internally.
+                return await normalize_existing_tags(session, dry_run=False)
+        finally:
+            await async_engine.dispose()
+
+    return asyncio.run(_run())