notplanned/claude.py

#!/usr/bin/env python3
"""Build a bounded, deterministic view of Claude-attributed GitHub commits."""
import argparse
import datetime as dt
import html
import json
import pathlib
import re
import urllib.error
import urllib.parse
import urllib.request

ROOT = pathlib.Path(__file__).resolve().parent
LOOKBACK_DAYS = 30
MAX_RESULTS_PER_QUERY = 50
SEARCHES = [
    "author:claude[bot]",
    "committer:claude[bot]",
    "author:claude",
    "committer:claude",
    '"Co-authored-by: Claude"',
]
COAUTHOR_RE = re.compile(r"(?im)^\s*co-authored-by:\s*(.*?)\s*<([^>]+)>\s*$")


class GithubApiError(RuntimeError):
    def __init__(self, status, detail, headers):
        super().__init__(detail)
        self.status = status
        self.headers = headers


def api(url):
    req = urllib.request.Request(url, headers={
        "User-Agent": "notplanned-claude-scout/0.1",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    })
    try:
        with urllib.request.urlopen(req, timeout=20) as response:
            return json.load(response), dict(response.headers)
    except urllib.error.HTTPError as error:
        try:
            body = json.load(error)
            detail = body.get("message", "GitHub API request failed")
        except Exception:
            detail = "GitHub API request failed"
        raise GithubApiError(error.code, detail, dict(error.headers)) from error


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


def parse_date(value):
    try:
        return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
    except Exception:
        return dt.datetime.min.replace(tzinfo=dt.timezone.utc)


def age_days(value, now):
    return max(0, (now - parse_date(value)).days)


def identity_text(identity):
    if not identity:
        return ""
    return " ".join(str(identity.get(key) or "") for key in ("login", "name", "email")).lower()


def is_claude_identity(identity):
    text = identity_text(identity)
    return bool(re.search(r"(?:claude\s*\[bot\]|claude\s+code|\bclaude\b)", text))


def item_identity(item, role):
    commit_identity = (item.get("commit") or {}).get(role) or {}
    account = item.get(role) or {}
    return {
        "login": account.get("login") or "",
        "name": commit_identity.get("name") or "",
        "email": commit_identity.get("email") or "",
    }


def coauthors(message):
    return [
        {"name": name.strip(), "email": email.strip()}
        for name, email in COAUTHOR_RE.findall(message or "")
        if is_claude_identity({"name": name, "email": email})
    ]


def category(message):
    text = (message or "").lower()
    buckets = [
        ("browser", ["browser", "playwright", "puppeteer", "chrome", "screenshot"]),
        ("integration", ["mcp", "api", "webhook", "integration", "plugin", "connector"]),
        ("workflow", ["workflow", "task", "agent", "command", "hook", "automation"]),
        ("context", ["context", "memory", "session", "history", "resume", "checkpoint"]),
        ("interface", ["ui", "ux", "sidebar", "dashboard", "panel", "menu"]),
        ("testing", ["test", "spec", "fixture", "assert", "coverage"]),
        ("infra", ["ci", "build", "deploy", "docker", "release", "dependency"]),
        ("docs", ["readme", "docs", "documentation", "changelog"]),
    ]
    for name, words in buckets:
        if any(word in text for word in words):
            return name
    return "implementation"


def attribution_kind(author_match, committer_match, coauthor_match):
    kinds = []
    if author_match:
        kinds.append("author")
    if committer_match:
        kinds.append("committer")
    if coauthor_match:
        kinds.append("co-author")
    return " + ".join(kinds)


def candidate(item, query):
    commit = item.get("commit") or {}
    message = commit.get("message") or ""
    author = item_identity(item, "author")
    committer = item_identity(item, "committer")
    claude_coauthors = coauthors(message)
    author_match = is_claude_identity(author)
    committer_match = is_claude_identity(committer)
    coauthor_match = bool(claude_coauthors)
    if not (author_match or committer_match or coauthor_match):
        return None

    repository = item.get("repository") or {}
    repo = repository.get("full_name") or ""
    if not repo:
        return None
    timestamp = (committer.get("date") or author.get("date") or "")
    # GitHub returns dates inside commit.author/commit.committer, not the flattened identities.
    timestamp = (commit.get("committer") or {}).get("date") or (commit.get("author") or {}).get("date") or timestamp
    official = repo.lower().startswith("anthropics/")
    return {
        "repo": repo,
        "repo_url": repository.get("html_url") or f"https://github.com/{repo}",
        "commit_sha": item.get("sha") or "",
        "commit_url": item.get("html_url") or "",
        "commit_message": message[:4000],
        "subject": (message.splitlines() or ["(no commit message)"])[0][:240],
        "author": author,
        "committer": committer,
        "timestamp": timestamp,
        "attribution_kind": attribution_kind(author_match, committer_match, coauthor_match),
        "coauthors": claude_coauthors,
        "category": category(message),
        "ecosystem": "official Anthropic" if official else "independent",
        "official_anthropic": official,
        "matched_queries": [query],
        "_author_match": author_match,
        "_committer_match": committer_match,
        "_coauthor_match": coauthor_match,
    }


def merge(existing, incoming):
    for field in ("_author_match", "_committer_match", "_coauthor_match"):
        existing[field] = existing[field] or incoming[field]
    for query in incoming["matched_queries"]:
        if query not in existing["matched_queries"]:
            existing["matched_queries"].append(query)
    if not existing["coauthors"] and incoming["coauthors"]:
        existing["coauthors"] = incoming["coauthors"]
    existing["attribution_kind"] = attribution_kind(
        existing["_author_match"], existing["_committer_match"], existing["_coauthor_match"]
    )


