sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(truthsocial): Add Truth Social as opt-in source

Browse Source 
Mastodon-compatible API at truthsocial.com/api/v2/search.
Opt-in via TRUTHSOCIAL_TOKEN env var (bearer token from browser).
Silent when unconfigured. Full pipeline: search, parse, normalize,
score, dedupe, render across all 10 pipeline files.

27 new tests, 440 total passing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Matt Van Horn
2026-03-10 00:14:39 -07:00
parent b6fd5ff406
commit b38703e53d
12 changed files with 871 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/lib/dedupe.py
+8
View File
@@ -234,6 +234,14 @@ def dedupe_bluesky(
return dedupe_items(items, threshold)
def dedupe_truthsocial(
items: List[schema.TruthSocialItem],
threshold: float = 0.7,
) -> List[schema.TruthSocialItem]:
"""Dedupe Truth Social items."""
return dedupe_items(items, threshold)
def dedupe_polymarket(
items: List[schema.PolymarketItem],
threshold: float = 0.7,
scripts/lib/env.py
+9
View File
@@ -257,6 +257,7 @@ def get_config() -> Dict[str, Any]:
('CT0', None),
('BSKY_HANDLE', None),
('BSKY_APP_PASSWORD', None),
('TRUTHSOCIAL_TOKEN', None),
]
for key, default in keys:
@@ -490,6 +491,14 @@ def is_bluesky_available(config: Dict[str, Any]) -> bool:
return bool(config.get('BSKY_HANDLE') and config.get('BSKY_APP_PASSWORD'))
def is_truthsocial_available(config: Dict[str, Any]) -> bool:
"""Check if Truth Social source is available.
Requires TRUTHSOCIAL_TOKEN (bearer token from browser dev tools).
"""
return bool(config.get('TRUTHSOCIAL_TOKEN'))
def is_polymarket_available() -> bool:
"""Check if Polymarket source is available.
scripts/lib/normalize.py
+43
View File
@@ -394,6 +394,49 @@ def normalize_bluesky_items(
return normalized
def normalize_truthsocial_items(
items: List[Dict[str, Any]],
from_date: str,
to_date: str,
) -> List[schema.TruthSocialItem]:
"""Normalize raw Truth Social items to schema.
Args:
items: Raw Truth Social items from Mastodon API
from_date: Start of date range
to_date: End of date range
Returns:
List of TruthSocialItem objects
"""
normalized = []
for i, item in enumerate(items):
eng_raw = item.get("engagement") or {}
engagement = schema.Engagement(
likes=eng_raw.get("likes"),
reposts=eng_raw.get("reposts"),
replies=eng_raw.get("replies"),
)
date_str = item.get("date")
normalized.append(schema.TruthSocialItem(
id=f"TS{i+1}",
text=item.get("text", ""),
url=item.get("url", ""),
author_handle=item.get("handle", ""),
display_name=item.get("display_name", ""),
date=date_str,
date_confidence="high",
engagement=engagement,
relevance=item.get("relevance", 0.5),
why_relevant=item.get("why_relevant", ""),
))
return normalized
def normalize_polymarket_items(
items: List[Dict[str, Any]],
from_date: str,
scripts/lib/render.py
+72 -2
View File
@@ -30,6 +30,10 @@ def _xref_tag(item) -> str:
source_names.add('Instagram')
elif ref_id.startswith('HN'):
source_names.add('HN')
elif ref_id.startswith('BS'):
source_names.add('Bluesky')
elif ref_id.startswith('TS'):
source_names.add('Truth Social')
elif ref_id.startswith('PM'):
source_names.add('Polymarket')
elif ref_id.startswith('W'):
@@ -60,13 +64,14 @@ def _assess_data_freshness(report: schema.Report) -> dict:
web_recent = sum(1 for w in report.web if w.date and w.date >= report.range_from)
hn_recent = sum(1 for h in report.hackernews if h.date and h.date >= report.range_from)
bsky_recent = sum(1 for b in report.bluesky if b.date and b.date >= report.range_from)
ts_recent = sum(1 for ts in report.truthsocial if ts.date and ts.date >= report.range_from)
pm_recent = sum(1 for p in report.polymarket if p.date and p.date >= report.range_from)
tiktok_recent = sum(1 for t in report.tiktok if t.date and t.date >= report.range_from)
ig_recent = sum(1 for ig in report.instagram if ig.date and ig.date >= report.range_from)
total_recent = reddit_recent + x_recent + web_recent + hn_recent + bsky_recent + pm_recent + tiktok_recent + ig_recent
total_items = len(report.reddit) + len(report.x) + len(report.web) + len(report.hackernews) + len(report.bluesky) + len(report.polymarket) + len(report.tiktok) + len(report.instagram)
total_recent = reddit_recent + x_recent + web_recent + hn_recent + bsky_recent + ts_recent + pm_recent + tiktok_recent + ig_recent
total_items = len(report.reddit) + len(report.x) + len(report.web) + len(report.hackernews) + len(report.bluesky) + len(report.truthsocial) + len(report.polymarket) + len(report.tiktok) + len(report.instagram)
return {
"reddit_recent": reddit_recent,
@@ -404,6 +409,42 @@ def render_compact(report: schema.Report, limit: int = 15, missing_keys: str = "
lines.append(f" *{item.why_relevant}*")
lines.append("")
# Truth Social items
if report.truthsocial_error:
lines.append("### Truth Social Posts")
lines.append("")
lines.append(f"**ERROR:** {report.truthsocial_error}")
lines.append("")
elif report.truthsocial:
lines.append("### Truth Social Posts")
lines.append("")
for item in report.truthsocial[:limit]:
eng_str = ""
if item.engagement:
eng = item.engagement
parts = []
if eng.likes is not None:
parts.append(f"{eng.likes}lk")
if eng.reposts is not None:
parts.append(f"{eng.reposts}rp")
if eng.replies is not None:
parts.append(f"{eng.replies}re")
if parts:
eng_str = f" [{', '.join(parts)}]"
date_str = f" ({item.date})" if item.date else ""
lines.append(f"**{item.id}** (score:{item.score}) @{item.author_handle}{date_str}{eng_str}{_xref_tag(item)}")
if item.text:
snippet = item.text[:200]
if len(item.text) > 200:
snippet += "..."
lines.append(f" {snippet}")
if item.url:
lines.append(f" {item.url}")
lines.append(f" *{item.why_relevant}*")
lines.append("")
# Polymarket items
if report.polymarket_error:
lines.append("### Prediction Markets (Polymarket)")
@@ -575,6 +616,13 @@ def render_source_status(report: schema.Report, source_info: dict = None) -> str
lines.append(f" ✅ Bluesky: {len(report.bluesky)} posts")
# Hide when zero results
# Truth Social
if report.truthsocial_error:
lines.append(f" ❌ Truth Social: error - {report.truthsocial_error}")
elif report.truthsocial:
lines.append(f" ✅ Truth Social: {len(report.truthsocial)} posts")
# Hide when zero results
# Polymarket
if report.polymarket_error:
lines.append(f" ❌ Polymarket: error - {report.polymarket_error}")
@@ -627,6 +675,8 @@ def render_context_snippet(report: schema.Report) -> str:
all_items.append((item.score, "HN", item.title[:50] + "...", item.hn_url))
for item in report.bluesky[:5]:
all_items.append((item.score, "Bluesky", item.text[:50] + "...", item.url))
for item in report.truthsocial[:5]:
all_items.append((item.score, "Truth Social", item.text[:50] + "...", item.url))
for item in report.polymarket[:5]:
all_items.append((item.score, "Polymarket", item.question[:50] + "...", item.url))
for item in report.web[:5]:
@@ -820,6 +870,26 @@ def render_full_report(report: schema.Report) -> str:
lines.append(f"> {item.text[:300]}")
lines.append("")
# Truth Social section
if report.truthsocial:
lines.append("## Truth Social Posts")
lines.append("")
for item in report.truthsocial:
lines.append(f"### {item.id}: @{item.author_handle}")
lines.append("")
lines.append(f"- **URL:** {item.url}")
lines.append(f"- **Date:** {item.date or 'Unknown'}")
lines.append(f"- **Score:** {item.score}/100")
lines.append(f"- **Relevance:** {item.why_relevant}")
if item.engagement:
eng = item.engagement
lines.append(f"- **Engagement:** {eng.likes or '?'} likes, {eng.reposts or '?'} reposts, {eng.replies or '?'} replies")
lines.append("")
lines.append(f"> {item.text[:300]}")
lines.append("")
# Polymarket section
if report.polymarket:
lines.append("## Prediction Markets (Polymarket)")
scripts/lib/schema.py
+67
View File
@@ -392,6 +392,43 @@ class BlueskyItem:
return d
@dataclass
class TruthSocialItem:
"""Normalized Truth Social post."""
id: str # "TS1", "TS2", ...
text: str
url: str # truthsocial.com permalink
author_handle: str # username
display_name: str
date: Optional[str] = None
date_confidence: str = "high" # Mastodon API has exact timestamps
engagement: Optional[Engagement] = None # likes, reposts, replies
relevance: float = 0.5
why_relevant: str = ""
subs: SubScores = field(default_factory=SubScores)
score: int = 0
cross_refs: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
d = {
'id': self.id,
'text': self.text,
'url': self.url,
'author_handle': self.author_handle,
'display_name': self.display_name,
'date': self.date,
'date_confidence': self.date_confidence,
'engagement': self.engagement.to_dict() if self.engagement else None,
'relevance': self.relevance,
'why_relevant': self.why_relevant,
'subs': self.subs.to_dict(),
'score': self.score,
}
if self.cross_refs:
d['cross_refs'] = self.cross_refs
return d
@dataclass
class PolymarketItem:
"""Normalized Polymarket prediction market item."""
@@ -453,6 +490,7 @@ class Report:
instagram: List[InstagramItem] = field(default_factory=list)
hackernews: List[HackerNewsItem] = field(default_factory=list)
bluesky: List[BlueskyItem] = field(default_factory=list)
truthsocial: List[TruthSocialItem] = field(default_factory=list)
polymarket: List[PolymarketItem] = field(default_factory=list)
best_practices: List[str] = field(default_factory=list)
prompt_pack: List[str] = field(default_factory=list)
@@ -466,6 +504,7 @@ class Report:
instagram_error: Optional[str] = None
hackernews_error: Optional[str] = None
bluesky_error: Optional[str] = None
truthsocial_error: Optional[str] = None
polymarket_error: Optional[str] = None
# Handle resolution
resolved_x_handle: Optional[str] = None
@@ -492,6 +531,7 @@ class Report:
'instagram': [ig.to_dict() for ig in self.instagram],
'hackernews': [h.to_dict() for h in self.hackernews],
'bluesky': [b.to_dict() for b in self.bluesky],
'truthsocial': [ts.to_dict() for ts in self.truthsocial],
'polymarket': [p.to_dict() for p in self.polymarket],
'best_practices': self.best_practices,
'prompt_pack': self.prompt_pack,
@@ -515,6 +555,8 @@ class Report:
d['hackernews_error'] = self.hackernews_error
if self.bluesky_error:
d['bluesky_error'] = self.bluesky_error
if self.truthsocial_error:
d['truthsocial_error'] = self.truthsocial_error
if self.polymarket_error:
d['polymarket_error'] = self.polymarket_error
if self.from_cache:
@@ -694,6 +736,29 @@ class Report:
cross_refs=h.get('cross_refs', []),
))
# Reconstruct Truth Social items (backward compat: key may not exist)
ts_items = []
for ts in data.get('truthsocial', []):
eng = None
if ts.get('engagement'):
eng = Engagement(**ts['engagement'])
subs = SubScores(**ts.get('subs', {})) if ts.get('subs') else SubScores()
ts_items.append(TruthSocialItem(
id=ts['id'],
text=ts['text'],
url=ts['url'],
author_handle=ts.get('author_handle', ''),
display_name=ts.get('display_name', ''),
date=ts.get('date'),
date_confidence=ts.get('date_confidence', 'high'),
engagement=eng,
relevance=ts.get('relevance', 0.5),
why_relevant=ts.get('why_relevant', ''),
subs=subs,
score=ts.get('score', 0),
cross_refs=ts.get('cross_refs', []),
))
# Reconstruct Polymarket items (backward compat: key may not exist)
pm_items = []
for p in data.get('polymarket', []):
@@ -735,6 +800,7 @@ class Report:
tiktok=tiktok_items,
instagram=ig_items,
hackernews=hn_items,
truthsocial=ts_items,
polymarket=pm_items,
best_practices=data.get('best_practices', []),
prompt_pack=data.get('prompt_pack', []),
@@ -746,6 +812,7 @@ class Report:
tiktok_error=data.get('tiktok_error'),
instagram_error=data.get('instagram_error'),
hackernews_error=data.get('hackernews_error'),
truthsocial_error=data.get('truthsocial_error'),
polymarket_error=data.get('polymarket_error'),
resolved_x_handle=data.get('resolved_x_handle'),
from_cache=data.get('from_cache', False),
scripts/lib/score.py
+56
View File
@@ -528,6 +528,62 @@ def score_bluesky_items(items: List[schema.BlueskyItem]) -> List[schema.BlueskyI
return items
def compute_truthsocial_engagement_raw(engagement: Optional[schema.Engagement]) -> Optional[float]:
"""Compute raw engagement score for Truth Social item.
Formula: 0.45*log1p(likes) + 0.30*log1p(reposts) + 0.25*log1p(replies)
Likes are primary signal; reposts indicate reach; replies indicate discussion.
"""
if engagement is None:
return None
if engagement.likes is None and engagement.reposts is None:
return None
likes = log1p_safe(engagement.likes)
reposts = log1p_safe(engagement.reposts)
replies = log1p_safe(engagement.replies)
return 0.45 * likes + 0.30 * reposts + 0.25 * replies
def score_truthsocial_items(items: List[schema.TruthSocialItem]) -> List[schema.TruthSocialItem]:
"""Compute scores for Truth Social items."""
if not items:
return items
eng_raw = [compute_truthsocial_engagement_raw(item.engagement) for item in items]
eng_normalized = normalize_to_100(eng_raw)
for i, item in enumerate(items):
rel_score = int(item.relevance * 100)
rec_score = dates.recency_score(item.date)
if eng_normalized[i] is not None:
eng_score = int(eng_normalized[i])
else:
eng_score = DEFAULT_ENGAGEMENT
item.subs = schema.SubScores(
relevance=rel_score,
recency=rec_score,
engagement=eng_score,
)
overall = (
WEIGHT_RELEVANCE * rel_score +
WEIGHT_RECENCY * rec_score +
WEIGHT_ENGAGEMENT * eng_score
)
if eng_raw[i] is None:
overall -= UNKNOWN_ENGAGEMENT_PENALTY
item.score = max(0, min(100, int(overall)))
return items
def compute_polymarket_engagement_raw(engagement: Optional[schema.Engagement]) -> Optional[float]:
"""Compute raw engagement score for Polymarket item.
scripts/lib/truthsocial.py
+183
View File
@@ -0,0 +1,183 @@
"""Truth Social search via Mastodon-compatible API (requires bearer token).
Uses truthsocial.com/api/v2/search endpoint.
Requires TRUTHSOCIAL_TOKEN env var (bearer token from browser dev tools).
"""
import math
import re
import sys
from typing import Any, Dict, List, Optional
from . import http
TRUTHSOCIAL_SEARCH_URL = "https://truthsocial.com/api/v2/search"
DEPTH_CONFIG = {
"quick": 15,
"default": 30,
"deep": 60,
}
def _log(msg: str):
"""Log to stderr (only in TTY mode to avoid cluttering Claude Code output)."""
if sys.stderr.isatty():
sys.stderr.write(f"[TruthSocial] {msg}\n")
sys.stderr.flush()
def _strip_html(html: str) -> str:
"""Strip HTML tags from Truth Social post content."""
text = re.sub(r'<br\s*/?>', '\n', html)
text = re.sub(r'<[^>]+>', '', text)
return text.strip()
def _extract_core_subject(topic: str) -> str:
"""Extract core subject from verbose query for Truth Social search."""
text = topic.lower().strip()
prefixes = [
'what are the best', 'what is the best', 'what are the latest',
'what are people saying about', 'what do people think about',
'how do i use', 'how to use', 'how to',
'what are', 'what is', 'tips for', 'best practices for',
]
for p in prefixes:
if text.startswith(p + ' '):
text = text[len(p):].strip()
noise = {
'best', 'top', 'good', 'great', 'awesome',
'latest', 'new', 'news', 'update', 'updates',
'trending', 'hottest', 'popular', 'viral',
'practices', 'features', 'recommendations', 'advice',
}
words = text.split()
filtered = [w for w in words if w not in noise]
result = ' '.join(filtered) if filtered else text
return result.rstrip('?!.')
def _parse_date(status: Dict[str, Any]) -> Optional[str]:
"""Parse date from Mastodon status to YYYY-MM-DD.
Mastodon uses ISO 8601 format in created_at field.
"""
val = status.get("created_at")
if val and isinstance(val, str) and len(val) >= 10:
return val[:10]
return None
def search_truthsocial(
topic: str,
from_date: str,
to_date: str,
depth: str = "default",
config: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Search Truth Social via Mastodon-compatible API.
Args:
topic: Search topic
from_date: Start date (YYYY-MM-DD)
to_date: End date (YYYY-MM-DD)
depth: 'quick', 'default', or 'deep'
config: Config dict with TRUTHSOCIAL_TOKEN
Returns:
Dict with 'statuses' list from Mastodon API response.
"""
config = config or {}
token = config.get("TRUTHSOCIAL_TOKEN", "")
if not token:
return {"statuses": [], "error": "Truth Social token not configured"}
count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
core_topic = _extract_core_subject(topic)
_log(f"Searching for '{core_topic}' (depth={depth}, limit={count})")
from urllib.parse import urlencode
params = {
"q": core_topic,
"type": "statuses",
"limit": str(min(count, 40)),
}
url = f"{TRUTHSOCIAL_SEARCH_URL}?{urlencode(params)}"
try:
response = http.request(
"GET", url,
headers={"Authorization": f"Bearer {token}"},
timeout=30,
)
except http.HTTPError as e:
if e.status_code == 401:
_log("Token expired")
return {"statuses": [], "error": "Truth Social token expired"}
elif e.status_code == 403:
_log("Access denied (Cloudflare)")
return {"statuses": [], "error": "Truth Social access denied (Cloudflare)"}
elif e.status_code == 429:
_log("Rate limited")
return {"statuses": [], "error": "Truth Social rate limited"}
else:
_log(f"Search failed: {e}")
return {"statuses": [], "error": f"Truth Social search failed: {e.status_code}"}
except Exception as e:
_log(f"Search failed: {e}")
return {"statuses": [], "error": str(e)}
statuses = response.get("statuses", [])
_log(f"Found {len(statuses)} posts")
return response
def parse_truthsocial_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Parse Mastodon API response into normalized item dicts.
Returns:
List of item dicts ready for normalization.
"""
statuses = response.get("statuses", [])
items = []
for i, status in enumerate(statuses):
content_html = status.get("content") or ""
text = _strip_html(content_html)
account = status.get("account") or {}
handle = account.get("acct") or account.get("username") or ""
display_name = account.get("display_name") or handle
url = status.get("url") or ""
likes = status.get("favourites_count") or 0
reposts = status.get("reblogs_count") or 0
replies = status.get("replies_count") or 0
date_str = _parse_date(status)
# Relevance: position-based (search results are ranked by relevance)
rank_score = max(0.3, 1.0 - (i * 0.02))
engagement_boost = min(0.2, math.log1p(likes + reposts) / 40)
relevance = min(1.0, rank_score * 0.7 + engagement_boost + 0.1)
items.append({
"handle": handle,
"display_name": display_name,
"text": text,
"url": url,
"date": date_str,
"engagement": {
"likes": likes,
"reposts": reposts,
"replies": replies,
},
"relevance": round(relevance, 2),
"why_relevant": f"Truth Social: @{handle}: {text[:60]}" if text else f"Truth Social: {handle}",
})
return items