Blame for backend/app/routers/topics.py

"""Topic map: topics as stars, co-occurrence as constellation lines, plus the insight a writer needs:
which topics are rising, which went dormant, which travel together, and how each one evolved."""

import uuid
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from itertools import combinations

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select

from app.auth import Ctx, get_ctx
from app.models import Document, DocumentTopic, Topic
from app.routers.sources import serialize_document
from app.services.jobs import create_job, serialize_job

router = APIRouter(prefix="/api/topics", tags=["topics"])

BUCKETS = 12  # timeline resolution across the archive's lifetime


def _aware(dt: datetime | None) -> datetime | None:
    return dt if dt is None or dt.tzinfo else dt.replace(tzinfo=UTC)


@dataclass
class TopicStats:
    topic: Topic
    dates: list[datetime] = field(default_factory=list)
    docs: set[uuid.UUID] = field(default_factory=set)


@dataclass
class Archive:
    start: datetime
    end: datetime
    recent_from: datetime
    total_docs: int
    recent_docs: int

    def bucket(self, dt: datetime) -> int:
        span = (self.end - self.start).total_seconds() or 1
        return min(BUCKETS - 1, max(0, int((dt - self.start).total_seconds() / span * BUCKETS)))


def _load(ctx: Ctx, topic_ids: set[uuid.UUID] | None = None):
    """Topic -> dated posts, and the archive's time frame."""
    q = (
        select(DocumentTopic.topic_id, DocumentTopic.document_id, Document.published_at, Document.created_at)
        .join(Document, Document.id == DocumentTopic.document_id)
        .where(Document.workspace_id == ctx.workspace.id)
    )
    if topic_ids is not None:
        q = q.where(DocumentTopic.topic_id.in_(topic_ids))
    rows = ctx.db.execute(q).all()
    all_dates = [
        _aware(p or c)
        for p, c in ctx.db.execute(
            select(Document.published_at, Document.created_at).where(
                Document.workspace_id == ctx.workspace.id, Document.path.is_not(None))
        ).all()
    ]
    now = datetime.now(UTC)
    start = min(all_dates, default=now)
    end = max(all_dates, default=now)
    span = end - start
    # "Recent" is the last quarter of the writer's own publishing history (at least 60 days), so an archive
    # that stopped in 2022 still shows what was rising at the end of it.
    window = max(span / 4, timedelta(days=60))
    recent_from = end - window
    archive = Archive(start, end, recent_from, len(all_dates), sum(1 for d in all_dates if d >= recent_from))
    return rows, archive


def _status(count: int, recent: int, last: datetime | None, archive: Archive) -> tuple[str, float]:
    """(status, momentum). Momentum compares the topic's share of recent posts to its share of all posts."""
    overall = count / max(archive.total_docs, 1)
    recent_share = recent / max(archive.recent_docs, 1)
    momentum = round(recent_share / overall, 2) if overall else 0.0
    if recent == 0 and last and last < archive.recent_from:
        return "dormant", momentum
    if momentum >= 1.4 and recent >= 2:
        return "rising", momentum
    return "steady", momentum


def _galaxies(node_ids: list[str], edges: Counter) -> dict[str, int]:
    """Weighted label propagation: topics that share posts settle into the same galaxy."""
    label = {n: i for i, n in enumerate(node_ids)}
    nbrs: dict[str, list[tuple[str, int]]] = defaultdict(list)
    for (a, b), w in edges.items():
        nbrs[a].append((b, w))
        nbrs[b].append((a, w))
    for _ in range(12):
        changed = False
        for n in node_ids:  # deterministic order: biggest topics first
            if not nbrs[n]:
                continue
            score: Counter = Counter()
            for m, w in nbrs[n]:
                score[label[m]] += w
            best = max(score.items(), key=lambda kv: (kv[1], -kv[0]))[0]
            if score[best] > score.get(label[n], 0) and best != label[n]:
                label[n] = best
                changed = True
        if not changed:
            break
    return label


