backend/fetlife_data.py

"""Backend fetlife data service."""
from __future__ import annotations

import json
from collections import defaultdict
from pathlib import Path
from typing import Any
from sqlalchemy import func
from sqlmodel import Session, select
from models import (Alias, Asset, Definition, FetlifeKinkMeta, FetlifePictureRef,
    FetlifeUserFetish, FetlifeUserSample, Kink, KinkContentType, KinkTypeOverride,
    SimilarityEdge)

from backend.recsys_graph import invalidate_similarity_graph_cache


def upsert_fetlife_kink_meta(
    self,
    kink_id: str,
    fetish_id: str,
    source_url: str,
    counts: dict[str, str],
    *,
    has_real_images: bool = False,
    similar_count: int = 0,
) -> None:
    counts_json = json.dumps(counts, sort_keys=True)
    popularity = self._parse_count_value(counts.get("Kinksters", "")) or self._parse_count_value(counts.get("kinksters", ""))
    with Session(self.engine) as session:
        meta = session.get(FetlifeKinkMeta, kink_id)
        if meta:
            meta.fetish_id = fetish_id
            meta.source_url = source_url
            meta.counts_json = counts_json
            if popularity:
                meta.popularity = popularity
            meta.has_real_images = meta.has_real_images or has_real_images
            meta.similar_count = max(meta.similar_count, similar_count)
            meta.updated_at = self._now_iso()
        else:
            session.add(
                FetlifeKinkMeta(
                    kink_id=kink_id,
                    fetish_id=fetish_id,
                    source_url=source_url,
                    counts_json=counts_json,
                    popularity=popularity,
                    has_real_images=has_real_images,
                    similar_count=similar_count,
                    updated_at=self._now_iso(),
                )
            )
        session.commit()
    self._invalidate_catalog_cache()

def fetlife_source_coverage(self, limit: int = 50) -> list[dict[str, Any]]:
    with Session(self.engine) as session:
        meta_rows = session.exec(select(FetlifeKinkMeta).order_by(FetlifeKinkMeta.popularity.desc())).all()
        kink_ids = [row.kink_id for row in meta_rows[:limit]]
        kink_rows = session.exec(select(Kink).where(Kink.id.in_(kink_ids))).all() if kink_ids else []
        type_rows = session.exec(select(KinkContentType).where(KinkContentType.kink_id.in_(kink_ids))).all() if kink_ids else []
        asset_rows = session.exec(select(Asset).where(Asset.kink_id.in_(kink_ids))).all() if kink_ids else []
        picture_ref_rows = (
            session.exec(select(FetlifePictureRef).where(FetlifePictureRef.kink_id.in_(kink_ids))).all()
            if kink_ids
            else []
        )
        kink_by_id = {row.id: row for row in kink_rows}
        type_by_id = {row.kink_id: row for row in type_rows}
        asset_count_by_id: dict[str, int] = defaultdict(int)
        asset_attachment_ids_by_kink: dict[str, set[str]] = defaultdict(set)
        for row in asset_rows:
            asset_count_by_id[row.kink_id] += 1
            if row.source_id == "fetlife_fetish_pages":
                asset_attachment_ids_by_kink[row.kink_id].add(Path(row.asset_url).stem)
    reuse_counts: dict[str, int] = defaultdict(int)
    for row in picture_ref_rows:
        reuse_counts[row.attachment_id] += 1
    filtered_asset_ids_by_kink: dict[str, set[str]] = defaultdict(set)
    suppressed_ids_by_kink: dict[str, set[str]] = defaultdict(set)
    for row in picture_ref_rows:
        reuse_count = reuse_counts.get(row.attachment_id, 1)
        if row.attachment_id not in asset_attachment_ids_by_kink.get(row.kink_id, set()):
            continue
        if reuse_count > 8:
            suppressed_ids_by_kink[row.kink_id].add(row.attachment_id)
            continue
        filtered_asset_ids_by_kink[row.kink_id].add(row.attachment_id)
    items = []
    for row in meta_rows[:limit]:
        kink = kink_by_id.get(row.kink_id)
        if not kink:
            continue
        type_row = type_by_id.get(row.kink_id)
        derived = self._derived_product_flags(
            {
                "name": kink.name,
                "cluster": kink.cluster,
                "content_kind": type_row.content_kind if type_row else self._derive_content_kind(kink.name, kink.cluster)[0],
                "definition": kink.short_definition,
                "summary": kink.short_definition,
                "examples": [],
                "popularity": row.popularity,
                "source_backed_popularity": row.popularity,
                "raw_asset_count": asset_count_by_id.get(row.kink_id, 0),
                "filtered_asset_count": len(filtered_asset_ids_by_kink.get(row.kink_id, set())),
                "similar_count": row.similar_count,
            }
        )
        items.append(
            {
                "kink_id": row.kink_id,
                "name": kink.name,
                "cluster": kink.cluster,
                "content_kind": type_row.content_kind if type_row else self._derive_content_kind(kink.name, kink.cluster)[0],
                "type_evidence": type_row.evidence if type_row else "",
                "fetish_id": row.fetish_id,
                "popularity": row.popularity,
                "has_real_images": row.has_real_images,
                "asset_count": asset_count_by_id.get(row.kink_id, 0),
                "filtered_asset_count": len(filtered_asset_ids_by_kink.get(row.kink_id, set())),
                "image_trust_state": derived["image_trust_state"],
                "starter_tier": derived["starter_tier"],
                "starter_eligible": derived["starter_eligible"],
                "shared_eligible": derived["shared_eligible"],
                "prompt_eligible": derived["prompt_eligible"],
                "similar_count": row.similar_count,
                "suppressed_attachment_examples": [
                    {"attachment_id": attachment_id, "reuse_count": reuse_counts.get(attachment_id, 1)}
                    for attachment_id in sorted(suppressed_ids_by_kink.get(row.kink_id, set()))[:5]
                ],
                "source_url": row.source_url,
                "updated_at": row.updated_at,
            }
        )
    return items

