| """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()) |
|
|
