"""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] ], }