def rate_snapshot(headers):
    if not headers:
        return {}
    return {
        key: headers.get(value)
        for key, value in (
            ("limit", "X-RateLimit-Limit"),
            ("remaining", "X-RateLimit-Remaining"),
            ("reset", "X-RateLimit-Reset"),
        )
        if headers.get(value) is not None
    }


def build():
    now = dt.datetime.now(dt.timezone.utc)
    cutoff = (now - dt.timedelta(days=LOOKBACK_DAYS)).date().isoformat()
    rows = {}
    errors = []
    query_log = []
    last_rate = {}
    filtered_future = 0

    for term in SEARCHES:
        today = now.date().isoformat()
        query = f"{term} committer-date:>={cutoff} committer-date:<={today}"
        url = "https://api.github.com/search/commits?" + urllib.parse.urlencode({
            "q": query,
            "sort": "committer-date",
            "order": "desc",
            "per_page": MAX_RESULTS_PER_QUERY,
        })
        try:
            payload, headers = api(url)
            last_rate = rate_snapshot(headers) or last_rate
            items = payload.get("items", [])
            query_log.append({"term": term, "query": query, "returned": len(items), "total_count": payload.get("total_count", 0)})
        except GithubApiError as error:
            last_rate = rate_snapshot(error.headers) or last_rate
            query_log.append({"term": term, "query": query, "returned": 0, "error": str(error)})
            errors.append({"term": term, "status": error.status, "error": str(error)})
            continue
        except Exception as error:
            query_log.append({"term": term, "query": query, "returned": 0, "error": type(error).__name__})
            errors.append({"term": term, "error": type(error).__name__})
            continue

        for item in items:
            row = candidate(item, term)
            if not row or not row["commit_sha"]:
                continue
            if parse_date(row["timestamp"]) > now:
                filtered_future += 1
                continue
            key = (row["repo"].lower(), row["commit_sha"])
            if key in rows:
                merge(rows[key], row)
            else:
                rows[key] = row

    candidates = list(rows.values())
    candidates.sort(key=lambda row: (
        1 if not row["official_anthropic"] else 0,
        row["timestamp"],
        row["repo"].lower(),
        row["commit_sha"],
    ), reverse=True)
    for row in candidates:
        for field in ("_author_match", "_committer_match", "_coauthor_match"):
            del row[field]

    return {
        "generated_at": now.isoformat(),
        "source": "https://api.github.com/search/commits",
        "rules": {
            "lookback_days": LOOKBACK_DAYS,
            "max_results_per_query": MAX_RESULTS_PER_QUERY,
            "dedupe_key": "repo + commit_sha",
            "searches": SEARCHES,
            "official_definition": "repo full_name starts with anthropics/",
        },
        "api": {"rate_limit": last_rate, "queries": query_log},
        "filters": {"future_timestamps_excluded": filtered_future},
        "candidates": candidates,
        "errors": errors,
    }


def identity_label(identity):
    return identity.get("login") or identity.get("name") or identity.get("email") or "unknown"


def card(row):
    authors = f"author {identity_label(row['author'])} · committer {identity_label(row['committer'])}"
    if row["coauthors"]:
        authors += " · co-author " + ", ".join(x["name"] for x in row["coauthors"])
    return f"""<a class=\"commit-card\" href=\"{esc(row['commit_url'])}\" target=\"_blank\" rel=\"noreferrer\">
      <div class=\"topline\"><span class=\"repo\">{esc(row['repo'])}</span><span class=\"ecosystem {'official' if row['official_anthropic'] else 'independent'}\">{esc(row['ecosystem'])}</span></div>
      <h3>{esc(row['subject'])}</h3>
      <div class=\"meta\">{esc(row['category'])} · {esc(row['attribution_kind'])} · {esc(row['timestamp'])}</div>
      <p class=\"message\">{esc(row['commit_message'])}</p>
      <div class=\"why\">{esc(authors)}</div>
    </a>"""


def render(out_html, out_json):
    data = build()
    independent = [row for row in data["candidates"] if not row["official_anthropic"]]
    official = [row for row in data["candidates"] if row["official_anthropic"]]
    template = (ROOT / "claude-template.html").read_text()
    updated = data["generated_at"][:16].replace("T", " ")
    page = (template
        .replace("__UPDATED__", esc(updated))
        .replace("__COUNT__", str(len(data["candidates"])))
        .replace("__INDEPENDENT__", str(len(independent)))
        .replace("__OFFICIAL__", str(len(official)))
        .replace("__REPOS__", str(len({row["repo"] for row in data["candidates"]})))
        .replace("__QUERIES__", str(len(data["api"]["queries"])))
        .replace("__INDEPENDENT_CARDS__", "\n".join(card(row) for row in independent) or '<div class="empty">No independent Claude-attributed commits cleared the bounded search window.</div>')
        .replace("__OFFICIAL_CARDS__", "\n".join(card(row) for row in official) or '<div class="empty">No official Anthropic commits matched this pass.</div>')
    )
    out_html = pathlib.Path(out_html)
    out_html.parent.mkdir(parents=True, exist_ok=True)
    out_html.write_text(page)
    pathlib.Path(out_json).write_text(json.dumps(data, indent=2) + "\n")
    print(json.dumps({
        "candidates": len(data["candidates"]),
        "independent": len(independent),
        "official": len(official),
        "repos": len({row["repo"] for row in data["candidates"]}),
        "queries": len(data["api"]["queries"]),
        "errors": len(data["errors"]),
        "out": str(out_html),
    }))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--html-out", default=str(ROOT / "claude/index.html"))
    parser.add_argument("--json-out", default=str(ROOT / "claude.json"))
    args = parser.parse_args()
    render(args.html_out, args.json_out)