feat(api): agent-friendly tag analysis endpoints — /tags/top + /tags/<id>/stats (#1136)

Fast, read-only, indexed aggregates shaped for ANALYSIS (not the paged UI
directory, which is alphabetical + builds previews and timed out at 10 min on a
full count sweep).

- GET /api/tags/top — top tags by image count, desc. ?kind, ?limit (cap 500),
  ?min_count, ?source=all|human|manual|accepted|auto (human=manual+ml_accepted,
  auto=head_auto+ccip_auto+ml_auto). One GROUP BY over image_tag (indexed on
  tag_id).
- GET /api/tags/<id>/stats — per-tag dataset health: total + per-source counts
  (manual/accepted/head_auto/ccip_auto), human vs auto rollups, rejection count,
  and whether a trained head exists. Backs concept-readiness + source-split
  analysis.

Plain-HTTP homelab posture, no auth change. Tests cover ranking, source filter,
min_count, the source breakdown, and 404.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa

backend/app/api/tags.py

@@ -1,11 +1,14 @@
 """Tags API: autocomplete, create, list/add/remove for an image."""
 
 from quart import Blueprint, jsonify, request
+from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.exc import IntegrityError
 
 from ..extensions import get_session
-from ..models import Tag, TagKind, TagPositiveConfirmation
+from ..models import Tag, TagHead, TagKind, TagPositiveConfirmation
+from ..models.tag import image_tag
+from ..models.tag_suggestion_rejection import TagSuggestionRejection
 from ..services.bulk_tag_service import BulkTagService
 from ..services.ml.aliases import AliasService
 from ..services.series_match_service import SeriesMatchService
@@ -59,6 +62,117 @@ def _parse_bulk_ids(
     return ids, None
 
 
+# Application-source groupings (image_tag.source). HUMAN = operator signal;
+# AUTO = machine-applied (heads/CCIP, + legacy Camie ml_auto).
+_SOURCE_GROUPS = {
+    "human": ("manual", "ml_accepted"),
+    "manual": ("manual",),
+    "accepted": ("ml_accepted",),
+    "auto": ("head_auto", "ccip_auto", "ml_auto"),
+}
+
+
+@tags_bp.route("/tags/top", methods=["GET"])
+async def tags_top():
+    """Top tags by image count — a fast indexed aggregate for ANALYSIS (not the
+    paged UI directory, which is alphabetical + builds previews). Params:
+    ?kind=general|character|fandom|… ?source=all|human|manual|accepted|auto
+    ?limit=50 (cap 500) ?min_count=N. → {tags:[{tag_id,name,kind,count}]} desc."""
+    kind = _coerce_kind(request.args.get("kind"))
+    try:
+        limit = min(max(int(request.args.get("limit", "50")), 1), 500)
+    except ValueError:
+        return jsonify({"error": "limit must be an integer"}), 400
+    min_count = None
+    if "min_count" in request.args:
+        try:
+            min_count = int(request.args["min_count"])
+        except ValueError:
+            return jsonify({"error": "min_count must be an integer"}), 400
+    src_vals = _SOURCE_GROUPS.get((request.args.get("source") or "all").lower())
+
+    cnt = func.count(image_tag.c.image_record_id)
+    stmt = (
+        select(Tag.id, Tag.name, Tag.kind, cnt.label("count"))
+        .select_from(Tag)
+        .join(image_tag, image_tag.c.tag_id == Tag.id)
+        .group_by(Tag.id, Tag.name, Tag.kind)
+        .order_by(cnt.desc(), Tag.name.asc())
+        .limit(limit)
+    )
+    if kind is not None:
+        stmt = stmt.where(Tag.kind == kind)
+    if src_vals is not None:
+        stmt = stmt.where(image_tag.c.source.in_(src_vals))
+    if min_count is not None:
+        stmt = stmt.having(cnt >= min_count)
+    async with get_session() as session:
+        rows = (await session.execute(stmt)).all()
+    return jsonify({"tags": [
+        {
+            "tag_id": r.id, "name": r.name,
+            "kind": r.kind.value if hasattr(r.kind, "value") else str(r.kind),
+            "count": r.count,
+        }
+        for r in rows
+    ]})
+
+
+@tags_bp.route("/tags/<int:tag_id>/stats", methods=["GET"])
+async def tag_stats(tag_id: int):
+    """Per-tag dataset health: total + per-source application counts (human vs
+    machine), rejection count, and whether a trained head exists. Read-only,
+    analysis-shaped — backs concept-readiness + source-split decisions."""
+    async with get_session() as session:
+        tag = await session.get(Tag, tag_id)
+        if tag is None:
+            return jsonify({"error": "not found"}), 404
+        by_source = {
+            src: n for src, n in (
+                await session.execute(
+                    select(image_tag.c.source, func.count())
+                    .where(image_tag.c.tag_id == tag_id)
+                    .group_by(image_tag.c.source)
+                )
+            ).all()
+        }
+        rejected = (
+            await session.execute(
+                select(func.count())
+                .select_from(TagSuggestionRejection)
+                .where(TagSuggestionRejection.tag_id == tag_id)
+            )
+        ).scalar_one()
+        has_head = (
+            await session.execute(
+                select(func.count())
+                .select_from(TagHead)
+                .where(TagHead.tag_id == tag_id)
+            )
+        ).scalar_one() > 0
+    human = by_source.get("manual", 0) + by_source.get("ml_accepted", 0)
+    auto = (
+        by_source.get("head_auto", 0)
+        + by_source.get("ccip_auto", 0)
+        + by_source.get("ml_auto", 0)
+    )
+    return jsonify({
+        "tag_id": tag_id,
+        "name": tag.name,
+        "kind": tag.kind.value if hasattr(tag.kind, "value") else str(tag.kind),
+        "count_total": sum(by_source.values()),
+        "count_human": human,
+        "count_manual": by_source.get("manual", 0),
+        "count_accepted": by_source.get("ml_accepted", 0),
+        "count_auto": auto,
+        "count_head_auto": by_source.get("head_auto", 0),
+        "count_ccip_auto": by_source.get("ccip_auto", 0),
+        "count_rejected": rejected,
+        "by_source": by_source,
+        "has_head": has_head,
+    })
+
+
 @tags_bp.route("/tags/autocomplete", methods=["GET"])
 async def autocomplete():
     q = request.args.get("q", "")