def fetlife_coverage_debt(self, limit: int = 12) -> dict[str, list[dict[str, Any]]]:
    type_map = self._content_type_map()
    with Session(self.engine) as session:
        rows = session.exec(select(FetlifeKinkMeta).order_by(FetlifeKinkMeta.popularity.desc())).all()
        kink_ids = [row.kink_id for row in rows[:1000]]
        kink_rows = session.exec(select(Kink).where(Kink.id.in_(kink_ids))).all() if kink_ids else []
    kink_by_id = {row.id: row for row in kink_rows}
    missing_images: list[dict[str, Any]] = []
    missing_similar: list[dict[str, Any]] = []
    starter_priority: list[dict[str, Any]] = []
    for row in rows:
        kink = kink_by_id.get(row.kink_id)
        if not kink:
            continue
        kind = type_map.get(row.kink_id, {}).get("content_kind", self._derive_content_kind(kink.name, kink.cluster)[0])
        if kind != "play":
            continue
        derived = self._derived_product_flags(
            {
                "name": kink.name,
                "cluster": kink.cluster,
                "content_kind": kind,
                "definition": kink.short_definition,
                "summary": kink.short_definition,
                "examples": [],
                "popularity": row.popularity,
                "source_backed_popularity": row.popularity,
                "filtered_asset_count": 1 if row.has_real_images else 0,
                "similar_count": row.similar_count,
            }
        )
        item = {
            "kink_id": row.kink_id,
            "name": kink.name,
            "fetish_id": row.fetish_id,
            "popularity": row.popularity,
            "starter_tier": derived["starter_tier"],
            "source_url": row.source_url,
            "needs_images": not row.has_real_images,
            "needs_similar": row.similar_count <= 0,
        }
        if not row.has_real_images and len(missing_images) < limit:
            missing_images.append(item)
        if row.similar_count <= 0 and len(missing_similar) < limit:
            missing_similar.append(item)
        if (
            derived["starter_tier"] in {"starter_core", "starter_later"}
            and (not row.has_real_images or row.similar_count <= 0)
            and len(starter_priority) < limit
        ):
            starter_priority.append(item)
        if len(missing_images) >= limit and len(missing_similar) >= limit and len(starter_priority) >= limit:
            break
    return {"starter_priority": starter_priority, "missing_images": missing_images, "missing_similar": missing_similar}

def fetlife_data_health(self, limit: int = 20) -> dict[str, Any]:
    debt = self.fetlife_coverage_debt(limit=limit)
    with Session(self.engine) as session:
        meta_total = int(session.exec(select(func.count()).select_from(FetlifeKinkMeta)).one())
        image_total = int(session.exec(select(func.count()).select_from(FetlifeKinkMeta).where(FetlifeKinkMeta.has_real_images == True)).one())
        similar_total = int(session.exec(select(func.count()).select_from(FetlifeKinkMeta).where(FetlifeKinkMeta.similar_count > 0)).one())
        picture_refs = int(session.exec(select(func.count()).select_from(FetlifePictureRef)).one())
        sampled_users = int(session.exec(select(func.count()).select_from(FetlifeUserSample)).one())
        sampled_fetishes = int(session.exec(select(func.count()).select_from(FetlifeUserFetish)).one())
        collab_edges = int(
            session.exec(
                select(func.count()).select_from(SimilarityEdge).where(SimilarityEdge.similarity_type == "fetlife_collab")
            ).one()
        )
        asset_total = int(session.exec(select(func.count()).select_from(Asset)).one())
        representative = int(session.exec(select(func.count(func.distinct(Asset.kink_id))).select_from(Asset)).one())
    return {
        "coverage": {
            "fetlife_meta": meta_total,
            "with_images": image_total,
            "with_similar": similar_total,
            "picture_refs": picture_refs,
            "assets": asset_total,
            "representative_image_kinks": representative,
            "sampled_users": sampled_users,
            "sampled_fetishes": sampled_fetishes,
            "collab_edges": collab_edges,
        },
        "debt": debt,
        "crawl": {
            "jobs": self.fetlife_job_stats(),
            "supervisor": self.fetlife_supervisor_status(),
        },
        "queue": {
            "next": self.fetlife_queue_preview(limit=limit),
        },
        "recommendation_readiness": {
            "starter_core_missing": len(debt.get("starter_priority", [])),
            "collab_ready": sampled_users >= 250 and collab_edges > 0,
            "image_ready_ratio": round(image_total / meta_total, 3) if meta_total else 0.0,
            "similar_ready_ratio": round(similar_total / meta_total, 3) if meta_total else 0.0,
        },
    }

