"""Ingest pulled GH JSONL into SQLite. Derives `truth_action` per issue from labels/state/assignee — heuristic ground truth. Usage: python scripts/ingest_to_sqlite.py --raw data/raw/huggingface_peft --db data/repo.db --repo huggingface/peft """ from __future__ import annotations import argparse import json import sqlite3 import sys from datetime import datetime, timezone from pathlib import Path SCHEMA = Path(__file__).resolve().parent.parent / "world" / "schema.sql" def parse_iso(s: str | None) -> datetime | None: if not s: return None return datetime.fromisoformat(s.replace("Z", "+00:00")) def derive_truth_action(issue: dict) -> tuple[str, list[str], str | None, str | None]: """Return (action, labels, close_reason, assignee_login).""" labels = [lab["name"] for lab in (issue.get("labels") or [])] label_set = {l.lower() for l in labels} state = issue.get("state", "open") is_pr = issue.get("pull_request") is not None assignees = issue.get("assignees") or [] assignee = assignees[0]["login"] if assignees else None close_reason = None if state == "closed": close_reason = (issue.get("state_reason") or "completed").lower() if is_pr and state == "closed" and (issue.get("pull_request") or {}).get("merged_at"): action = "merge_pr" elif "duplicate" in label_set: action = "link_duplicate" elif label_set & {"needs-info", "needs-information", "needs-reproduction", "more-info", "question"}: action = "request_info" elif label_set & {"spam", "invalid", "not-a-bug", "wontfix"}: action = "close_spam" elif assignee: action = "assign" elif labels: action = "label" else: action = "comment" return action, labels, close_reason, assignee def tier_for(issue: dict) -> int: """Difficulty tier 1-5 from text length, label count, comment count.""" body_len = len(issue.get("body") or "") title_len = len(issue.get("title") or "") n_labels = len(issue.get("labels") or []) n_comments = issue.get("comments", 0) score = (body_len > 200) + (title_len > 60) + (n_labels >= 2) + (n_comments >= 3) + (n_comments >= 8) return max(1, min(5, score + 1)) def upsert_contributor(cur: sqlite3.Cursor, user: dict | None, now_dt: datetime): if not user or not user.get("login"): return login = user["login"] created = parse_iso(user.get("created_at")) age = (now_dt - created).days if created else 0 cur.execute( "INSERT OR IGNORE INTO contributors (login, account_age_days, hidden_reputation) VALUES (?, ?, ?)", (login, age, 0.5), ) def main(): ap = argparse.ArgumentParser() ap.add_argument("--raw", required=True) ap.add_argument("--db", default="data/repo.db") ap.add_argument("--repo", required=True) args = ap.parse_args() raw = Path(args.raw) db_path = Path(args.db) db_path.parent.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(db_path) cur = con.cursor() cur.executescript(SCHEMA.read_text(encoding="utf-8")) labels_path = raw / "labels.jsonl" if labels_path.exists(): for line in labels_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue lab = json.loads(line) cur.execute( "INSERT OR REPLACE INTO labels (repo, name, description) VALUES (?, ?, ?)", (args.repo, lab["name"], lab.get("description")), ) issues_path = raw / "issues.jsonl" now_dt = datetime.now(timezone.utc) n_issues = 0 for line in issues_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue issue = json.loads(line) author = issue.get("user") upsert_contributor(cur, author, now_dt) action, labels, close_reason, assignee = derive_truth_action(issue) if assignee: upsert_contributor(cur, {"login": assignee}, now_dt) cur.execute( """INSERT OR REPLACE INTO issues (issue_id, number, repo, title, body, author_login, created_at, closed_at, state, is_pr, tier, is_synthetic_spam, spam_pattern, truth_action, truth_labels, truth_close_reason, truth_assignee) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( issue["id"], issue["number"], args.repo, issue.get("title", "")[:500], issue.get("body") or "", (author or {}).get("login", "ghost"), issue.get("created_at"), issue.get("closed_at"), issue.get("state", "open"), 1 if issue.get("pull_request") else 0, tier_for(issue), 0, None, action, json.dumps(labels), close_reason, assignee, ), ) n_issues += 1 comments_path = raw / "comments.jsonl" n_comments = 0 if comments_path.exists(): for line in comments_path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue c = json.loads(line) issue_num = c.get("_issue_number") row = cur.execute( "SELECT issue_id FROM issues WHERE repo=? AND number=?", (args.repo, issue_num), ).fetchone() if not row: continue user = c.get("user") or {} upsert_contributor(cur, user, now_dt) cur.execute( """INSERT OR REPLACE INTO comments (comment_id, issue_id, author_login, body, created_at) VALUES (?, ?, ?, ?, ?)""", ( c["id"], row[0], user.get("login", "ghost"), (c.get("body") or "")[:4000], c.get("created_at"), ), ) n_comments += 1 cur.execute( """UPDATE contributors SET public_pr_count = (SELECT COUNT(*) FROM issues WHERE author_login=contributors.login AND is_pr=1), public_issue_count = (SELECT COUNT(*) FROM issues WHERE author_login=contributors.login AND is_pr=0)""" ) cur.execute( """UPDATE contributors SET hidden_reputation = MIN(1.0, 0.3 + 0.4 * MIN(1.0, public_pr_count / 10.0) + 0.2 * MIN(1.0, account_age_days / 730.0) + 0.1 * MIN(1.0, public_issue_count / 20.0))""" ) con.commit() print(f"DONE: ingested {n_issues} issues, {n_comments} comments into {db_path}", flush=True) counts = cur.execute("SELECT tier, COUNT(*) FROM issues GROUP BY tier ORDER BY tier").fetchall() print(f" tier distribution: {counts}", flush=True) actions = cur.execute( "SELECT truth_action, COUNT(*) FROM issues GROUP BY truth_action ORDER BY 2 DESC" ).fetchall() print(f" action distribution: {actions}", flush=True) con.close() if __name__ == "__main__": sys.exit(main())