Spaces:
Sleeping
Sleeping
File size: 4,923 Bytes
ef78361 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """Supabase research queries — topic clusters, cross-model analysis.
Module-level functions extracted from supabase_client.py.
All functions use get_supabase_client() from the parent module.
"""
def get_topic_clusters(campaign_id: int, source: str | None = None) -> list[dict]:
"""Fetch topic clusters with scores for a campaign, optionally filtered by source.
Returns:
List of cluster dicts sorted by opportunity_score DESC.
"""
from core.supabase_client import get_supabase_client
client = get_supabase_client()
query = (
client.table("topic_clusters")
.select(
"id, cluster_label, fanout_count, unique_questions, "
"attention_score, citation_density, opportunity_score, "
"sample_fanouts, top_sources, source"
)
.eq("campaign_id", campaign_id)
)
if source:
query = query.eq("source", source)
result = query.order("opportunity_score", desc=True).execute()
return result.data or []
def get_topic_map_snapshot(campaign_id: int, source: str | None = None) -> dict | None:
"""Fetch latest UMAP 2D coordinates for visualization.
Returns:
Dict with coordinates and algorithm_params, or None.
"""
from core.supabase_client import get_supabase_client
client = get_supabase_client()
result = (
client.table("topic_map_snapshots")
.select("coordinates, algorithm_params")
.eq("campaign_id", campaign_id)
.order("created_at", desc=True)
.execute()
)
if not result.data:
return None
if source:
for snap in result.data:
params = snap.get("algorithm_params") or {}
if params.get("source") == source:
return snap
return None
return result.data[0]
def get_cross_model_analysis(
campaign_chatgpt: int,
campaign_gemini: int,
) -> dict | None:
"""Fetch cross-model analysis summary (NMI, match count).
Returns:
Dict with nmi_score, total_matched_topics, algorithm_params, or None.
"""
from core.supabase_client import get_supabase_client
client = get_supabase_client()
result = (
client.table("cross_model_analysis")
.select("nmi_score, total_matched_topics, algorithm_params, created_at")
.eq("campaign_chatgpt", campaign_chatgpt)
.eq("campaign_gemini", campaign_gemini)
.limit(1)
.execute()
)
return result.data[0] if result.data else None
def get_gap_scores(
campaign_chatgpt: int,
campaign_gemini: int,
) -> list[dict]:
"""Fetch cross-model topic matches with GapScore and quadrant.
Returns:
List of match dicts with cluster labels, sorted by gap_score DESC.
"""
from core.supabase_client import get_supabase_client
client = get_supabase_client()
result = (
client.table("cross_model_topic_matches")
.select(
"id, chatgpt_cluster_id, gemini_cluster_id, "
"match_score, label_similarity, centroid_similarity, "
"demand_percentile, supply_percentile, gap_score, quadrant"
)
.eq("campaign_chatgpt", campaign_chatgpt)
.eq("campaign_gemini", campaign_gemini)
.not_.is_("gap_score", "null")
.order("gap_score", desc=True)
.execute()
)
matches = result.data or []
# Enrich with cluster labels (batch query instead of N+1)
if matches:
chatgpt_ids = [m["chatgpt_cluster_id"] for m in matches]
gemini_ids = [m["gemini_cluster_id"] for m in matches]
all_ids = list(set(chatgpt_ids + gemini_ids))
label_map = {}
label_result = (
client.table("topic_clusters")
.select("id, cluster_label")
.in_("id", all_ids)
.execute()
)
for row in (label_result.data or []):
label_map[row["id"]] = row.get("cluster_label", "")
for m in matches:
m["chatgpt_label"] = label_map.get(m["chatgpt_cluster_id"], "")
m["gemini_label"] = label_map.get(m["gemini_cluster_id"], "")
return matches
def find_cross_model_pair(campaign_id: int) -> dict | None:
"""Find cross-model analysis pair containing this campaign_id.
Checks both chatgpt and gemini sides so the sidebar only needs one ID.
Returns:
Dict with campaign_chatgpt, campaign_gemini, nmi_score,
total_matched_topics, or None if no pair exists.
"""
from core.supabase_client import get_supabase_client
client = get_supabase_client()
resp = (
client.table("cross_model_analysis")
.select("campaign_chatgpt, campaign_gemini, nmi_score, total_matched_topics")
.or_(f"campaign_chatgpt.eq.{campaign_id},campaign_gemini.eq.{campaign_id}")
.limit(1)
.execute()
)
if resp.data:
return resp.data[0]
return None
|