def content_type_coverage(self, limit: int = 20) -> dict[str, Any]:
    with Session(self.engine) as session:
        type_rows = session.exec(select(KinkContentType)).all()
        kink_rows = session.exec(select(Kink)).all()
        overrides = session.exec(select(KinkTypeOverride).order_by(KinkTypeOverride.cluster, KinkTypeOverride.normalized_name)).all()
    kink_by_id = {row.id: row for row in kink_rows}
    by_kind: dict[str, int] = defaultdict(int)
    by_cluster: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    samples: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in type_rows:
        kink = kink_by_id.get(row.kink_id)
        if not kink:
            continue
        by_kind[row.content_kind] += 1
        by_cluster[kink.cluster][row.content_kind] += 1
        if len(samples[row.content_kind]) < limit:
            samples[row.content_kind].append(
                {
                    "kink_id": row.kink_id,
                    "name": kink.name,
                    "cluster": kink.cluster,
                    "evidence": row.evidence,
                }
            )
    cluster_rows = []
    for cluster, counts in by_cluster.items():
        cluster_rows.append({"cluster": cluster, **counts, "total": sum(counts.values())})
    cluster_rows.sort(key=lambda item: item["total"], reverse=True)
    return {
        "by_kind": dict(sorted(by_kind.items())),
        "by_cluster": cluster_rows[:limit * 2],
        "samples": dict(samples),
        "overrides": [
            {
                "id": row.id,
                "normalized_name": row.normalized_name,
                "cluster": row.cluster,
                "content_kind": row.content_kind,
                "note": row.note,
            }
            for row in overrides
        ],
    }

def upsert_fetlife_user_sample(
    self,
    nickname: str,
    *,
    profile_url: str,
    sampled_from_fetish_id: str,
    state: str = "ok",
) -> None:
    with Session(self.engine) as session:
        row = session.get(FetlifeUserSample, nickname)
        if row:
            row.profile_url = profile_url
            row.sampled_from_fetish_id = sampled_from_fetish_id
            row.state = state
            row.updated_at = self._now_iso()
        else:
            session.add(
                FetlifeUserSample(
                    nickname=nickname,
                    profile_url=profile_url,
                    sampled_from_fetish_id=sampled_from_fetish_id,
                    state=state,
                    updated_at=self._now_iso(),
                )
            )
        session.commit()

def replace_fetlife_user_fetishes(
    self,
    nickname: str,
    *,
    profile_url: str,
    sampled_from_fetish_id: str,
    fetishes_by_bucket: dict[str, list[dict[str, Any]]],
) -> None:
    with Session(self.engine) as session:
        seen_user_fetishes: set[tuple[str, str]] = set()
        sample = session.get(FetlifeUserSample, nickname)
        if sample:
            sample.profile_url = profile_url
            sample.sampled_from_fetish_id = sampled_from_fetish_id
            sample.state = "ok"
            sample.updated_at = self._now_iso()
        else:
            session.add(
                FetlifeUserSample(
                    nickname=nickname,
                    profile_url=profile_url,
                    sampled_from_fetish_id=sampled_from_fetish_id,
                    state="ok",
                    updated_at=self._now_iso(),
                )
            )
        existing = session.exec(select(FetlifeUserFetish).where(FetlifeUserFetish.nickname == nickname)).all()
        for row in existing:
            session.delete(row)
        for bucket, rows in fetishes_by_bucket.items():
            for item in rows:
                fetish_id = str(item.get("id", "")).strip()
                if not fetish_id:
                    continue
                dedupe_key = (nickname, fetish_id)
                if dedupe_key in seen_user_fetishes:
                    continue
                seen_user_fetishes.add(dedupe_key)
                kink_id = f"fetlife_{fetish_id}"
                name = str(item.get("name", "")).strip() or f"Fetish {fetish_id}"
                if "\n" in name:
                    name = name.split("\n", 1)[0].strip() or f"Fetish {fetish_id}"
                activity = str(item.get("activity", "")).strip()
                kink = session.get(Kink, kink_id)
                if kink:
                    if "\n" in kink.name:
                        kink.name = kink.name.split("\n", 1)[0].strip()
                else:
                    session.add(
                        Kink(
                            id=kink_id,
                            name=name,
                            cluster="fetlife_fetish",
                            short_definition="",
                            notes="",
                            risk_level="general",
                            is_extreme=False,
                        )
                    )
                session.add(
                    FetlifeUserFetish(
                        nickname=nickname,
                        fetish_id=fetish_id,
                        kink_id=kink_id,
                        bucket=bucket,
                        activity=activity,
                        updated_at=self._now_iso(),
                    )
                )
        session.commit()
    self._invalidate_catalog_cache()

