Pejman Pour-Moezzi
643 lines
25 KiB
Python
643 lines
25 KiB
Python
"""Cluster-first rendering for the v3 pipeline."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
|
|
from . import dates, schema
|
|
|
|
SOURCE_LABELS = {
|
|
"grounding": "Web",
|
|
"hackernews": "Hacker News",
|
|
"truthsocial": "Truth Social",
|
|
"xiaohongshu": "Xiaohongshu",
|
|
"x": "X",
|
|
"github": "GitHub",
|
|
"perplexity": "Perplexity",
|
|
}
|
|
|
|
|
|
_FUN_LEVELS = {
|
|
"low": {"threshold": 80.0, "limit": 2},
|
|
"medium": {"threshold": 70.0, "limit": 5},
|
|
"high": {"threshold": 55.0, "limit": 8},
|
|
}
|
|
|
|
|
|
def render_compact(report: schema.Report, cluster_limit: int = 8, fun_level: str = "medium") -> str:
|
|
non_empty = [s for s, items in sorted(report.items_by_source.items()) if items]
|
|
lines = [
|
|
f"# last30days v3.0.0: {report.topic}",
|
|
"",
|
|
f"- Date range: {report.range_from} to {report.range_to}",
|
|
f"- Sources: {len(non_empty)} active ({', '.join(_source_label(s) for s in non_empty)})" if non_empty else "- Sources: none",
|
|
"",
|
|
]
|
|
|
|
freshness_warning = _assess_data_freshness(report)
|
|
if freshness_warning:
|
|
lines.extend([
|
|
"## Freshness",
|
|
f"- {freshness_warning}",
|
|
"",
|
|
])
|
|
|
|
if report.warnings:
|
|
lines.append("## Warnings")
|
|
lines.extend(f"- {warning}" for warning in report.warnings)
|
|
lines.append("")
|
|
|
|
lines.append("## Ranked Evidence Clusters")
|
|
lines.append("")
|
|
candidate_by_id = {candidate.candidate_id: candidate for candidate in report.ranked_candidates}
|
|
for index, cluster in enumerate(report.clusters[:cluster_limit], start=1):
|
|
lines.append(
|
|
f"### {index}. {cluster.title} "
|
|
f"(score {cluster.score:.0f}, {len(cluster.candidate_ids)} item{'s' if len(cluster.candidate_ids) != 1 else ''}, "
|
|
f"sources: {', '.join(_source_label(source) for source in cluster.sources)})"
|
|
)
|
|
if cluster.uncertainty:
|
|
lines.append(f"- Uncertainty: {cluster.uncertainty}")
|
|
for rep_index, candidate_id in enumerate(cluster.representative_ids, start=1):
|
|
candidate = candidate_by_id.get(candidate_id)
|
|
if not candidate:
|
|
continue
|
|
lines.extend(_render_candidate(candidate, prefix=f"{rep_index}."))
|
|
lines.append("")
|
|
|
|
lines.extend(_render_stats(report))
|
|
|
|
fun_params = _FUN_LEVELS.get(fun_level, _FUN_LEVELS["medium"])
|
|
best_takes = _render_best_takes(report.ranked_candidates, limit=fun_params["limit"], threshold=fun_params["threshold"])
|
|
if best_takes:
|
|
lines.extend([""] + best_takes)
|
|
|
|
lines.extend(_render_source_coverage(report))
|
|
return "\n".join(lines).strip() + "\n"
|
|
|
|
|
|
def render_full(report: schema.Report) -> str:
|
|
"""Full data dump: ALL clusters + ALL items by source. For saved files and debugging."""
|
|
# Start with the same header as compact
|
|
non_empty = [s for s, items in sorted(report.items_by_source.items()) if items]
|
|
lines = [
|
|
f"# last30days v3.0.0: {report.topic}",
|
|
"",
|
|
f"- Date range: {report.range_from} to {report.range_to}",
|
|
f"- Sources: {len(non_empty)} active ({', '.join(_source_label(s) for s in non_empty)})" if non_empty else "- Sources: none",
|
|
"",
|
|
]
|
|
|
|
if report.warnings:
|
|
lines.append("## Warnings")
|
|
lines.extend(f"- {warning}" for warning in report.warnings)
|
|
lines.append("")
|
|
|
|
# ALL clusters (no limit)
|
|
lines.append("## Ranked Evidence Clusters")
|
|
lines.append("")
|
|
candidate_by_id = {c.candidate_id: c for c in report.ranked_candidates}
|
|
for index, cluster in enumerate(report.clusters, start=1):
|
|
lines.append(
|
|
f"### {index}. {cluster.title} "
|
|
f"(score {cluster.score:.0f}, {len(cluster.candidate_ids)} item{'s' if len(cluster.candidate_ids) != 1 else ''}, "
|
|
f"sources: {', '.join(_source_label(s) for s in cluster.sources)})"
|
|
)
|
|
if cluster.uncertainty:
|
|
lines.append(f"- Uncertainty: {cluster.uncertainty}")
|
|
for rep_index, cid in enumerate(cluster.representative_ids, start=1):
|
|
candidate = candidate_by_id.get(cid)
|
|
if not candidate:
|
|
continue
|
|
lines.extend(_render_candidate(candidate, prefix=f"{rep_index}."))
|
|
lines.append("")
|
|
|
|
best_takes = _render_best_takes(report.ranked_candidates)
|
|
if best_takes:
|
|
lines.extend(best_takes)
|
|
lines.append("")
|
|
|
|
# ALL items by source (flat dump, v2-style)
|
|
lines.append("## All Items by Source")
|
|
lines.append("")
|
|
source_order = ["reddit", "x", "youtube", "tiktok", "instagram", "threads", "pinterest",
|
|
"hackernews", "bluesky", "truthsocial", "polymarket", "grounding", "xiaohongshu", "github", "perplexity"]
|
|
for source in source_order:
|
|
items = report.items_by_source.get(source, [])
|
|
if not items:
|
|
continue
|
|
lines.append(f"### {_source_label(source)} ({len(items)} items)")
|
|
lines.append("")
|
|
for item in items:
|
|
score = item.local_rank_score if item.local_rank_score is not None else 0
|
|
lines.append(f"**{item.item_id}** (score:{score:.0f}) {item.author or ''} ({item.published_at or 'date unknown'}) [{_format_item_engagement(item)}]")
|
|
lines.append(f" {item.title}")
|
|
if item.url:
|
|
lines.append(f" {item.url}")
|
|
if item.container:
|
|
lines.append(f" *{item.container}*")
|
|
if item.snippet:
|
|
lines.append(f" {item.snippet[:500]}")
|
|
# Top comments for Reddit
|
|
top_comments = item.metadata.get("top_comments", [])
|
|
if top_comments and isinstance(top_comments[0], dict):
|
|
for tc in top_comments[:3]:
|
|
excerpt = tc.get("excerpt", tc.get("text", ""))[:200]
|
|
tc_score = tc.get("score", "")
|
|
lines.append(f" Top comment ({tc_score} upvotes): {excerpt}")
|
|
# Comment insights for Reddit
|
|
insights = item.metadata.get("comment_insights", [])
|
|
if insights:
|
|
lines.append(" Insights:")
|
|
for ins in insights[:3]:
|
|
lines.append(f" - {ins[:200]}")
|
|
# Transcript highlights for YouTube
|
|
highlights = item.metadata.get("transcript_highlights", [])
|
|
if highlights:
|
|
lines.append(" Highlights:")
|
|
for hl in highlights[:5]:
|
|
lines.append(f' - "{hl[:200]}"')
|
|
# Full transcript snippet for YouTube
|
|
transcript = item.metadata.get("transcript_snippet", "")
|
|
if transcript and len(transcript) > 100:
|
|
lines.append(f" <details><summary>Transcript ({len(transcript.split())} words)</summary>")
|
|
lines.append(f" {transcript[:5000]}")
|
|
lines.append(" </details>")
|
|
# Polymarket outcome prices and market details
|
|
outcome_prices = item.metadata.get("outcome_prices") or []
|
|
if outcome_prices and item.source == "polymarket":
|
|
question = item.metadata.get("question") or ""
|
|
if question and question != item.title:
|
|
lines.append(f" Question: {question}")
|
|
odds_parts = []
|
|
for name, price in outcome_prices:
|
|
if isinstance(price, (int, float)):
|
|
pct = f"{price * 100:.0f}%" if price >= 0.1 else f"{price * 100:.1f}%"
|
|
odds_parts.append(f"{name}: {pct}")
|
|
if odds_parts:
|
|
lines.append(f" Odds: {' | '.join(odds_parts)}")
|
|
remaining = item.metadata.get("outcomes_remaining") or 0
|
|
if remaining:
|
|
lines.append(f" (+{remaining} more outcomes)")
|
|
end_date = item.metadata.get("end_date")
|
|
if end_date:
|
|
lines.append(f" Closes: {end_date}")
|
|
lines.append("")
|
|
|
|
lines.extend(_render_stats(report))
|
|
lines.extend(_render_source_coverage(report))
|
|
return "\n".join(lines).strip() + "\n"
|
|
|
|
|
|
def _format_item_engagement(item: schema.SourceItem) -> str:
|
|
"""Format engagement metrics for a SourceItem in the full dump."""
|
|
eng = item.engagement
|
|
if not eng:
|
|
return ""
|
|
parts = []
|
|
for key in ["score", "likes", "views", "points", "reposts", "replies", "comments",
|
|
"play_count", "digg_count", "share_count", "num_comments"]:
|
|
val = eng.get(key)
|
|
if val is not None and val != 0:
|
|
parts.append(f"{val} {key}")
|
|
return ", ".join(parts) if parts else ""
|
|
|
|
|
|
def render_context(report: schema.Report, cluster_limit: int = 6) -> str:
|
|
candidate_by_id = {candidate.candidate_id: candidate for candidate in report.ranked_candidates}
|
|
lines = [
|
|
f"Topic: {report.topic}",
|
|
f"Intent: {report.query_plan.intent}",
|
|
]
|
|
freshness_warning = _assess_data_freshness(report)
|
|
if freshness_warning:
|
|
lines.append(f"Freshness warning: {freshness_warning}")
|
|
lines.append("Top clusters:")
|
|
for cluster in report.clusters[:cluster_limit]:
|
|
lines.append(f"- {cluster.title} [{', '.join(_source_label(source) for source in cluster.sources)}]")
|
|
for candidate_id in cluster.representative_ids[:2]:
|
|
candidate = candidate_by_id.get(candidate_id)
|
|
if not candidate:
|
|
continue
|
|
detail_parts = [
|
|
schema.candidate_source_label(candidate),
|
|
candidate.title,
|
|
schema.candidate_best_published_at(candidate) or "date unknown",
|
|
candidate.url,
|
|
]
|
|
lines.append(f" - {' | '.join(detail_parts)}")
|
|
if candidate.snippet:
|
|
lines.append(f" Evidence: {_truncate(candidate.snippet, 180)}")
|
|
if report.warnings:
|
|
lines.append("Warnings:")
|
|
lines.extend(f"- {warning}" for warning in report.warnings)
|
|
return "\n".join(lines).strip() + "\n"
|
|
|
|
|
|
def _render_candidate(candidate: schema.Candidate, prefix: str) -> list[str]:
|
|
primary = schema.candidate_primary_item(candidate)
|
|
detail_parts = [
|
|
_format_date(primary),
|
|
_format_actor(primary),
|
|
_format_engagement(primary),
|
|
f"score:{candidate.final_score:.0f}",
|
|
]
|
|
if candidate.fun_score is not None and candidate.fun_score >= 50:
|
|
detail_parts.append(f"fun:{candidate.fun_score:.0f}")
|
|
details = " | ".join(part for part in detail_parts if part)
|
|
lines = [
|
|
f"{prefix} [{schema.candidate_source_label(candidate)}] {candidate.title}",
|
|
f" - {details}",
|
|
f" - URL: {candidate.url}",
|
|
]
|
|
corroboration = _format_corroboration(candidate)
|
|
if corroboration:
|
|
lines.append(f" - {corroboration}")
|
|
explanation = _format_explanation(candidate)
|
|
if explanation:
|
|
lines.append(f" - Why: {explanation}")
|
|
if candidate.snippet:
|
|
lines.append(f" - Evidence: {_truncate(candidate.snippet, 360)}")
|
|
for tc in _top_comments_list(primary):
|
|
excerpt = tc.get("excerpt") or tc.get("text") or ""
|
|
score = tc.get("score", "")
|
|
lines.append(f" - Comment ({score} upvotes): {_truncate(excerpt.strip(), 240)}")
|
|
insight = _comment_insight(primary)
|
|
if insight:
|
|
lines.append(f" - Insight: {_truncate(insight, 220)}")
|
|
highlights = _transcript_highlights(primary)
|
|
if highlights:
|
|
lines.append(" - Highlights:")
|
|
for hl in highlights:
|
|
lines.append(f' - "{_truncate(hl, 200)}"')
|
|
return lines
|
|
|
|
|
|
def _format_volume_short(volume: float) -> str:
|
|
"""Format volume as short string: 66000 -> '$66K', 1200000 -> '$1.2M'."""
|
|
if volume >= 1_000_000:
|
|
return f"${volume / 1_000_000:.1f}M"
|
|
if volume >= 1_000:
|
|
return f"${volume / 1_000:.0f}K"
|
|
if volume >= 1:
|
|
return f"${volume:.0f}"
|
|
return ""
|
|
|
|
|
|
def _polymarket_top_markets(items: list[schema.SourceItem], limit: int = 3) -> list[str]:
|
|
"""Build short summary strings for the top Polymarket markets by volume.
|
|
|
|
Returns list like: ['"BULLY <300k": 96% ($66K)', '"Top Spotify": Kanye 6.5% ($21K)']
|
|
"""
|
|
# Sort by volume descending
|
|
sorted_items = sorted(
|
|
items,
|
|
key=lambda it: it.engagement.get("volume") or 0,
|
|
reverse=True,
|
|
)
|
|
|
|
summaries = []
|
|
for item in sorted_items[:limit]:
|
|
outcome_prices = item.metadata.get("outcome_prices") or []
|
|
if not outcome_prices:
|
|
continue
|
|
|
|
# Pick the leading outcome (first one, already sorted by relevance in polymarket.py)
|
|
lead_name, lead_price = outcome_prices[0]
|
|
# For binary Yes/No markets, show "Yes: 96%" format
|
|
# For multi-outcome, show "OutcomeName: X%"
|
|
if isinstance(lead_price, (int, float)):
|
|
pct = f"{lead_price * 100:.0f}%" if lead_price >= 0.1 else f"{lead_price * 100:.1f}%"
|
|
else:
|
|
continue
|
|
|
|
# Short title
|
|
title = item.metadata.get("question") or item.title
|
|
if len(title) > 30:
|
|
title = title[:27] + "..."
|
|
|
|
summaries.append(f'"{title}": {lead_name} {pct}')
|
|
|
|
return summaries
|
|
|
|
|
|
def _render_source_coverage(report: schema.Report) -> list[str]:
|
|
lines = [
|
|
"## Source Coverage",
|
|
"",
|
|
]
|
|
for source, items in sorted(report.items_by_source.items()):
|
|
lines.append(f"- {_source_label(source)}: {len(items)} item{'s' if len(items) != 1 else ''}")
|
|
if report.errors_by_source:
|
|
lines.append("")
|
|
lines.append("## Source Errors")
|
|
lines.append("")
|
|
for source, error in sorted(report.errors_by_source.items()):
|
|
lines.append(f"- {_source_label(source)}: {error}")
|
|
return lines
|
|
|
|
|
|
def _render_stats(report: schema.Report) -> list[str]:
|
|
lines = [
|
|
"## Stats",
|
|
"",
|
|
]
|
|
non_empty_sources = {
|
|
source: items
|
|
for source, items in sorted(report.items_by_source.items())
|
|
if items
|
|
}
|
|
total_items = sum(len(items) for items in non_empty_sources.values())
|
|
if not non_empty_sources:
|
|
lines.append("- No usable source metrics available.")
|
|
lines.append("")
|
|
return lines
|
|
|
|
lines.append(
|
|
f"- Total evidence: {total_items} item{'s' if total_items != 1 else ''} across "
|
|
f"{len(non_empty_sources)} source{'s' if len(non_empty_sources) != 1 else ''}"
|
|
)
|
|
top_voices = _top_voices_overall(non_empty_sources)
|
|
if top_voices:
|
|
lines.append(f"- Top voices: {', '.join(top_voices)}")
|
|
for source, items in non_empty_sources.items():
|
|
if source == "polymarket":
|
|
# Polymarket gets a richer stats line with top market odds
|
|
market_summaries = _polymarket_top_markets(items)
|
|
if market_summaries:
|
|
label = f"{len(items)} market{'s' if len(items) != 1 else ''}"
|
|
parts_str = f"{label} | " + " | ".join(market_summaries)
|
|
else:
|
|
parts_str = f"{len(items)} market{'s' if len(items) != 1 else ''}"
|
|
engagement_summary = _aggregate_engagement(source, items)
|
|
if engagement_summary:
|
|
parts_str += f" | {engagement_summary}"
|
|
lines.append(f"- {_source_label(source)}: {parts_str}")
|
|
continue
|
|
parts = [f"{len(items)} item{'s' if len(items) != 1 else ''}"]
|
|
engagement_summary = _aggregate_engagement(source, items)
|
|
if engagement_summary:
|
|
parts.append(engagement_summary)
|
|
actor_summary = _top_actor_summary(source, items)
|
|
if actor_summary:
|
|
parts.append(actor_summary)
|
|
lines.append(f"- {_source_label(source)}: {' | '.join(parts)}")
|
|
lines.append("")
|
|
return lines
|
|
|
|
|
|
def _assess_data_freshness(report: schema.Report) -> str | None:
|
|
dated_items = [
|
|
item
|
|
for items in report.items_by_source.values()
|
|
for item in items
|
|
if item.published_at
|
|
]
|
|
if not dated_items:
|
|
return "Limited recent data: no usable dated evidence made it into the retrieved pool."
|
|
recent_items = [
|
|
item
|
|
for item in dated_items
|
|
if (_days_ago := dates.days_ago(item.published_at)) is not None and _days_ago <= 7
|
|
]
|
|
if len(recent_items) < 3:
|
|
return f"Limited recent data: only {len(recent_items)} of {len(dated_items)} dated items are from the last 7 days."
|
|
if len(recent_items) * 2 < len(dated_items):
|
|
return f"Recent evidence is thin: only {len(recent_items)} of {len(dated_items)} dated items are from the last 7 days."
|
|
return None
|
|
|
|
|
|
def _format_date(item: schema.SourceItem | None) -> str:
|
|
if not item or not item.published_at:
|
|
return "date unknown [date:low]"
|
|
if item.date_confidence == "high":
|
|
return item.published_at
|
|
return f"{item.published_at} [date:{item.date_confidence}]"
|
|
|
|
|
|
def _format_actor(item: schema.SourceItem | None) -> str | None:
|
|
if not item:
|
|
return None
|
|
if item.source == "reddit" and item.container:
|
|
return f"r/{item.container}"
|
|
if item.source in {"x", "bluesky", "truthsocial"} and item.author:
|
|
return f"@{item.author.lstrip('@')}"
|
|
if item.source == "youtube" and item.author:
|
|
return item.author
|
|
if item.container and item.container != "Polymarket":
|
|
return item.container
|
|
if item.author:
|
|
return item.author
|
|
return None
|
|
|
|
|
|
# Per-source engagement display fields: list of (field_name, label) tuples.
|
|
ENGAGEMENT_DISPLAY: dict[str, list[tuple[str, str]]] = {
|
|
"reddit": [("score", "pts"), ("num_comments", "cmt")],
|
|
"x": [("likes", "likes"), ("reposts", "rt"), ("replies", "re")],
|
|
"youtube": [("views", "views"), ("likes", "likes"), ("comments", "cmt")],
|
|
"tiktok": [("views", "views"), ("likes", "likes"), ("comments", "cmt")],
|
|
"instagram": [("views", "views"), ("likes", "likes"), ("comments", "cmt")],
|
|
"threads": [("likes", "likes"), ("replies", "re")],
|
|
"pinterest": [("saves", "saves"), ("comments", "cmt")],
|
|
"hackernews": [("points", "pts"), ("comments", "cmt")],
|
|
"bluesky": [("likes", "likes"), ("reposts", "rt"), ("replies", "re")],
|
|
"truthsocial": [("likes", "likes"), ("reposts", "rt"), ("replies", "re")],
|
|
"polymarket": [],
|
|
"github": [("reactions", "react"), ("comments", "cmt")],
|
|
"perplexity": [("citations", "cite")],
|
|
}
|
|
|
|
|
|
def _format_engagement(item: schema.SourceItem | None) -> str | None:
|
|
if not item or not item.engagement:
|
|
return None
|
|
engagement = item.engagement
|
|
fields = ENGAGEMENT_DISPLAY.get(item.source)
|
|
if fields:
|
|
text = _fmt_pairs([(engagement.get(field), label) for field, label in fields])
|
|
else:
|
|
# Generic fallback: engagement.items() yields (key, value) but
|
|
# _fmt_pairs expects (value, label), so swap them.
|
|
text = _fmt_pairs([(value, key) for key, value in list(engagement.items())[:3]])
|
|
return f"[{text}]" if text else None
|
|
|
|
|
|
def _fmt_pairs(pairs: list[tuple[object, str]]) -> str:
|
|
rendered = []
|
|
for value, suffix in pairs:
|
|
if value in (None, "", 0, 0.0):
|
|
continue
|
|
rendered.append(f"{_format_number(value)}{suffix}")
|
|
return ", ".join(rendered)
|
|
|
|
|
|
def _format_number(value: object) -> str:
|
|
try:
|
|
numeric = float(value)
|
|
except (TypeError, ValueError):
|
|
return str(value)
|
|
if numeric >= 1000 and numeric.is_integer():
|
|
return f"{int(numeric):,}"
|
|
if numeric.is_integer():
|
|
return str(int(numeric))
|
|
return f"{numeric:.1f}"
|
|
|
|
|
|
def _aggregate_engagement(source: str, items: list[schema.SourceItem]) -> str | None:
|
|
fields = ENGAGEMENT_DISPLAY.get(source)
|
|
if not fields:
|
|
return None
|
|
totals: list[tuple[float | int | None, str]] = []
|
|
for field, label in fields:
|
|
total = 0
|
|
found = False
|
|
for item in items:
|
|
value = item.engagement.get(field)
|
|
if value in (None, ""):
|
|
continue
|
|
found = True
|
|
total += value
|
|
totals.append((total if found else None, label))
|
|
return _fmt_pairs(totals) or None
|
|
|
|
|
|
def _top_actor_summary(source: str, items: list[schema.SourceItem]) -> str | None:
|
|
actors = _top_actors_for_source(source, items)
|
|
if not actors:
|
|
return None
|
|
label = {
|
|
"reddit": "communities",
|
|
"grounding": "domains",
|
|
"youtube": "channels",
|
|
"hackernews": "domains",
|
|
}.get(source, "voices")
|
|
return f"{label}: {', '.join(actors)}"
|
|
|
|
|
|
def _top_actors_for_source(source: str, items: list[schema.SourceItem], limit: int = 3) -> list[str]:
|
|
counts: Counter[str] = Counter()
|
|
for item in items:
|
|
actor = _stats_actor(item)
|
|
if actor:
|
|
counts[actor] += 1
|
|
return [actor for actor, _ in counts.most_common(limit)]
|
|
|
|
|
|
def _top_voices_overall(items_by_source: dict[str, list[schema.SourceItem]], limit: int = 5) -> list[str]:
|
|
counts: Counter[str] = Counter()
|
|
for items in items_by_source.values():
|
|
for item in items:
|
|
actor = _stats_actor(item)
|
|
if actor:
|
|
counts[actor] += 1
|
|
return [actor for actor, _ in counts.most_common(limit)]
|
|
|
|
|
|
def _stats_actor(item: schema.SourceItem) -> str | None:
|
|
if item.source == "reddit" and item.container:
|
|
return f"r/{item.container}"
|
|
if item.source in {"x", "bluesky", "truthsocial"} and item.author:
|
|
return f"@{item.author.lstrip('@')}"
|
|
if item.source == "grounding" and item.container:
|
|
return item.container
|
|
if item.source == "youtube" and item.author:
|
|
return item.author
|
|
if item.container and item.container != "Polymarket":
|
|
return item.container
|
|
if item.author:
|
|
return item.author
|
|
return None
|
|
|
|
|
|
def _format_corroboration(candidate: schema.Candidate) -> str | None:
|
|
corroborating = [
|
|
_source_label(source)
|
|
for source in schema.candidate_sources(candidate)
|
|
if source != candidate.source
|
|
]
|
|
if not corroborating:
|
|
return None
|
|
return f"Also on: {', '.join(corroborating)}"
|
|
|
|
|
|
def _format_explanation(candidate: schema.Candidate) -> str | None:
|
|
if not candidate.explanation or candidate.explanation == "fallback-local-score":
|
|
return None
|
|
return candidate.explanation
|
|
|
|
|
|
def _top_comments_list(item: schema.SourceItem | None, limit: int = 3, min_score: int = 10) -> list[dict]:
|
|
"""Return up to `limit` top comments with score >= min_score."""
|
|
if not item:
|
|
return []
|
|
comments = item.metadata.get("top_comments") or []
|
|
if not comments or not isinstance(comments[0], dict):
|
|
return []
|
|
return [c for c in comments if (c.get("score") or 0) >= min_score][:limit]
|
|
|
|
|
|
def _top_comment_excerpt(item: schema.SourceItem | None) -> str | None:
|
|
if not item:
|
|
return None
|
|
comments = item.metadata.get("top_comments") or []
|
|
if not comments or not isinstance(comments[0], dict):
|
|
return None
|
|
top = comments[0]
|
|
return str(top.get("excerpt") or top.get("text") or "").strip() or None
|
|
|
|
|
|
def _comment_insight(item: schema.SourceItem | None) -> str | None:
|
|
if not item:
|
|
return None
|
|
insights = item.metadata.get("comment_insights") or []
|
|
if not insights:
|
|
return None
|
|
return str(insights[0]).strip() or None
|
|
|
|
|
|
def _transcript_highlights(item: schema.SourceItem | None) -> list[str]:
|
|
if not item or item.source != "youtube":
|
|
return []
|
|
return (item.metadata.get("transcript_highlights") or [])[:5]
|
|
|
|
|
|
def _source_label(source: str) -> str:
|
|
return SOURCE_LABELS.get(source, source.replace("_", " ").title())
|
|
|
|
|
|
|
|
def _render_best_takes(candidates, limit=5, threshold=70.0):
|
|
gems = sorted(
|
|
(c for c in candidates if c.fun_score is not None and c.fun_score >= threshold),
|
|
key=lambda c: -(c.fun_score or 0),
|
|
)
|
|
if len(gems) < 2:
|
|
return []
|
|
lines = ["## Best Takes", ""]
|
|
for candidate in gems[:limit]:
|
|
text = candidate.title.strip()
|
|
for item in candidate.source_items:
|
|
for comment in item.metadata.get("top_comments", [])[:3]:
|
|
body = (comment.get("body") or comment.get("text") or "") if isinstance(comment, dict) else str(comment)
|
|
body = body.strip()
|
|
if body and len(body) < len(text) and len(body) > 10:
|
|
text = body
|
|
source_label = _source_label(candidate.source)
|
|
author = candidate.source_items[0].author if candidate.source_items else None
|
|
attribution = f"@{author} on {source_label}" if author and candidate.source in ("x", "tiktok", "instagram", "threads") else f"{source_label}"
|
|
if author and candidate.source == "reddit":
|
|
container = candidate.source_items[0].container if candidate.source_items else None
|
|
attribution = f"r/{container} comment" if container else "Reddit"
|
|
score_tag = f"(fun:{candidate.fun_score:.0f})"
|
|
reason = f" -- {candidate.fun_explanation}" if candidate.fun_explanation and candidate.fun_explanation != "heuristic-fallback" else ""
|
|
lines.append(f'- "{_truncate(text, 280)}" -- {attribution} {score_tag}{reason}')
|
|
return lines
|
|
|
|
|
|
def _truncate(text: str, limit: int) -> str:
|
|
text = text.strip()
|
|
if len(text) <= limit:
|
|
return text
|
|
return text[: limit - 3].rstrip() + "..."
|