notplanned/scout.py

#!/usr/bin/env python3
import argparse, datetime as dt, html, json, math, pathlib, urllib.request, urllib.error

ROOT = pathlib.Path(__file__).resolve().parent

def api(url):
    req = urllib.request.Request(url, headers={
        "User-Agent": "notplanned-radar/0.2",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    })
    with urllib.request.urlopen(req, timeout=20) as r:
        return json.load(r)

def esc(v): return html.escape(str(v), quote=True)

def repo_heat(repo, now):
    pushed = repo.get("pushed_at") or repo.get("updated_at")
    try:
        age = max(0, (now - dt.datetime.fromisoformat(pushed.replace("Z","+00:00"))).days)
    except Exception:
        age = 999
    activity = 15 if age <= 7 else 10 if age <= 30 else 5 if age <= 90 else 0
    stars = int(repo.get("stargazers_count") or 0)
    star_score = min(15, round(math.log10(max(stars, 1)) * 3.2))
    return activity + star_score, {"repo_activity": activity, "repo_scale": star_score}

def issue_score(issue, repo_meta, cfg, now):
    labels = {str(x.get("name","")).strip().lower() for x in issue.get("labels", [])}
    if labels & {x.lower() for x in cfg["exclude_labels"]}:
        return None
    rejection = 0
    reason = issue.get("state_reason")
    if reason == "not_planned":
        rejection = max(rejection, 32)
    for label, pts in cfg["rejection_labels"].items():
        if label.lower() in labels:
            rejection = max(rejection, pts)
    if rejection == 0:
        return None

    closed = issue.get("closed_at") or issue.get("updated_at")
    try:
        age = max(0, (now - dt.datetime.fromisoformat(closed.replace("Z","+00:00"))).days)
    except Exception:
        age = 999
    recency = 40 if age <= 7 else 30 if age <= 30 else 16 if age <= 90 else 6 if age <= 180 else 0

    comments = min(20, int(issue.get("comments") or 0) * 2)
    reactions = issue.get("reactions") or {}
    reaction_score = min(20, int(reactions.get("total_count") or 0) * 2)

    title = (issue.get("title") or "").lower()
    body = (issue.get("body") or "").lower()
    text = title + "\n" + body[:4000]
    ext = 0
    matched = []
    for word, pts in cfg["externalizable_keywords"].items():
        if word.lower() in text:
            ext += pts
            matched.append(word)
    ext = min(28, ext)

    heat, heat_breakdown = repo_heat(repo_meta, now)
    breakdown = {
        "rejection": rejection,
        "recency": recency,
        "engagement": comments + reaction_score,
        "externalizable": ext,
        **heat_breakdown,
    }
    score = sum(breakdown.values())
    return {
        "score": score,
        "breakdown": breakdown,
        "age_days": age,
        "matched": matched[:6],
    }

def category(title):
    t = title.lower()
    buckets = [
        ("browser", ["browser","upload","preview","localhost","tab","form"]),
        ("context", ["context","compact","session","memory","history"]),
        ("authority", ["permission","approval","sandbox","security","guardrail"]),
        ("integration", ["mcp","api","webhook","integration","plugin","extension"]),
        ("interface", ["ui","sidebar","dashboard","preset","menu","export"]),
        ("workflow", ["task","project","workflow","plan","goal"]),
    ]
    for name, words in buckets:
        if any(w in t for w in words): return name
    return "wrapper"

def move_for(cat):
    return {
        "browser":"external browser sidecar / extension",
        "context":"session/context sidecar",
        "authority":"policy proxy / scoped authority layer",
        "integration":"adapter / MCP / integration bridge",
        "interface":"thin UI / control surface",
        "workflow":"supervisor / workflow wrapper",
        "wrapper":"thin wrapper around existing capability",
    }[cat]

def card(o):
    b=o["breakdown"]
    why=f"rej {b['rejection']} · rec {b['recency']} · eng {b['engagement']} · ext {b['externalizable']} · heat {b['repo_activity']+b['repo_scale']}"
    return f"""<a class="card" href="{esc(o['html_url'])}" target="_blank" rel="noreferrer">
      <div class="topline"><span class="repo">{esc(o['repo'])} · #{o['number']}</span><span class="score">{o['score']}</span></div>
      <h3>{esc(o['title'])}</h3>
      <div class="meta">{esc(o['category'])} · closed {o['age_days']}d ago</div>
      <p class="move">{esc(o['move'])}</p>
      <div class="why">{esc(why)}</div>
    </a>"""

def scan():
    cfg=json.loads((ROOT/"radar.json").read_text())
    now=dt.datetime.now(dt.timezone.utc)
    rows=[]
    for repo_name in cfg["repos"]:
        try:
            repo=api(f"https://api.github.com/repos/{repo_name}")
            issues=api(f"https://api.github.com/repos/{repo_name}/issues?state=closed&sort=updated&direction=desc&per_page=100")
        except Exception as e:
            print(f"warning {repo_name}: {type(e).__name__}")
            continue
        for issue in issues:
            if "pull_request" in issue: continue
            scored=issue_score(issue, repo, cfg, now)
            if not scored or scored["score"] < cfg["threshold"]: continue
            cat=category(issue.get("title",""))
            rows.append({
                "repo":repo_name,
                "number":issue.get("number"),
                "title":issue.get("title",""),
                "html_url":issue.get("html_url",""),
                "closed_at":issue.get("closed_at") or "",
                "category":cat,
                "move":move_for(cat),
                **scored,
            })
    rows.sort(key=lambda x:(x["score"], x["closed_at"]), reverse=True)
    return cfg, rows[:cfg["limit"]]

def render(out_path):
    cfg, rows=scan()
    tpl=(ROOT/"template.html").read_text()
    now=dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M")
    html_out=(tpl
      .replace("__COUNT__",str(len(rows)))
      .replace("__UPDATED__",now)
      .replace("__THRESHOLD__",str(cfg["threshold"]))
      .replace("__REPOS__",str(len(cfg["repos"])))
      .replace("__CARDS__","\n".join(card(x) for x in rows) if rows else '<div class="empty">No candidate cleared the deterministic threshold on this pass.</div>')
    )
    pathlib.Path(out_path).write_text(html_out)
    print(json.dumps({"candidates":len(rows),"threshold":cfg["threshold"],"repos":len(cfg["repos"]),"out":str(out_path)}))

if __name__=="__main__":
    p=argparse.ArgumentParser()
    p.add_argument("--out",default=str(ROOT/"index.html"))
    a=p.parse_args()
    render(a.out)