def rebuild_fetlife_collaborative_edges(self, min_support: int = 2, max_neighbors: int = 64) -> None:
    positive_buckets = ("into", "curious_about")
    with self._sqlite() as conn:
        conn.execute("DELETE FROM similarityedge WHERE similarity_type = 'fetlife_collab'")
        conn.execute("DROP TABLE IF EXISTS temp.collab_pair_counts")
        conn.execute("DROP TABLE IF EXISTS temp.collab_directed")
        conn.execute(
            """
            CREATE TEMP TABLE collab_pair_counts AS
            SELECT
                a.kink_id AS left_kink_id,
                b.kink_id AS right_kink_id,
                COUNT(*) AS support
            FROM fetlifeuserfetish a
            JOIN kink ka ON ka.id = a.kink_id
            JOIN fetlifeuserfetish b
              ON a.nickname = b.nickname
             AND a.kink_id < b.kink_id
            JOIN kink kb ON kb.id = b.kink_id
            WHERE a.bucket IN (?, ?)
              AND b.bucket IN (?, ?)
            GROUP BY a.kink_id, b.kink_id
            """,
            positive_buckets + positive_buckets,
        )
        max_support_row = conn.execute(
            "SELECT COALESCE(MAX(support), 1) AS max_support FROM collab_pair_counts"
        ).fetchone()
        max_support = int(max_support_row["max_support"] or 1)
        conn.execute(
            """
            CREATE TEMP TABLE collab_directed AS
            SELECT left_kink_id, right_kink_id, support FROM collab_pair_counts
            UNION ALL
            SELECT right_kink_id AS left_kink_id, left_kink_id AS right_kink_id, support
            FROM collab_pair_counts
            """
        )
        conn.execute(
            """
            INSERT INTO similarityedge (
                id, left_kink_id, right_kink_id, similarity_type, score, method, version
            )
            SELECT
                'sim_fetlife_collab_' || left_kink_id || '_' || right_kink_id,
                left_kink_id,
                right_kink_id,
                similarity_type,
                score,
                method,
                version
            FROM (
                SELECT
                    left_kink_id,
                    right_kink_id,
                    'fetlife_collab' AS similarity_type,
                    CAST(support AS REAL) / ? AS score,
                    'fetlife_user_sample:' || support AS method,
                    'v1' AS version,
                    ROW_NUMBER() OVER (
                        PARTITION BY left_kink_id
                        ORDER BY support DESC, right_kink_id
                    ) AS neighbor_rank
                FROM collab_directed
                WHERE support >= ?
            )
            WHERE neighbor_rank <= ?
            """,
            (max_support, min_support, max_neighbors),
        )
        conn.execute("DROP TABLE IF EXISTS temp.collab_pair_counts")
        conn.execute("DROP TABLE IF EXISTS temp.collab_directed")
        conn.commit()
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
    self._invalidate_catalog_cache(force=True)
    invalidate_similarity_graph_cache(self)

def fetlife_sample_coverage(self, limit: int = 20) -> dict[str, Any]:
    with Session(self.engine) as session:
        samples = session.exec(select(FetlifeUserSample).order_by(FetlifeUserSample.updated_at.desc())).all()
        rows = session.exec(select(FetlifeUserFetish)).all()
    bucket_counts: dict[str, int] = defaultdict(int)
    for row in rows:
        bucket_counts[row.bucket] += 1
    return {
        "sampled_users": len(samples),
        "sampled_fetishes": len(rows),
        "by_bucket": dict(bucket_counts),
        "latest_users": [
            {
                "nickname": row.nickname,
                "profile_url": row.profile_url,
                "sampled_from_fetish_id": row.sampled_from_fetish_id,
                "state": row.state,
                "updated_at": row.updated_at,
            }
            for row in samples[:limit]
        ],
    }