#!/usr/bin/env python3 """Apache GitHub issue/PR open-event scraper and reporter. python3 main.py scrape python3 main.py import-committers python3 main.py report Scrape stores author github id + login + created_at only. import-committers fills the committers table from ASF LDAP. report classifies each event at query time (bot / committer / unknown). """ from __future__ import annotations import argparse import csv import json import pathlib import sqlite3 import sys import time from datetime import datetime, timezone import ldap import requests import yaml from easydict import EasyDict as edict # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- THIS_DIR = pathlib.Path(__file__).resolve().parent CFG = edict(yaml.safe_load(open(THIS_DIR / "config.yaml"))) # --------------------------------------------------------------------------- # Hard-coded LDAP constants (not configurable) # --------------------------------------------------------------------------- LDAP_PEOPLE_BASE = "ou=people,dc=apache,dc=org" LDAP_COMMITTERS_DN = "cn=committers,ou=groups,dc=apache,dc=org" LDAP_ATTR_UID = "uid" LDAP_ATTR_MEMBER_UID = "memberUid" LDAP_ATTR_GH_NUMERIC = "asf-githubNumericID" LDAP_ATTR_GH_STRING = "asf-githubStringID" LDAP_ATTR_GH_USERNAME = "githubUsername" # --------------------------------------------------------------------------- # Paths / helpers # --------------------------------------------------------------------------- def _resolve(p: str) -> pathlib.Path: path = pathlib.Path(p) if not path.is_absolute(): path = THIS_DIR / path return path.resolve() def _require_token() -> str: token = (CFG.github.token or "").strip() if not token: raise SystemExit("github.token is empty in config.yaml") return token def _events_db() -> pathlib.Path: return _resolve(CFG.paths.events_db) def _report_dir() -> pathlib.Path: return _resolve(CFG.paths.report_dir) def _now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def is_bot(login: str | None) -> bool: if not login: return False if login in set(CFG.bots.logins or []): return True suffix = CFG.bots.suffix or "[bot]" return login.endswith(suffix) # --------------------------------------------------------------------------- # Database # --------------------------------------------------------------------------- SCHEMA = """ CREATE TABLE IF NOT EXISTS repos ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, is_archived INTEGER NOT NULL DEFAULT 0, is_fork INTEGER NOT NULL DEFAULT 0, updated_at_gh TEXT, scraped_at TEXT ); CREATE TABLE IF NOT EXISTS events ( repo_id INTEGER NOT NULL, number INTEGER NOT NULL, kind TEXT NOT NULL CHECK (kind IN ('issue', 'pr')), author_github_id INTEGER, author_login TEXT, created_at TEXT NOT NULL, PRIMARY KEY (repo_id, number), FOREIGN KEY (repo_id) REFERENCES repos(id) ); CREATE TABLE IF NOT EXISTS committers ( asf_id TEXT PRIMARY KEY, github_login TEXT, github_id INTEGER, imported_at TEXT ); CREATE TABLE IF NOT EXISTS scrape_state ( repo_id INTEGER PRIMARY KEY, phase TEXT NOT NULL DEFAULT 'issues', cursor TEXT, last_error TEXT, updated_at TEXT, FOREIGN KEY (repo_id) REFERENCES repos(id) ); CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT ); CREATE INDEX IF NOT EXISTS idx_events_created ON events(created_at); CREATE INDEX IF NOT EXISTS idx_events_author_id ON events(author_github_id); CREATE INDEX IF NOT EXISTS idx_events_author_login ON events(author_login); CREATE INDEX IF NOT EXISTS idx_committers_gh_id ON committers(github_id); CREATE INDEX IF NOT EXISTS idx_committers_gh_login ON committers(github_login); """ META_REPOS_LISTED_AT = "repos_listed_at" def connect_events(create: bool = True) -> sqlite3.Connection: path = _events_db() if create: path.parent.mkdir(parents=True, exist_ok=True) elif not path.is_file(): raise SystemExit(f"events DB not found: {path} (run scrape first)") conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") if create: conn.executescript(SCHEMA) conn.commit() return conn def meta_get(conn: sqlite3.Connection, key: str) -> str | None: row = conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() return row["value"] if row else None def meta_set(conn: sqlite3.Connection, key: str, value: str) -> None: conn.execute( """ INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value """, (key, value), ) def upsert_repo(conn: sqlite3.Connection, repo: dict) -> None: conn.execute( """ INSERT INTO repos (id, name, is_archived, is_fork, updated_at_gh) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET name = excluded.name, is_archived = excluded.is_archived, is_fork = excluded.is_fork, updated_at_gh = excluded.updated_at_gh """, ( repo["id"], repo["name"], 1 if repo["is_archived"] else 0, 1 if repo["is_fork"] else 0, repo.get("updated_at_gh"), ), ) def _repo_row_matches_filters(repo: dict) -> bool: """Apply scrape.only_repos / skip_forks / include_archived to a repo dict.""" only = set(CFG.scrape.only_repos or []) if only and repo["name"] not in only: return False if CFG.scrape.skip_forks and repo.get("is_fork"): return False if not CFG.scrape.include_archived and repo.get("is_archived"): return False return True def load_repos_from_db(conn: sqlite3.Connection) -> list[dict]: rows = conn.execute( """ SELECT id, name, is_archived, is_fork, updated_at_gh FROM repos ORDER BY name """ ).fetchall() out = [] for r in rows: repo = { "id": r["id"], "name": r["name"], "is_archived": bool(r["is_archived"]), "is_fork": bool(r["is_fork"]), "updated_at_gh": r["updated_at_gh"], } if _repo_row_matches_filters(repo): out.append(repo) return out def cached_repo_count(conn: sqlite3.Connection) -> int: row = conn.execute("SELECT COUNT(*) AS n FROM repos").fetchone() return int(row["n"]) def upsert_event( conn: sqlite3.Connection, repo_id: int, number: int, kind: str, author_github_id: int | None, author_login: str | None, created_at: str, ) -> None: conn.execute( """ INSERT INTO events (repo_id, number, kind, author_github_id, author_login, created_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(repo_id, number) DO UPDATE SET kind = excluded.kind, author_github_id = excluded.author_github_id, author_login = excluded.author_login, created_at = excluded.created_at """, (repo_id, number, kind, author_github_id, author_login, created_at), ) def set_scrape_state( conn: sqlite3.Connection, repo_id: int, phase: str, cursor: str | None = None, last_error: str | None = None, ) -> None: conn.execute( """ INSERT INTO scrape_state (repo_id, phase, cursor, last_error, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(repo_id) DO UPDATE SET phase = excluded.phase, cursor = excluded.cursor, last_error = excluded.last_error, updated_at = excluded.updated_at """, (repo_id, phase, cursor, last_error, _now_iso()), ) # --------------------------------------------------------------------------- # GraphQL + rate limits # --------------------------------------------------------------------------- class GraphQLError(Exception): def __init__( self, message: str, *, rate_limited: bool = False, retry_after: float | None = None, ): super().__init__(message) self.rate_limited = rate_limited self.retry_after = retry_after def _header_int(headers, name: str) -> int | None: raw = headers.get(name) if raw is None: return None try: return int(raw) except ValueError: return None def _rate_info_from_response(resp: requests.Response, body_rate: dict | None = None) -> dict: """Primary GraphQL rate limit from response headers (+ optional body rateLimit).""" info = { "limit": _header_int(resp.headers, "X-RateLimit-Limit"), "remaining": _header_int(resp.headers, "X-RateLimit-Remaining"), "used": _header_int(resp.headers, "X-RateLimit-Used"), "reset": _header_int(resp.headers, "X-RateLimit-Reset"), "resource": resp.headers.get("X-RateLimit-Resource"), "cost": None, "reset_at": None, } if body_rate: if info["remaining"] is None and body_rate.get("remaining") is not None: info["remaining"] = body_rate.get("remaining") info["cost"] = body_rate.get("cost") info["reset_at"] = body_rate.get("resetAt") return info def format_duration(secs: float) -> str: """Human duration: '45s', '37m', or '1.7h' (tri-partite thresholds).""" if secs < 0: secs = 0.0 if secs < 90: return f"{int(round(secs))}s" if secs < 90 * 60: return f"{int(round(secs / 60.0))}m" return f"{secs / 3600.0:.1f}h" def _reset_unix(info: dict) -> int | None: reset = info.get("reset") if reset is not None: return int(reset) reset_at = info.get("reset_at") if not reset_at: return None try: return int( datetime.fromisoformat(str(reset_at).replace("Z", "+00:00")).timestamp() ) except ValueError: return None REPO_NAME_WIDTH = 20 PHASE_WIDTH = 6 ITEM_WIDTH = 7 ETA_WIDTH = 5 # e.g. "1.7h", "23m", "45s" def format_api_quota(info: dict | None) -> str: """GitHub GraphQL point budget + reset: 'API 1806/5000 · reset in 23m (10:09 local)'.""" if not info: return "API ?/?" rem = info.get("remaining") limit = info.get("limit") cost = info.get("cost") if rem is not None and limit is not None: budget = f"API {rem}/{limit}" elif rem is not None: budget = f"API rem={rem}" else: budget = "API ?" # cost=1 is noise; only shout when unexpected if cost is not None and cost != 1: budget += f" COST={cost}" reset_ts = _reset_unix(info) if reset_ts is None: return budget delta = max(reset_ts - time.time(), 0.0) try: local = datetime.fromtimestamp(reset_ts).strftime("%H:%M") return f"{budget} · reset in {format_duration(delta)} ({local} local)" except (OSError, OverflowError, ValueError): return f"{budget} · reset in {format_duration(delta)}" def format_progress_line( repo_name: str, phase: str, items_done: int | None, total_count: int | None, avg_secs: float, eta_secs: float | None, rate_info: dict | None, ) -> str: """Aligned scrape progress line for human scanning.""" name = repo_name if len(repo_name) <= REPO_NAME_WIDTH else repo_name[: REPO_NAME_WIDTH - 1] + "…" name = f"{name:<{REPO_NAME_WIDTH}}" phase_s = f"{phase:<{PHASE_WIDTH}}" if items_done is not None and total_count is not None: items_s = f"{items_done:>{ITEM_WIDTH}}/{total_count:<{ITEM_WIDTH}}" elif items_done is not None: items_s = f"{items_done:>{ITEM_WIDTH}}/{'?':<{ITEM_WIDTH}}" else: items_s = f"{'?':>{ITEM_WIDTH}}/{'?':<{ITEM_WIDTH}}" # avg wall time per GraphQL request (up to PAGE_SIZE items) secs_s = f"{avg_secs:5.2f}s" if eta_secs is not None: eta_s = f"ETA {format_duration(eta_secs):<{ETA_WIDTH}}" else: eta_s = f"ETA {'?':<{ETA_WIDTH}}" return f" {name} {phase_s} {items_s} {secs_s} {eta_s} | {format_api_quota(rate_info)}" # Phrases that indicate primary or secondary rate limiting (HTTP or GraphQL errors). _RATE_LIMIT_MARKERS = ( "rate limit", "rate_limit", "secondary rate", "abuse detection", "abuse-detection", "you have exceeded", "please wait", "wait a few minutes", "api rate limit exceeded", ) # Cap consecutive rate-limit retries on a single page so we leave state and move on. MAX_RATE_LIMIT_RETRIES = 20 def _looks_like_rate_limit(text: str) -> bool: lower = text.lower() return any(m in lower for m in _RATE_LIMIT_MARKERS) def _sleep_for_rate_limit(resp: requests.Response | None = None) -> None: """Sleep based on Retry-After or X-RateLimit-Reset (or a default).""" info = _rate_info_from_response(resp) if resp is not None else {} if resp is not None: retry_after = resp.headers.get("Retry-After") if retry_after is not None: try: secs = float(retry_after) print( f" rate limit sleep: Retry-After={secs}s ({format_api_quota(info)})", flush=True, ) time.sleep(max(secs, 1.0)) return except ValueError: pass reset = info.get("reset") if reset is not None: now = time.time() secs = max(reset - now, 1.0) + 1.0 print( f" rate limit sleep: ~{secs:.0f}s until reset ({format_api_quota(info)})", flush=True, ) time.sleep(secs) return print( f" rate limit sleep: 60s ({format_api_quota(info) if info else 'no headers'})", flush=True, ) time.sleep(60.0) def graphql(token: str, query: str, variables: dict | None = None) -> tuple[dict, dict]: """POST GraphQL. Returns (data, rate_info) from headers + body rateLimit.""" headers = { "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", } payload = {"query": query, "variables": variables or {}} resp = requests.post( CFG.github.api_url, headers=headers, data=json.dumps(payload), timeout=120, ) remaining = _header_int(resp.headers, "X-RateLimit-Remaining") if remaining is not None and remaining <= 1 and resp.status_code == 200: # Nearly exhausted; pause before next call _sleep_for_rate_limit(resp) if resp.status_code == 429 or ( resp.status_code == 403 and ( _looks_like_rate_limit(resp.text) or resp.headers.get("X-RateLimit-Remaining") == "0" ) ): _sleep_for_rate_limit(resp) raise GraphQLError( f"HTTP {resp.status_code}: rate limited", rate_limited=True, ) if resp.status_code != 200: raise GraphQLError(f"HTTP {resp.status_code}: {resp.text[:500]}") body = resp.json() if "errors" in body: msg = json.dumps(body["errors"])[:800] rate = _looks_like_rate_limit(msg) if rate: _sleep_for_rate_limit(resp) raise GraphQLError(msg, rate_limited=rate) if "data" not in body: raise GraphQLError(f"No data in response: {str(body)[:500]}") data = body["data"] rate_info = _rate_info_from_response(resp, data.get("rateLimit") if data else None) return data, rate_info def _page_sleep() -> None: secs = float(CFG.github.page_sleep or 0) if secs > 0: time.sleep(secs) REPOS_QUERY = """ query($org: String!, $cursor: String) { organization(login: $org) { repositories(first: 100, after: $cursor, orderBy: {field: NAME, direction: ASC}) { pageInfo { hasNextPage endCursor } nodes { databaseId name isArchived isFork isPrivate updatedAt } } } rateLimit { remaining resetAt cost } } """ PAGE_SIZE = 100 ISSUES_QUERY = """ query($owner: String!, $name: String!, $cursor: String) { repository(owner: $owner, name: $name) { issues(first: 100, after: $cursor, orderBy: {field: CREATED_AT, direction: ASC}, filterBy: {states: [OPEN, CLOSED]}) { totalCount pageInfo { hasNextPage endCursor } nodes { number createdAt author { login # databaseId exists on User/Bot/Mannequin, not EnterpriseUserAccount ... on User { databaseId } ... on Bot { databaseId } ... on Mannequin { databaseId } } } } } rateLimit { remaining resetAt cost } } """ PRS_QUERY = """ query($owner: String!, $name: String!, $cursor: String) { repository(owner: $owner, name: $name) { pullRequests(first: 100, after: $cursor, orderBy: {field: CREATED_AT, direction: ASC}, states: [OPEN, CLOSED, MERGED]) { totalCount pageInfo { hasNextPage endCursor } nodes { number createdAt author { login ... on User { databaseId } ... on Bot { databaseId } ... on Mannequin { databaseId } } } } } rateLimit { remaining resetAt cost } } """ def _author_fields(author: dict | None) -> tuple[int | None, str | None]: if not author: return None, None return author.get("databaseId"), author.get("login") def _maybe_sleep_graphql_rate(rate_info: dict) -> None: """If point budget is nearly exhausted, sleep until window reset.""" remaining = rate_info.get("remaining") if remaining is None or remaining >= 50: return print(f" low API budget: {format_api_quota(rate_info)}", flush=True) if remaining >= 5: return reset = rate_info.get("reset") if reset is not None: secs = max(reset - time.time(), 1.0) + 2.0 print(f" sleeping {secs:.0f}s for GraphQL point budget", flush=True) time.sleep(secs) return reset_at = rate_info.get("reset_at") if reset_at: try: reset_dt = datetime.fromisoformat(str(reset_at).replace("Z", "+00:00")) secs = (reset_dt - datetime.now(timezone.utc)).total_seconds() + 2 if secs > 0: print(f" sleeping {secs:.0f}s for GraphQL point budget", flush=True) time.sleep(secs) return except ValueError: pass time.sleep(30) def fetch_all_repos(token: str) -> tuple[list[dict], int]: """GraphQL-list org repos. Returns (repos, page_count).""" org = CFG.github.org cursor = None out: list[dict] = [] pages = 0 rate_limit_retries = 0 while True: try: data, rate_info = graphql( token, REPOS_QUERY, {"org": org, "cursor": cursor} ) rate_limit_retries = 0 except GraphQLError as e: if e.rate_limited: rate_limit_retries += 1 if rate_limit_retries > MAX_RATE_LIMIT_RETRIES: raise GraphQLError( f"gave up listing repos after {MAX_RATE_LIMIT_RETRIES} " f"rate-limit retries: {e}", rate_limited=False, ) from e continue raise _maybe_sleep_graphql_rate(rate_info) pages += 1 if pages == 1 or pages % 10 == 0: # Align with scrape progress: fixed name/phase columns, then API block name = f"{'list':<{REPO_NAME_WIDTH}}" phase_s = f"{'repos':<{PHASE_WIDTH}}" print( f" {name} {phase_s} page {pages:<5} | {format_api_quota(rate_info)}", flush=True, ) org_data = data.get("organization") if not org_data: raise SystemExit(f"organization {org!r} not found or not accessible") repos = org_data["repositories"] for node in repos["nodes"]: if node is None: continue if node.get("isPrivate"): continue repo = { "id": node["databaseId"], "name": node["name"], "is_archived": bool(node.get("isArchived")), "is_fork": bool(node.get("isFork")), "updated_at_gh": node.get("updatedAt"), } # Store everything non-private; filters applied when building work list out.append(repo) page = repos["pageInfo"] if not page["hasNextPage"]: break cursor = page["endCursor"] _page_sleep() return out, pages def refresh_repos_from_github( conn: sqlite3.Connection, token: str ) -> list[dict]: """Query GraphQL for the org repo list, time it, upsert into SQLite.""" org = CFG.github.org print(f"listing repositories for org={org!r} via GraphQL …", flush=True) t0 = time.perf_counter() all_repos, pages = fetch_all_repos(token) elapsed = time.perf_counter() - t0 for repo in all_repos: upsert_repo(conn, repo) listed_at = _now_iso() meta_set(conn, META_REPOS_LISTED_AT, listed_at) conn.commit() work = [r for r in all_repos if _repo_row_matches_filters(r)] print( f" listed {len(all_repos)} repos in {elapsed:.1f}s ({pages} pages); " f"{len(work)} match scrape filters; cached listed_at={listed_at}", flush=True, ) return work def _scrape_connection( conn: sqlite3.Connection, token: str, repo: dict, phase: str, query: str, conn_key: str, kind: str, start_cursor: str | None, ) -> None: org = CFG.github.org cursor = start_cursor # Items already stored for this repo/kind (resume-aware progress) items_done = conn.execute( "SELECT COUNT(*) AS n FROM events WHERE repo_id = ? AND kind = ?", (repo["id"], kind), ).fetchone()["n"] total_count: int | None = None page_times: list[float] = [] pages_this_call = 0 rate_limit_retries = 0 while True: t0 = time.perf_counter() try: data, rate_info = graphql( token, query, {"owner": org, "name": repo["name"], "cursor": cursor}, ) rate_limit_retries = 0 except GraphQLError as e: if e.rate_limited: rate_limit_retries += 1 if rate_limit_retries > MAX_RATE_LIMIT_RETRIES: msg = ( f"gave up after {MAX_RATE_LIMIT_RETRIES} rate-limit retries " f"on {phase}: {e}" ) set_scrape_state(conn, repo["id"], phase, cursor, msg) conn.commit() raise GraphQLError(msg, rate_limited=False) from e continue # Keep phase as issues|prs (never "error") so resume does not # skip work and then mark the repo done. set_scrape_state(conn, repo["id"], phase, cursor, str(e)) conn.commit() raise page_elapsed = time.perf_counter() - t0 page_times.append(page_elapsed) _maybe_sleep_graphql_rate(rate_info) repo_data = data.get("repository") if repo_data is None: msg = "repository null (deleted, renamed, or inaccessible)" set_scrape_state(conn, repo["id"], phase, cursor, msg) conn.commit() # Raise so cmd_scrape does not advance phase or mark done. raise GraphQLError(msg) connection = repo_data[conn_key] if total_count is None and connection.get("totalCount") is not None: total_count = int(connection["totalCount"]) n_nodes = 0 for node in connection["nodes"]: if node is None: continue n_nodes += 1 gh_id, login = _author_fields(node.get("author")) upsert_event( conn, repo["id"], node["number"], kind, gh_id, login, node["createdAt"], ) items_done += n_nodes pages_this_call += 1 page = connection["pageInfo"] cursor = page["endCursor"] if page["hasNextPage"] else None set_scrape_state(conn, repo["id"], phase, cursor, None) conn.commit() # First request (totals kickoff), every 10th, and last (final ETA 0) log_progress = ( pages_this_call == 1 or pages_this_call % 10 == 0 or not page["hasNextPage"] ) if log_progress: avg = sum(page_times) / len(page_times) if total_count is not None: remaining_items = max(total_count - items_done, 0) remaining_reqs = ( (remaining_items + PAGE_SIZE - 1) // PAGE_SIZE if remaining_items else 0 ) eta_secs = avg * remaining_reqs else: eta_secs = None print( format_progress_line( repo["name"], phase, items_done, total_count, avg, eta_secs, rate_info, ), flush=True, ) if not page["hasNextPage"]: break _page_sleep() def cmd_scrape(args: argparse.Namespace) -> None: token = _require_token() conn = connect_events(create=True) only = list(CFG.scrape.only_repos or []) max_to_try = int(getattr(CFG.scrape, "max_to_try", 0) or 0) print( f"scrape config: org={CFG.github.org!r} " f"only_repos={only or '(all)'} max_to_try={max_to_try or '(unlimited)'} " f"skip_forks={bool(CFG.scrape.skip_forks)} " f"include_archived={bool(CFG.scrape.include_archived)} " f"refresh_repos={bool(args.refresh_repos)}", flush=True, ) n_cached = cached_repo_count(conn) if args.refresh_repos or n_cached == 0: if n_cached == 0 and not args.refresh_repos: print("repository cache empty; querying GraphQL …", flush=True) repos = refresh_repos_from_github(conn, token) else: listed_at = meta_get(conn, META_REPOS_LISTED_AT) or "unknown" repos = load_repos_from_db(conn) print( f"using cached repository list ({n_cached} in db, {len(repos)} after " f"filters; last listed {listed_at}; pass --refresh-repos to re-query)", flush=True, ) if not repos: raise SystemExit("no repositories to scrape (check filters / refresh list)") already_done = 0 tried_this_run = 0 ok_this_run = 0 for repo in repos: state = conn.execute( "SELECT phase, cursor, last_error FROM scrape_state WHERE repo_id = ?", (repo["id"],), ).fetchone() if state and state["phase"] == "done": already_done += 1 continue if max_to_try and tried_this_run >= max_to_try: print( f"max_to_try={max_to_try} reached; stopping " f"({ok_this_run} ok, {tried_this_run - ok_this_run} failed)", flush=True, ) break # phases: issues | prs | done. Legacy "error" (or anything else) → issues. if state: phase = state["phase"] cursor = state["cursor"] last_error = state["last_error"] else: phase = "issues" cursor = None last_error = None if phase not in ("issues", "prs"): print( f" ! {repo['name']}: unknown phase {phase!r} " f"(treating as issues; last_error={last_error!r})", flush=True, ) phase = "issues" # Keep cursor if present — may still be valid for issues mid-stream. # If it was a PR-phase error stored as "error", worst case we re-walk issues (idempotent). tried_this_run += 1 retry_note = "" if last_error: snippet = last_error if len(last_error) <= 80 else last_error[:80] + "…" retry_note = f" [retry after: {snippet}]" print( f"scraping {repo['name']} (phase={phase}; " f"try {tried_this_run}" + (f"/{max_to_try}" if max_to_try else "") + f"){retry_note} …", flush=True, ) try: if phase == "issues": _scrape_connection( conn, token, repo, "issues", ISSUES_QUERY, "issues", "issue", cursor, ) phase = "prs" cursor = None set_scrape_state(conn, repo["id"], phase, cursor, None) conn.commit() if phase == "prs": _scrape_connection( conn, token, repo, "prs", PRS_QUERY, "pullRequests", "pr", cursor, ) set_scrape_state(conn, repo["id"], "done", None, None) conn.execute( "UPDATE repos SET scraped_at = ? WHERE id = ?", (_now_iso(), repo["id"]), ) conn.commit() ok_this_run += 1 print(f" done {repo['name']}", flush=True) except GraphQLError as e: # scrape_state already has phase=issues|prs + cursor + last_error print(f" error on {repo['name']}: {e}", flush=True) continue n_events = conn.execute("SELECT COUNT(*) FROM events").fetchone()[0] n_done_total = conn.execute( "SELECT COUNT(*) FROM scrape_state WHERE phase = 'done'" ).fetchone()[0] # Work-list stats (after only_repos / filters) vs whole-DB done count print( f"scrape finished: events={n_events}; " f"work_list={len(repos)} already_done={already_done} " f"this_run tried={tried_this_run} ok={ok_this_run}; " f"db repos_phase_done={n_done_total}", flush=True, ) conn.close() # --------------------------------------------------------------------------- # LDAP import-committers # --------------------------------------------------------------------------- def _ldap_first(attrs: dict, name: str) -> str | None: vals = attrs.get(name) if not vals: return None raw = vals[0] if isinstance(raw, bytes): raw = raw.decode("utf-8", errors="replace") s = str(raw).strip() return s or None def _ldap_filter_escape(value: str) -> str: """Escape a value for use inside an LDAP filter assertion.""" out = [] for ch in value: if ch in r"\*()": out.append(f"\\{ord(ch):02x}") elif ord(ch) < 32: out.append(f"\\{ord(ch):02x}") else: out.append(ch) return "".join(out) def _require_ldap_config() -> tuple[str, str, str]: uri = (CFG.ldap.uri or "").strip() bind_dn = (CFG.ldap.bind_dn or "").strip() bind_password = CFG.ldap.bind_password or "" if not uri: raise SystemExit("ldap.uri is empty in config.yaml") if not bind_dn: raise SystemExit("ldap.bind_dn is empty in config.yaml (anonymous not allowed)") if not bind_password: raise SystemExit("ldap.bind_password is empty in config.yaml") return uri, bind_dn, bind_password def cmd_import_committers(_args: argparse.Namespace) -> None: uri, bind_dn, bind_password = _require_ldap_config() conn = connect_events(create=True) print(f"LDAP bind to {uri} as {bind_dn} …", flush=True) try: lc = ldap.initialize(uri) lc.set_option(ldap.OPT_REFERRALS, 0) lc.simple_bind_s(bind_dn, bind_password) except ldap.LDAPError as e: raise SystemExit(f"LDAP bind failed: {e}") from e try: # Committer uids from the committers group results = lc.search_s( LDAP_COMMITTERS_DN, ldap.SCOPE_BASE, "(objectClass=*)", [LDAP_ATTR_MEMBER_UID], ) if not results: raise SystemExit(f"no LDAP entry for {LDAP_COMMITTERS_DN}") _dn, attrs = results[0] member_uids = attrs.get(LDAP_ATTR_MEMBER_UID) or [] uids = [] for u in member_uids: if isinstance(u, bytes): u = u.decode("utf-8", errors="replace") u = str(u).strip() if u: uids.append(u) if not uids: raise SystemExit("committers group has zero memberUid values") print(f" {len(uids)} committer uids; fetching GitHub attrs …", flush=True) # asf_id -> (github_id, github_login); start with all uids present by_uid: dict[str, tuple[int | None, str | None]] = { uid: (None, None) for uid in uids } batch_size = 50 want = [ LDAP_ATTR_UID, LDAP_ATTR_GH_NUMERIC, LDAP_ATTR_GH_STRING, LDAP_ATTR_GH_USERNAME, ] for i in range(0, len(uids), batch_size): batch = uids[i : i + batch_size] filt = "(|" + "".join( f"({LDAP_ATTR_UID}={_ldap_filter_escape(uid)})" for uid in batch ) + ")" try: found = lc.search_s( LDAP_PEOPLE_BASE, ldap.SCOPE_ONELEVEL, filt, want, ) except ldap.LDAPError as e: print(f" ! LDAP people batch {i // batch_size}: {e}", flush=True) continue for _pdn, pattrs in found: if not isinstance(pattrs, dict): continue asf_id = _ldap_first(pattrs, LDAP_ATTR_UID) if not asf_id: continue numeric_s = _ldap_first(pattrs, LDAP_ATTR_GH_NUMERIC) gh_id = None if numeric_s is not None: try: gh_id = int(numeric_s) except ValueError: gh_id = None # string ID preferred over githubUsername login = _ldap_first(pattrs, LDAP_ATTR_GH_STRING) or _ldap_first( pattrs, LDAP_ATTR_GH_USERNAME ) by_uid[asf_id] = (gh_id, login) finally: try: lc.unbind_s() except ldap.LDAPError: pass imported_at = _now_iso() conn.execute("DELETE FROM committers") with_id = 0 with_login = 0 for asf_id, (gh_id, login) in sorted(by_uid.items()): if gh_id is not None: with_id += 1 if login: with_login += 1 conn.execute( """ INSERT INTO committers (asf_id, github_login, github_id, imported_at) VALUES (?, ?, ?, ?) """, (asf_id, login, gh_id, imported_at), ) no_github = sum( 1 for gh_id, login in by_uid.values() if gh_id is None and not login ) conn.commit() print( f"import-committers: rows={len(by_uid)} with_github_id={with_id} " f"with_github_login={with_login} with_neither={no_github}", flush=True, ) conn.close() # --------------------------------------------------------------------------- # Report (classify at query time) # --------------------------------------------------------------------------- def classify_row( author_login: str | None, author_github_id: int | None, by_id: dict[int, str], by_login: dict[str, str], ) -> str: if is_bot(author_login): return "bot" if author_github_id is not None and author_github_id in by_id: return "committer" if author_login and author_login.lower() in by_login: return "committer" return "unknown" def cmd_report(_args: argparse.Namespace) -> None: """Monthly aggregates only (period = YYYY-MM).""" conn = connect_events(create=False) period_sql = "substr(e.created_at, 1, 7)" # Load committer maps once by_id: dict[int, str] = {} by_login: dict[str, str] = {} for row in conn.execute( "SELECT asf_id, github_id, github_login FROM committers" ): if row["github_id"] is not None: by_id[int(row["github_id"])] = row["asf_id"] if row["github_login"]: by_login[row["github_login"].lower()] = row["asf_id"] if not by_id and not by_login: print( "warning: committers table empty — run import-committers " "(all non-bots will be 'unknown')", flush=True, ) # Aggregate in Python: month × kind × group counts: dict[tuple[str, str, str], int] = {} for row in conn.execute( f""" SELECT {period_sql} AS period, e.kind AS kind, e.author_github_id AS author_github_id, e.author_login AS author_login FROM events e """ ): group = classify_row( row["author_login"], row["author_github_id"], by_id, by_login ) key = (row["period"], row["kind"], group) counts[key] = counts.get(key, 0) + 1 conn.close() out_dir = _report_dir() out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / "activity_by_month.csv" with open(out_path, "w", newline="") as f: w = csv.writer(f) w.writerow(["period", "kind", "group_name", "count"]) for (period, kind, group), n in sorted(counts.items()): w.writerow([period, kind, group, n]) print(f"wrote {len(counts)} rows → {out_path}", flush=True) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> None: parser = argparse.ArgumentParser( description="Apache GitHub issue/PR open-event scraper and reporter", ) sub = parser.add_subparsers(dest="command", required=True) p_scrape = sub.add_parser("scrape", help="Fetch issue/PR opens into events SQLite") p_scrape.add_argument( "--refresh-repos", action="store_true", help="Re-query the org repository list from GraphQL instead of using the SQLite cache", ) p_scrape.set_defaults(func=cmd_scrape) p_imp = sub.add_parser( "import-committers", help="Load ASF committers + GitHub ids from LDAP into committers table", ) p_imp.set_defaults(func=cmd_import_committers) p_rep = sub.add_parser( "report", help="Classify events and write CSV aggregates by period/kind/group", ) p_rep.set_defaults(func=cmd_report) args = parser.parse_args(argv) args.func(args) if __name__ == "__main__": main()