Files
notplanned/scout.py
T

151 lines
6.6 KiB
Python

#!/usr/bin/env python3
import argparse
import datetime as dt
import html
import json
import math
import pathlib
import urllib.request
import urllib.error
ROOT = pathlib.Path(__file__).resolve().parent
WATCHED = [
"openai/codex",
"anthropics/claude-code",
"vercel-labs/agent-browser",
"browser-use/browser-use",
"block/goose",
"modelcontextprotocol/servers",
]
KEYWORDS = {
"browser": 9, "upload": 12, "file": 3, "preview": 10, "localhost": 8,
"session": 7, "context": 9, "compact": 10, "goal": 10, "plan": 7,
"mcp": 9, "sandbox": 10, "permission": 9, "approval": 8, "guardrail": 10,
"model": 5, "effort": 8, "preset": 10, "workflow": 6, "task": 5,
"project": 5, "sidebar": 5, "hook": 7, "export": 9, "integration": 5,
"extension": 7, "plugin": 8, "api": 5, "cli": 4, "agent": 5,
}
def api(url):
req = urllib.request.Request(url, headers={
"User-Agent": "notplanned-scout/0.1",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
})
with urllib.request.urlopen(req, timeout=15) as r:
return json.load(r)
def score_issue(issue, now):
text = (issue.get("title","") + " " + (issue.get("body") or "")).lower()
closed = issue.get("closed_at") or issue.get("updated_at")
try:
when = dt.datetime.fromisoformat(closed.replace("Z","+00:00"))
age = max(0, (now - when).days)
except Exception:
age = 365
recency = max(0, 90 - age)
comments = min(24, int(issue.get("comments") or 0) * 2)
reactions = issue.get("reactions") or {}
reaction_score = min(24, int(reactions.get("total_count") or 0) * 3)
keyword_score = sum(weight for word, weight in KEYWORDS.items() if word in text)
return recency + comments + reaction_score + min(48, keyword_score)
def suggested_move(title):
t = title.lower()
if "upload" in t or "file input" in t:
return "Ship a deterministic upload bridge outside upstream."
if "preview" in t or "localhost" in t:
return "Ship a preview/port sidecar and test it on real agent-built apps."
if "context" in t or "compact" in t:
return "Build a transcript/context sidecar; measure saved tokens and resumability."
if "goal" in t or "session" in t:
return "Wrap session state externally; prove persistence across runs."
if "model" in t or "effort" in t or "preset" in t:
return "Compose the existing knobs into one named preset action."
if "permission" in t or "sandbox" in t or "approval" in t or "guardrail" in t:
return "Put a narrow authority policy in front of the existing agent."
if "browser" in t or "domain" in t or "form" in t:
return "Wrap browser actions with scoped policy + receipts."
if "project" in t or "task" in t or "workflow" in t:
return "Ship an external supervisor instead of waiting for upstream PM."
return "Find the smallest external wrapper that satisfies the request; test one user."
def esc(s):
return html.escape(str(s), quote=True)
def curated_card(o):
return f"""<a class="card" href="{esc(o['source'])}" target="_blank" rel="noreferrer">
<div class="topline"><span class="repo">{esc(o['repo'])} · #{esc(o['issue'])}</span><span class="heat {esc(o['heat'])}">{esc(o['heat'])}</span></div>
<h3>{esc(o['name'])}</h3>
<div class="issue-title">{esc(o['title'])}</div>
<p class="thesis">{esc(o['thesis'])}</p>
<div class="kv">
<div class="row"><span class="key">signal</span><span>{esc(o['signal'])}</span></div>
<div class="row"><span class="key">move</span><span class="go">{esc(o['move'])}</span></div>
</div>
</a>"""
def fresh_card(o):
heat = "hot" if o["score"] >= 105 else ("warm" if o["score"] >= 75 else "watch")
return f"""<a class="fresh-card" href="{esc(o['html_url'])}" target="_blank" rel="noreferrer">
<div class="topline"><span class="repo">{esc(o['repo'])} · #{esc(o['number'])}</span><span class="heat {heat}">{heat}</span></div>
<h4>{esc(o['title'])}</h4>
<div class="fresh-meta">auto score {o['score']} · {esc(o.get('closed_at','')[:10])}</div>
<div class="fresh-move">{esc(suggested_move(o['title']))}</div>
</a>"""
def find_fresh(curated_urls, limit=6):
now = dt.datetime.now(dt.timezone.utc)
found = []
for repo in WATCHED:
url = f"https://api.github.com/repos/{repo}/issues?state=closed&sort=updated&direction=desc&per_page=100"
try:
items = api(url)
except Exception as e:
print(f"scout warning: {repo}: {type(e).__name__}", flush=True)
continue
for issue in items:
if "pull_request" in issue:
continue
if issue.get("state_reason") != "not_planned":
continue
if issue.get("html_url") in curated_urls:
continue
score = score_issue(issue, now)
if score < 45:
continue
found.append({
"repo": repo,
"number": issue.get("number"),
"title": issue.get("title",""),
"html_url": issue.get("html_url",""),
"closed_at": issue.get("closed_at") or issue.get("updated_at") or "",
"score": score,
})
found.sort(key=lambda x: (x["score"], x["closed_at"]), reverse=True)
return found[:limit]
def render(out_path, no_network=False):
data = json.loads((ROOT / "opportunities.json").read_text())
curated = data["opportunities"]
curated_urls = {x["source"] for x in curated}
fresh = [] if no_network else find_fresh(curated_urls)
template = (ROOT / "template.html").read_text()
now = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M")
page = (template
.replace("__CURATED_COUNT__", str(len(curated)))
.replace("__FRESH_COUNT__", str(len(fresh)))
.replace("__UPDATED__", now)
.replace("__CURATED_CARDS__", "\n".join(curated_card(x) for x in curated))
.replace("__FRESH_CARDS__", "\n".join(fresh_card(x) for x in fresh) if fresh else '<div class="fresh-card"><h4>scout pass quiet</h4><div class="fresh-move">No machine-surfaced candidate cleared the threshold on this pass. That is a valid result.</div></div>')
)
pathlib.Path(out_path).write_text(page)
print(json.dumps({"curated": len(curated), "fresh": len(fresh), "out": str(out_path)}))
if __name__ == "__main__":
p = argparse.ArgumentParser()
p.add_argument("--out", default=str(ROOT / "index.html"))
p.add_argument("--no-network", action="store_true")
args = p.parse_args()
render(args.out, args.no_network)