@router.get("")
def topic_map(ctx: Ctx = Depends(get_ctx), min_posts: int = 1, limit: int = 80):
    topics = ctx.db.scalars(
        select(Topic).where(Topic.workspace_id == ctx.workspace.id, Topic.post_count >= min_posts)
        .order_by(Topic.post_count.desc(), Topic.name).limit(min(limit, 300))
    ).all()
    by_id = {t.id: TopicStats(t) for t in topics}
    rows, archive = _load(ctx, set(by_id)) if by_id else ([], None)

    by_doc: dict[uuid.UUID, list[str]] = defaultdict(list)
    for topic_id, doc_id, published, created in rows:
        st = by_id[topic_id]
        if doc_id not in st.docs:
            st.docs.add(doc_id)
            st.dates.append(_aware(published or created))
            by_doc[doc_id].append(str(topic_id))
    edges: Counter[tuple[str, str]] = Counter()
    for tids in by_doc.values():
        for a, b in combinations(sorted(set(tids)), 2):
            edges[(a, b)] += 1

    node_ids = [str(t.id) for t in topics]
    galaxy_of = _galaxies(node_ids, edges)
    nodes = []
    for t in topics:
        st = by_id[t.id]
        dates = sorted(st.dates)
        count = len(dates) or t.post_count
        recent = sum(1 for d in dates if archive and d >= archive.recent_from)
        first, last = (dates[0], dates[-1]) if dates else (None, None)
        status, momentum = _status(count, recent, last, archive) if archive else ("steady", 1.0)
        timeline = [0] * BUCKETS
        for d in dates:
            timeline[archive.bucket(d)] += 1
        span = (archive.end - archive.start).total_seconds() if archive else 0
        recency = round((last - archive.start).total_seconds() / span, 3) if last and span else 1.0
        nodes.append({
            "id": str(t.id), "name": t.name, "summary": t.summary, "post_count": count,
            "first_at": first.isoformat() if first else None, "last_at": last.isoformat() if last else None,
            "recent_posts": recent, "momentum": momentum, "status": status, "recency": recency,
            "timeline": timeline, "galaxy": galaxy_of[str(t.id)],
        })

    # Galaxies: named after their brightest (most written about) star.
    members: dict[int, list[dict]] = defaultdict(list)
    for n in nodes:
        members[n["galaxy"]].append(n)
    galaxies = []
    for gid, ns in sorted(members.items(), key=lambda kv: -sum(n["post_count"] for n in kv[1])):
        lead = max(ns, key=lambda n: n["post_count"])
        galaxies.append({"id": gid, "name": lead["name"], "topic_ids": [n["id"] for n in ns],
                         "post_count": sum(n["post_count"] for n in ns), "size": len(ns)})
    names = {n["id"]: n["name"] for n in nodes}
    rising = sorted((n for n in nodes if n["status"] == "rising"), key=lambda n: -n["momentum"])[:5]
    dormant = sorted((n for n in nodes if n["status"] == "dormant"), key=lambda n: -n["post_count"])[:5]
    untagged = ctx.db.scalar(
        select(Document.id).where(Document.workspace_id == ctx.workspace.id, Document.path.is_not(None),
                                  Document.id.not_in(select(DocumentTopic.document_id))).limit(1)
    )
    return {
        "nodes": nodes,
        "edges": [{"source": a, "target": b, "weight": w} for (a, b), w in edges.most_common(400)],
        "galaxies": galaxies,
        "insights": {
            "rising": [{"id": n["id"], "name": n["name"], "momentum": n["momentum"]} for n in rising],
            "dormant": [{"id": n["id"], "name": n["name"], "last_at": n["last_at"], "post_count": n["post_count"]}
                        for n in dormant],
            "pairs": [{"a": names[a], "b": names[b], "a_id": a, "b_id": b, "posts": w}
                      for (a, b), w in edges.most_common(5)],
        },
        "archive": {
            "start": archive.start.isoformat() if archive else None,
            "end": archive.end.isoformat() if archive else None,
            "recent_from": archive.recent_from.isoformat() if archive else None,
            "posts": archive.total_docs if archive else 0,
        },
        "has_untagged_posts": untagged is not None,
    }


@router.get("/{topic_id}")
def topic_detail(topic_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
    topic = ctx.db.scalar(select(Topic).where(Topic.id == topic_id, Topic.workspace_id == ctx.workspace.id))
    if not topic:
        raise HTTPException(404, "Topic not found")
    rows = ctx.db.execute(
        select(Document, DocumentTopic.weight)
        .join(DocumentTopic, DocumentTopic.document_id == Document.id)
        .where(DocumentTopic.topic_id == topic.id)
        .order_by(DocumentTopic.weight.desc(), Document.published_at.desc())
    ).all()
    doc_ids = [d.id for d, _ in rows]
    _, archive = _load(ctx, {topic.id})
    dates = sorted(_aware(d.published_at or d.created_at) for d, _ in rows)
    recent = sum(1 for d in dates if d >= archive.recent_from)
    status, momentum = _status(len(dates), recent, dates[-1] if dates else None, archive)
    timeline = [0] * BUCKETS
    for d in dates:
        timeline[archive.bucket(d)] += 1
    # Topics that share posts with this one, strongest first.
    related = Counter()
    if doc_ids:
        for other_id, in ctx.db.execute(
            select(DocumentTopic.topic_id).where(DocumentTopic.document_id.in_(doc_ids),
                                                 DocumentTopic.topic_id != topic.id)
        ).all():
            related[other_id] += 1
    related_topics = {t.id: t for t in ctx.db.scalars(select(Topic).where(Topic.id.in_(list(related))))} \
        if related else {}
    return {
        "id": str(topic.id), "name": topic.name, "summary": topic.summary, "post_count": len(dates),
        "first_at": dates[0].isoformat() if dates else None, "last_at": dates[-1].isoformat() if dates else None,
        "recent_posts": recent, "momentum": momentum, "status": status, "timeline": timeline,
        "timeline_start": archive.start.isoformat(), "timeline_end": archive.end.isoformat(),
        "related": [{"id": str(tid), "name": related_topics[tid].name, "shared_posts": n}
                    for tid, n in related.most_common(6) if tid in related_topics],
        "posts": [{**serialize_document(d), "weight": w} for d, w in rows],
    }


@router.post("/rebuild")
def rebuild(ctx: Ctx = Depends(get_ctx), full: bool = False):
    if full:
        # Clear per post tags so every post is mapped again.
        for d in ctx.db.scalars(select(Document).where(Document.workspace_id == ctx.workspace.id)):
            if d.metadata_json and ({"topics", "topics_failures"} & d.metadata_json.keys()):
                d.metadata_json = {k: v for k, v in d.metadata_json.items() if k not in ("topics", "topics_failures")}
        ctx.db.commit()
    job = create_job(ctx.db, ctx.workspace.id, "topics", {}, max_attempts=2)
    return serialize_job(job)