← back to Last30 Skill
skills/last30/scripts/last30.py
613 lines
#!/usr/bin/env python3
"""
last30 — engagement-ranked, recency-filtered multi-source research engine.
Researches what people actually said about a topic in the last N days across
FREE, no-key public APIs, scores each item by engagement x recency, and emits
a structured, cited result set for an agent to synthesize.
Zero third-party dependencies — Python 3.9+ standard library only. This is the
deterministic core of the /last30 skill; the SKILL.md contract decides intent,
runs this engine, and writes the human brief.
Sources (all free, no API key):
reddit Reddit search + top comments (upvotes)
hn Hacker News via Algolia (points)
arxiv arXiv Atom API (recency)
github GitHub repo search (stars)
polymarket Polymarket Gamma public-search (odds / $ volume)
Usage:
last30.py "<topic>" research the topic (all free sources)
last30.py "<topic>" --days 7 narrow the window to 7 days
last30.py "<topic>" --sources reddit,hn only these sources
last30.py "<topic>" --emit markdown human-readable instead of JSON
last30.py "<topic>" --no-comments skip Reddit comment fetching (faster)
last30.py doctor health-check every source
Inspired by mvanhorn/last30days-skill (MIT). This is an independent, no-key,
$0-local reimplementation — see references/sources.md for the lineage.
"""
from __future__ import annotations
import argparse
import concurrent.futures
import json
import math
import re
import sys
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
# Reddit blocklists descriptive scraper UAs (403 Blocked), so present as a
# standard desktop browser. From a residential IP this passes the public
# JSON endpoints for every source below.
USER_AGENT = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
# Engine default = the clean no-key JSON APIs that work from any IP.
# Reddit is available (--sources reddit,...) but ships OFF by default: its
# unauthenticated JSON is IP-blocked on many networks (403 Blocked), so the
# SKILL.md contract covers Reddit + other social platforms via the agent's
# own host web-search tools instead. See references/sources.md.
DEFAULT_SOURCES = ["hn", "arxiv", "github", "polymarket"]
ALL_SOURCES = ["reddit", "hn", "arxiv", "github", "polymarket"]
HTTP_TIMEOUT = 20 # seconds per request
# --------------------------------------------------------------------------- #
# HTTP helpers #
# --------------------------------------------------------------------------- #
def _get(url: str, headers: dict | None = None, timeout: int = HTTP_TIMEOUT) -> bytes:
"""GET a URL with a browser-ish UA. Raises on network/HTTP error."""
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})})
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.read()
def _get_json(url: str, headers: dict | None = None, timeout: int = HTTP_TIMEOUT):
return json.loads(_get(url, headers, timeout).decode("utf-8", "replace"))
def _now() -> datetime:
return datetime.now(timezone.utc)
def _since_ts(days: int) -> int:
return int(_now().timestamp()) - days * 86400
def _age_days(ts: float) -> float:
return max(0.0, (_now().timestamp() - ts) / 86400.0)
def _recency_weight(age_days: float, window_days: int) -> float:
"""Linear decay from 1.0 (now) to ~0.15 at the edge of the window."""
if window_days <= 0:
return 1.0
return max(0.15, 1.0 - 0.85 * (age_days / window_days))
def _clip(text: str, n: int = 280) -> str:
text = re.sub(r"\s+", " ", (text or "")).strip()
return text if len(text) <= n else text[: n - 1] + "…"
# --------------------------------------------------------------------------- #
# Source fetchers — each returns {"source", "items": [...]} or {"error": ...} #
# Every fetcher is fault-isolated: it must never raise past its own boundary. #
# --------------------------------------------------------------------------- #
def fetch_reddit(topic: str, days: int, limit: int, comments: bool) -> dict:
"""Reddit search (top posts in window) + a couple of top comments each."""
try:
q = urllib.parse.quote(topic)
window = "week" if days <= 7 else ("month" if days <= 31 else "year")
url = (
f"https://www.reddit.com/search.json?q={q}"
f"&sort=top&t={window}&limit={min(limit * 2, 50)}"
)
data = _get_json(url)
cutoff = _since_ts(days)
items = []
for child in data.get("data", {}).get("children", []):
p = child.get("data", {})
created = p.get("created_utc", 0)
if created < cutoff:
continue
ups = int(p.get("ups", 0) or 0)
permalink = "https://www.reddit.com" + p.get("permalink", "")
item = {
"title": _clip(p.get("title", ""), 200),
"url": permalink,
"external_url": p.get("url_overridden_by_dest") or p.get("url"),
"subreddit": "r/" + p.get("subreddit", ""),
"engagement": ups,
"engagement_label": f"{ups:,} upvotes",
"comment_count": int(p.get("num_comments", 0) or 0),
"created_utc": created,
"age_days": round(_age_days(created), 1),
"snippet": _clip(p.get("selftext", ""), 300),
"top_comments": [],
}
items.append(item)
items.sort(key=lambda x: x["engagement"], reverse=True)
items = items[:limit]
if comments:
for it in items[: min(len(items), 6)]: # only enrich the strongest
try:
c_url = it["url"].rstrip("/") + ".json?limit=3&sort=top"
cdata = _get_json(c_url)
if len(cdata) > 1:
for cc in cdata[1].get("data", {}).get("children", [])[:2]:
cd = cc.get("data", {})
body = cd.get("body")
if body:
it["top_comments"].append(
{
"text": _clip(body, 240),
"upvotes": int(cd.get("ups", 0) or 0),
}
)
except Exception:
pass # comment enrichment is best-effort
return {"source": "reddit", "items": items}
except Exception as e:
hint = ""
if "403" in str(e):
hint = " (Reddit blocks unauthenticated JSON on many IPs — " \
"the agent should cover Reddit via host web search instead)"
return {"source": "reddit", "error": f"{type(e).__name__}: {e}{hint}"}
def fetch_hn(topic: str, days: int, limit: int) -> dict:
"""Hacker News stories in the window, ranked by points, via Algolia."""
try:
q = urllib.parse.quote(topic)
cutoff = _since_ts(days)
url = (
f"https://hn.algolia.com/api/v1/search?query={q}"
f"&tags=story&numericFilters=created_at_i>{cutoff}&hitsPerPage={limit}"
)
data = _get_json(url)
items = []
for h in data.get("hits", []):
created = h.get("created_at_i", 0) or 0
points = int(h.get("points", 0) or 0)
obj_id = h.get("objectID")
items.append(
{
"title": _clip(h.get("title") or h.get("story_title") or "", 200),
"url": h.get("url") or f"https://news.ycombinator.com/item?id={obj_id}",
"discussion_url": f"https://news.ycombinator.com/item?id={obj_id}",
"engagement": points,
"engagement_label": f"{points:,} points",
"comment_count": int(h.get("num_comments", 0) or 0),
"author": h.get("author"),
"created_utc": created,
"age_days": round(_age_days(created), 1),
"snippet": _clip(h.get("story_text") or "", 300),
}
)
items.sort(key=lambda x: x["engagement"], reverse=True)
return {"source": "hn", "items": items[:limit]}
except Exception as e:
return {"source": "hn", "error": f"{type(e).__name__}: {e}"}
def fetch_arxiv(topic: str, days: int, limit: int) -> dict:
"""Recent arXiv papers matching the topic (Atom API)."""
try:
q = urllib.parse.quote(f"all:{topic}")
url = (
f"http://export.arxiv.org/api/query?search_query={q}"
f"&sortBy=submittedDate&sortOrder=descending&max_results={limit * 2}"
)
raw = _get(url)
ns = {"a": "http://www.w3.org/2005/Atom"}
root = ET.fromstring(raw)
cutoff = _since_ts(days)
items = []
for entry in root.findall("a:entry", ns):
pub = entry.findtext("a:published", default="", namespaces=ns)
try:
ts = datetime.strptime(pub, "%Y-%m-%dT%H:%M:%SZ").replace(
tzinfo=timezone.utc
).timestamp()
except ValueError:
continue
if ts < cutoff:
continue
title = entry.findtext("a:title", default="", namespaces=ns)
link = entry.findtext("a:id", default="", namespaces=ns)
summary = entry.findtext("a:summary", default="", namespaces=ns)
authors = [
a.findtext("a:name", default="", namespaces=ns)
for a in entry.findall("a:author", ns)
]
items.append(
{
"title": _clip(title, 220),
"url": link,
"engagement": 0,
"engagement_label": "new paper",
"authors": authors[:4],
"created_utc": ts,
"age_days": round(_age_days(ts), 1),
"snippet": _clip(summary, 320),
}
)
items.sort(key=lambda x: x["created_utc"], reverse=True)
return {"source": "arxiv", "items": items[:limit]}
except Exception as e:
return {"source": "arxiv", "error": f"{type(e).__name__}: {e}"}
def fetch_github(topic: str, days: int, limit: int) -> dict:
"""GitHub repos pushed within the window, ranked by stars (unauth search)."""
try:
since = datetime.fromtimestamp(_since_ts(days), timezone.utc).strftime("%Y-%m-%d")
q = urllib.parse.quote(f"{topic} pushed:>={since}")
url = (
f"https://api.github.com/search/repositories?q={q}"
f"&sort=stars&order=desc&per_page={limit}"
)
data = _get_json(url, headers={"Accept": "application/vnd.github+json"})
items = []
for r in data.get("items", []):
stars = int(r.get("stargazers_count", 0) or 0)
pushed = r.get("pushed_at", "")
try:
ts = datetime.strptime(pushed, "%Y-%m-%dT%H:%M:%SZ").replace(
tzinfo=timezone.utc
).timestamp()
except ValueError:
ts = _now().timestamp()
items.append(
{
"title": _clip(r.get("full_name", ""), 120),
"url": r.get("html_url"),
"engagement": stars,
"engagement_label": f"{stars:,} stars",
"language": r.get("language"),
"created_utc": ts,
"age_days": round(_age_days(ts), 1),
"snippet": _clip(r.get("description") or "", 280),
}
)
return {"source": "github", "items": items[:limit]}
except Exception as e:
return {"source": "github", "error": f"{type(e).__name__}: {e}"}
def fetch_polymarket(topic: str, days: int, limit: int) -> dict:
"""Active Polymarket markets matching the topic — odds backed by real money."""
try:
q = urllib.parse.quote(topic)
url = (
f"https://gamma-api.polymarket.com/public-search?q={q}"
f"&limit_per_type={limit}&events_status=active"
)
data = _get_json(url)
events = data.get("events", []) if isinstance(data, dict) else []
items = []
for ev in events[:limit]:
vol = float(ev.get("volume", 0) or 0)
markets = ev.get("markets", []) or []
odds = []
for m in markets[:4]:
outcome = m.get("groupItemTitle") or m.get("question") or ""
price = m.get("outcomePrices") or m.get("lastTradePrice")
if isinstance(price, str):
try:
price = json.loads(price)
except Exception:
price = None
if isinstance(price, list) and price:
try:
odds.append(f"{outcome}: {round(float(price[0]) * 100)}%")
except Exception:
pass
elif isinstance(price, (int, float)):
odds.append(f"{outcome}: {round(float(price) * 100)}%")
items.append(
{
"title": _clip(ev.get("title", ""), 200),
"url": "https://polymarket.com/event/" + (ev.get("slug", "")),
"engagement": int(vol),
"engagement_label": f"${vol:,.0f} volume",
"odds": odds,
"created_utc": _now().timestamp(),
"age_days": 0.0,
"snippet": _clip(ev.get("description") or "", 240),
}
)
items.sort(key=lambda x: x["engagement"], reverse=True)
return {"source": "polymarket", "items": items[:limit]}
except Exception as e:
return {"source": "polymarket", "error": f"{type(e).__name__}: {e}"}
FETCHERS = {
"reddit": lambda t, d, l, c: fetch_reddit(t, d, l, c),
"hn": lambda t, d, l, c: fetch_hn(t, d, l),
"arxiv": lambda t, d, l, c: fetch_arxiv(t, d, l),
"github": lambda t, d, l, c: fetch_github(t, d, l),
"polymarket": lambda t, d, l, c: fetch_polymarket(t, d, l),
}
# --------------------------------------------------------------------------- #
# Relevance gate — kill keyword collisions (e.g. Kimi-the-model vs Kimi-the- #
# F1-driver) before scoring, using topic-token coverage. Dependency-free. #
# --------------------------------------------------------------------------- #
_STOPWORDS = {
"the", "a", "an", "of", "and", "or", "for", "in", "on", "to", "is", "are",
"vs", "v", "versus", "with", "about", "what", "whats", "people", "saying",
"say", "latest", "recent", "buzz", "current", "state", "new", "news", "this",
"that", "how", "why", "best", "top", "review", "reviews", "reaction",
}
def _topic_tokens(topic: str) -> list[str]:
"""Distinctive lowercase tokens from the topic (stopwords/separators dropped)."""
raw = re.split(r"[^a-z0-9]+", topic.lower())
seen, out = set(), []
for t in raw:
if len(t) >= 2 and t not in _STOPWORDS and t not in seen:
seen.add(t)
out.append(t)
return out
def _item_haystack(it: dict) -> str:
parts = [
str(it.get("title", "")), str(it.get("snippet", "")),
str(it.get("subreddit", "")), str(it.get("language", "")),
" ".join(it.get("authors", []) or []),
" ".join(it.get("odds", []) or []),
]
return " ".join(parts).lower()
def apply_relevance(results: list[dict], topic: str, threshold: float) -> None:
"""Filter each source's items by topic-token coverage, in-place.
An item is kept if it matches at least `ceil(len(tokens) * threshold)`
distinct topic tokens (min 1). Every kept item gets a `relevance` fraction;
each source result records `filtered_out` for transparency. A single-token
topic keeps anything containing that token. `threshold <= 0` disables.
"""
tokens = _topic_tokens(topic)
if not tokens or threshold <= 0:
for r in results:
for it in r.get("items", []):
it["relevance"] = 1.0
return
need = max(1, math.ceil(len(tokens) * threshold))
for r in results:
items = r.get("items")
if not items:
continue
kept = []
dropped = 0
for it in items:
hay = _item_haystack(it)
matched = sum(1 for tk in tokens if tk in hay)
it["relevance"] = round(matched / len(tokens), 2)
if matched >= need:
kept.append(it)
else:
dropped += 1
r["items"] = kept
if dropped:
r["filtered_out"] = dropped
# --------------------------------------------------------------------------- #
# Cross-source scoring #
# --------------------------------------------------------------------------- #
def score_items(results: list[dict], window_days: int) -> None:
"""Attach a normalized 0-100 `score` to every item in-place.
Per source we normalize engagement to its own max (so 5k HN points and
50k reddit upvotes are comparable), then multiply by a recency weight.
arXiv/polymarket carry engagement of 0/volume — recency dominates there.
"""
for r in results:
items = r.get("items", [])
if not items:
continue
peak = max((it.get("engagement", 0) for it in items), default=0) or 1
for it in items:
eng_norm = (it.get("engagement", 0) / peak) if peak else 0.0
rec = _recency_weight(it.get("age_days", 0.0), window_days)
# engagement-led sources weight engagement 0.7 / recency 0.3;
# discovery sources (no engagement signal) lean fully on recency.
if peak > 1:
score = 100 * (0.7 * eng_norm + 0.3 * rec)
else:
score = 100 * rec
it["score"] = round(score, 1)
def top_across(results: list[dict], n: int) -> list[dict]:
"""Flatten every source's items and return the global top-N by score."""
flat = []
for r in results:
for it in r.get("items", []):
flat.append({**it, "source": r["source"]})
flat.sort(key=lambda x: x.get("score", 0), reverse=True)
return flat[:n]
# --------------------------------------------------------------------------- #
# Emitters #
# --------------------------------------------------------------------------- #
def emit_json(topic, days, sources, results, elapsed):
scored_sources = {}
for r in results:
if "error" in r:
scored_sources[r["source"]] = {"error": r["error"], "items": []}
else:
scored_sources[r["source"]] = {
"count": len(r["items"]),
"filtered_out": r.get("filtered_out", 0),
"items": r["items"],
}
payload = {
"topic": topic,
"window_days": days,
"generated_at": _now().isoformat(),
"elapsed_seconds": round(elapsed, 1),
"sources_requested": sources,
"sources": scored_sources,
"top_overall": top_across(results, 15),
}
print(json.dumps(payload, indent=2, ensure_ascii=False))
def emit_markdown(topic, days, sources, results, elapsed):
out = [f"# /last30 — “{topic}” (last {days} days)", ""]
total_filtered = sum(r.get("filtered_out", 0) for r in results)
filt = f" · {total_filtered} off-topic filtered" if total_filtered else ""
out.append(f"_generated {_now().strftime('%Y-%m-%d %H:%M UTC')} · {elapsed:.1f}s · "
f"sources: {', '.join(sources)}{filt}_")
out.append("")
top = top_across(results, 12)
if top:
out.append("## Top across all sources")
out.append("")
for i, it in enumerate(top, 1):
out.append(
f"{i}. **[{it['source']}·{it.get('score',0)}]** "
f"[{it['title']}]({it.get('url','')}) — "
f"{it.get('engagement_label','')}, {it.get('age_days','?')}d ago"
)
if it.get("odds"):
out.append(f" - odds: {', '.join(it['odds'])}")
if it.get("snippet"):
out.append(f" - {it['snippet']}")
out.append("")
for r in results:
out.append(f"## {r['source']}")
if "error" in r:
out.append(f"> ⚠️ unavailable — {r['error']}")
out.append("")
continue
if not r["items"]:
out.append("> (nothing in the window)")
out.append("")
continue
for it in r["items"]:
out.append(
f"- [{it['title']}]({it.get('url','')}) · "
f"{it.get('engagement_label','')} · {it.get('age_days','?')}d"
)
for c in it.get("top_comments", []):
out.append(f" - 💬 ({c['upvotes']}▲) {c['text']}")
out.append("")
print("\n".join(out))
# --------------------------------------------------------------------------- #
# Doctor #
# --------------------------------------------------------------------------- #
def doctor():
print("last30 doctor — probing free sources (no keys required)\n")
probes = {
"reddit": lambda: fetch_reddit("openai", 30, 3, False),
"hn": lambda: fetch_hn("openai", 30, 3),
"arxiv": lambda: fetch_arxiv("language model", 30, 3),
"github": lambda: fetch_github("agent", 30, 3),
"polymarket": lambda: fetch_polymarket("election", 90, 3),
}
hard_fail = False
for name, fn in probes.items():
t0 = time.time()
r = fn()
dt = time.time() - t0
if "error" in r:
# Reddit is opt-in/best-effort (IP-blocked on many networks) — its
# 403/429 is expected and never fails the doctor; the agent covers
# Reddit via host web search instead.
if name == "reddit":
code = "403" if "403" in r["error"] else ("429" if "429" in r["error"] else "err")
print(f" ⚠ {name:11s} {dt:5.1f}s blocked ({code}) — "
f"agent covers Reddit via host web search")
else:
hard_fail = True
print(f" ✗ {name:11s} {dt:5.1f}s {r['error']}")
else:
print(f" ✓ {name:11s} {dt:5.1f}s {len(r['items'])} items")
print("\nAll default sources healthy." if not hard_fail else
"\nA default source failed — likely a transient rate-limit; retry in a minute.")
return 0 if not hard_fail else 1
# --------------------------------------------------------------------------- #
# CLI #
# --------------------------------------------------------------------------- #
def main() -> int:
ap = argparse.ArgumentParser(
prog="last30",
description="Engagement-ranked, recency-filtered multi-source research.",
)
ap.add_argument("topic", nargs="?", help='research topic, or "doctor"')
ap.add_argument("--days", type=int, default=30, help="lookback window (default 30)")
ap.add_argument("--sources", default=",".join(DEFAULT_SOURCES),
help="comma-separated: " + ",".join(DEFAULT_SOURCES))
ap.add_argument("--limit", type=int, default=12, help="items per source (default 12)")
ap.add_argument("--emit", choices=["json", "markdown"], default="json")
ap.add_argument("--no-comments", action="store_true",
help="skip Reddit comment enrichment (faster)")
ap.add_argument("--min-relevance", type=float, default=0.4,
help="topic-token coverage gate 0..1 (default 0.4); "
"kills keyword collisions. 0 disables.")
ap.add_argument("--no-filter", action="store_true",
help="disable the relevance gate (same as --min-relevance 0)")
args = ap.parse_args()
if not args.topic or args.topic == "doctor":
return doctor() if args.topic == "doctor" else (ap.print_help() or 1)
sources = [s.strip() for s in args.sources.split(",") if s.strip() in FETCHERS]
if not sources:
print(f"No valid sources in '{args.sources}'. Valid: {list(FETCHERS)}",
file=sys.stderr)
return 1
t0 = time.time()
results: list[dict] = []
with concurrent.futures.ThreadPoolExecutor(max_workers=len(sources)) as ex:
futs = {
ex.submit(FETCHERS[s], args.topic, args.days, args.limit,
not args.no_comments): s
for s in sources
}
by_source = {}
for fut in concurrent.futures.as_completed(futs):
r = fut.result()
by_source[r["source"]] = r
results = [by_source[s] for s in sources if s in by_source]
threshold = 0.0 if args.no_filter else args.min_relevance
apply_relevance(results, args.topic, threshold)
score_items(results, args.days)
elapsed = time.time() - t0
if args.emit == "markdown":
emit_markdown(args.topic, args.days, sources, results, elapsed)
else:
emit_json(args.topic, args.days, sources, results, elapsed)
return 0
if __name__ == "__main__":
sys.exit(main())