Refactor: Deduplicate shared code across modules

src/analyze.py

@@ -6,7 +6,6 @@
     GEMINI_API_KEY, GEMINI_API_URL,
     LLM_SYSTEM_PROMPT, LLM_USER_PROMPT_TEMPLATE,
     TAXONOMY_PATH, RETRY_ATTEMPTS, RETRY_BACKOFF,
-    RATE_LIMIT_DELAY,
 )
 
 
@@ -42,7 +41,7 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]:
 
     raw_response = _call_gemini(user_prompt)
     if not raw_response:
-        print(f"    DEBUG: _call_gemini returned None")
+        print("    DEBUG: _call_gemini returned None")
         return None
 
     parsed = _parse_llm_response(raw_response)
@@ -53,6 +52,16 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]:
     if parsed.get("pattern_id") not in taxonomy_ids:
         parsed["pattern_id"] = "UNCLASSIFIED"
 
+    return build_analysis_result(advisory, parsed)
+
+
+def build_analysis_result(advisory: dict, parsed: dict) -> dict:
+    """Build a standardized analysis result from advisory metadata and LLM output."""
+    def _str(val):
+        if isinstance(val, (dict, list)):
+            return json.dumps(val)
+        return str(val) if val else ""
+
     return {
         "ghsa_id": advisory["ghsa_id"],
         "date": advisory["published_at"][:10],
@@ -63,12 +72,12 @@ def analyze_advisory(advisory: dict, filtered_diff: str) -> Optional[dict]:
         "cvss_score": advisory["cvss_score"],
         "package_name": advisory["package_name"],
         "pattern_id": parsed["pattern_id"],
-        "vuln_type": parsed.get("vuln_type", ""),
-        "root_cause": parsed.get("root_cause", ""),
-        "impact": parsed.get("impact", ""),
-        "fix_summary": parsed.get("fix_summary", ""),
-        "key_diff": parsed.get("key_diff", ""),
-        "confidence": parsed.get("confidence", "LOW"),
+        "vuln_type": _str(parsed.get("vuln_type", "")),
+        "root_cause": _str(parsed.get("root_cause", "")),
+        "impact": _str(parsed.get("impact", "")),
+        "fix_summary": _str(parsed.get("fix_summary", "")),
+        "key_diff": _str(parsed.get("key_diff", "")),
+        "confidence": _str(parsed.get("confidence", "LOW")),
         "commit_url": advisory["commit_url"],
     }
 
@@ -107,13 +116,13 @@ def _call_gemini(user_prompt: str) -> Optional[str]:
             data = response.json()
             candidates = data.get("candidates", [])
             if not candidates:
-                print(f"    Gemini: no candidates in response")
+                print("    Gemini: no candidates in response")
                 return None
 
             content = candidates[0].get("content", {})
             parts = content.get("parts", [])
             if not parts:
-                print(f"    Gemini: no parts in response")
+                print("    Gemini: no parts in response")
                 return None
 
             return parts[0].get("text", "")

src/backfill_local.py

@@ -1,16 +1,12 @@
-import json
 import sys
-import time
 import requests
 from datetime import datetime, timedelta, timezone
 from src.config import (
-    RATE_LIMIT_DELAY, MAX_DIFF_LINES,
     LLM_SYSTEM_PROMPT, LLM_USER_PROMPT_TEMPLATE,
-    TAXONOMY_PATH, STATE_PATH,
 )
 from src.fetch import fetch_advisories, fetch_commit_diff
 from src.diff_filter import filter_diff
-from src.analyze import load_taxonomy, _parse_llm_response, _ecosystem_to_language
+from src.analyze import load_taxonomy, _parse_llm_response, build_analysis_result
 from src.db import (
     rebuild_from_jsonl, advisory_exists,
     insert_analysis, export_to_jsonl, get_stats,
@@ -64,7 +60,7 @@ def analyze_with_ollama(advisory: dict, filtered_diff: str) -> dict | None:
 
     raw_response = call_ollama(user_prompt)
     if not raw_response:
-        print(f"    Ollama returned nothing")
+        print("    Ollama returned nothing")
         return None
 
     parsed = _parse_llm_response(raw_response)
@@ -75,31 +71,7 @@ def analyze_with_ollama(advisory: dict, filtered_diff: str) -> dict | None:
     if parsed.get("pattern_id") not in taxonomy_ids:
         parsed["pattern_id"] = "UNCLASSIFIED"
 
-    def _str(val):
-        if isinstance(val, dict):
-            return json.dumps(val)
-        if isinstance(val, list):
-            return json.dumps(val)
-        return str(val) if val else ""
-
-    return {
-        "ghsa_id": advisory["ghsa_id"],
-        "date": advisory["published_at"][:10],
-        "cve_id": "",
-        "repo": advisory["repo"],
-        "language": _ecosystem_to_language(advisory["ecosystem"]),
-        "severity": advisory["severity"],
-        "cvss_score": advisory["cvss_score"],
-        "package_name": advisory["package_name"],
-        "pattern_id": parsed["pattern_id"],
-        "vuln_type": _str(parsed.get("vuln_type", "")),
-        "root_cause": _str(parsed.get("root_cause", "")),
-        "impact": _str(parsed.get("impact", "")),
-        "fix_summary": _str(parsed.get("fix_summary", "")),
-        "key_diff": _str(parsed.get("key_diff", "")),
-        "confidence": _str(parsed.get("confidence", "LOW")),
-        "commit_url": advisory["commit_url"],
-    }
+    return build_analysis_result(advisory, parsed)
 
 
 def backfill_local(days: int):
@@ -154,13 +126,13 @@ def backfill_local(days: int):
 
         raw_diff = fetch_commit_diff(advisory["commit_url"])
         if not raw_diff:
-            print(f"    SKIP: no diff")
+            print("    SKIP: no diff")
             errors += 1
             continue
 
         filtered = filter_diff(raw_diff)
         if not filtered:
-            print(f"    SKIP: no relevant files")
+            print("    SKIP: no relevant files")
             continue
 
         result = analyze_with_ollama(advisory, filtered)
@@ -195,12 +167,12 @@ def backfill_local(days: int):
     render_html_index()
 
     stats = get_stats()
-    print(f"\n=== Summary ===")
+    print("\n=== Summary ===")
     print(f"Processed: {processed}")
     print(f"New patterns: {new_patterns}")
     print(f"Errors: {errors}")
     print(f"Total DB: {stats['total_advisories']} advisories, {stats['total_patterns']} patterns")
-    print(f"\nCommit and push:")
+    print("\nCommit and push:")
     print(f"  git add -A && git commit -m 'feat: local backfill — {processed} advisories' && git push")
 
 

src/db.py

@@ -1,6 +1,5 @@
 import json
 import sqlite3
-from pathlib import Path
 from typing import Optional
 from src.config import DB_PATH, JSONL_PATH
 

src/deep_scan.py

@@ -1,58 +1,17 @@
 import json
 import sys
 import time
-import os
 import requests
 from datetime import datetime, timezone
-from pathlib import Path
-from src.config import DATA_DIR
+from src.config import DATA_DIR, GITHUB_TOKEN
 from src.heuristics import score_commit
-from src.fingerprint import match_fingerprints
+from src.fingerprint import score_with_fingerprints
+from src.github_api import github_get, get_link_next, GITHUB_API, REQUEST_DELAY
 
 
-GITHUB_API = "https://api.github.com"
-GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
-REQUEST_DELAY = 0.8
 RESULTS_DIR = DATA_DIR / "deep_scans"
 
 
-def github_get(url: str, params: dict = None) -> dict | list | None:
-    headers = {"Accept": "application/vnd.github.v3+json"}
-    if GITHUB_TOKEN:
-        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
-
-    full_url = f"{GITHUB_API}{url}" if url.startswith("/") else url
-
-    try:
-        response = requests.get(full_url, headers=headers, params=params, timeout=30)
-        if response.status_code == 403:
-            remaining = response.headers.get("X-RateLimit-Remaining", "?")
-            reset = response.headers.get("X-RateLimit-Reset", "?")
-            print(f"  Rate limited (remaining: {remaining}, reset: {reset})")
-            if remaining == "0":
-                wait = max(int(reset) - int(time.time()), 10)
-                print(f"  Waiting {wait}s for rate limit reset...")
-                time.sleep(wait)
-                return github_get(url, params)
-            return None
-        if response.status_code == 404:
-            return None
-        response.raise_for_status()
-        return response.json()
-    except requests.RequestException as exc:
-        print(f"  API error: {exc}")
-        return None
-
-
-def get_link_next(response_headers: dict) -> str | None:
-    link = response_headers.get("Link", "")
-    for part in link.split(","):
-        if 'rel="next"' in part:
-            url = part.split(";")[0].strip().strip("<>")
-            return url
-    return None
-
-
 def fetch_all_commits(repo: str, since: str = None, until: str = None, per_page: int = 100) -> list[dict]:
     all_commits = []
     params = {"per_page": per_page}
@@ -146,7 +105,7 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int
         message = commit.get("commit", {}).get("message", "").split("\n")[0]
 
         if (i + 1) % 50 == 0 or (i + 1) == len(new_commits):
-            print(f"  [{i+1}/{len(new_commits)}] {suspects_count} suspects so far...", flush=True) if 'suspects_count' in dir() else None
+            print(f"  [{i+1}/{len(new_commits)}] {len(suspects)} suspects so far...", flush=True)
 
         detail = github_get(f"/repos/{repo}/commits/{sha}")
         if not detail:
@@ -166,16 +125,9 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int
             time.sleep(REQUEST_DELAY)
             continue
 
-        combined_patch = "\n".join(f.get("patch", "") for f in files if f.get("patch"))
-        fingerprint_matches = match_fingerprints(combined_patch)
-
-        best_fp = fingerprint_matches[0] if fingerprint_matches else None
-        fp_score = best_fp["score"] if best_fp else 0.0
-
-        raw_combined = heuristic_result["score"] + (fp_score * 20)
-        normalized = heuristic_result["normalized_score"]
-        if best_fp:
-            normalized = min(normalized + (fp_score * 30), 100)
+        normalized, best_fp, fp_score = score_with_fingerprints(
+            heuristic_result, files
+        )
 
         top_file = heuristic_result["files"][0] if heuristic_result["files"] else {}
         author = detail.get("commit", {}).get("author", {})
@@ -206,7 +158,6 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int
             f.write(json.dumps(result, ensure_ascii=False) + "\n")
 
         suspects.append(result)
-        suspects_count = len(suspects)
 
         severity = "HIGH" if normalized >= 60 else "MEDIUM" if normalized >= 30 else "LOW"
         print(f"  [{i+1}/{len(new_commits)}] {severity} score={normalized} {sha[:8]} {message[:60]}")
@@ -221,7 +172,7 @@ def deep_scan(repo: str, since: str = None, until: str = None, max_commits: int
 
     if suspects:
         print(f"\nResults saved to: {results_path}")
-        print(f"\nTop suspects:")
+        print("\nTop suspects:")
         top = sorted(suspects, key=lambda s: s["normalized_score"], reverse=True)[:10]
         for s in top:
             print(f"  score={s['normalized_score']:5.1f}  {s['commit_sha'][:8]}  {s['message'][:60]}")

src/fetch.py

@@ -28,7 +28,7 @@ def graphql_request(query: str, variables: dict) -> dict:
             if "errors" in data:
                 raise RuntimeError(f"GraphQL errors: {data['errors']}")
             return data
-        except (requests.RequestException, RuntimeError) as exc:
+        except (requests.RequestException, RuntimeError):
             if attempt == RETRY_ATTEMPTS - 1:
                 raise
             time.sleep(RETRY_BACKOFF[attempt])

src/fingerprint.py

@@ -1,6 +1,5 @@
 import json
 import re
-from pathlib import Path
 from src.config import DATA_DIR
 
 
@@ -115,4 +114,26 @@ def get_best_match(patch_text: str) -> dict | None:
     matches = match_fingerprints(patch_text)
     if matches and matches[0]["score"] >= 0.1:
         return matches[0]
-    return None
+    return None
+
+
+def score_with_fingerprints(
+    heuristic_result: dict, files: list[dict]
+) -> tuple[float, dict | None, float]:
+    """Combine heuristic score with fingerprint matching.
+
+    Returns (normalized_score, best_fingerprint, fp_score).
+    """
+    combined_patch = "\n".join(
+        f.get("patch", "") for f in files if f.get("patch")
+    )
+    fingerprint_matches = match_fingerprints(combined_patch)
+
+    best_fp = fingerprint_matches[0] if fingerprint_matches else None
+    fp_score = best_fp["score"] if best_fp else 0.0
+
+    normalized = heuristic_result["normalized_score"]
+    if best_fp:
+        normalized = min(normalized + (fp_score * 30), 100)
+
+    return normalized, best_fp, fp_score

src/fingerprint_builder.py

@@ -2,7 +2,6 @@
 import re
 import sys
 from pathlib import Path
-from collections import defaultdict
 
 
 EXPERT_FINGERPRINTS = {
@@ -246,7 +245,7 @@ def build_fingerprints(output_path: str = "data/fingerprints.json"):
     print(f"  Expert CWE: {len(EXPERT_FINGERPRINTS)}")
     print(f"  OSDC live: {len([k for k in fingerprints if k.startswith('OSDC:')])}")
 
-    print(f"\nTop patterns by token count:")
+    print("\nTop patterns by token count:")
     top = sorted(fingerprints.items(), key=lambda x: len(x[1]["add_tokens"]) + len(x[1]["del_tokens"]), reverse=True)[:10]
     for pid, data in top:
         total = len(data["add_tokens"]) + len(data["del_tokens"])

src/github_api.py

@@ -0,0 +1,44 @@
+import time
+import requests
+from src.config import GITHUB_TOKEN
+
+
+GITHUB_API = "https://api.github.com"
+REQUEST_DELAY = 0.8
+
+
+def github_get(endpoint: str, params: dict = None) -> dict | list | None:
+    headers = {"Accept": "application/vnd.github.v3+json"}
+    if GITHUB_TOKEN:
+        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
+
+    url = f"{GITHUB_API}{endpoint}" if endpoint.startswith("/") else endpoint
+
+    try:
+        response = requests.get(url, headers=headers, params=params, timeout=30)
+        if response.status_code == 403:
+            remaining = response.headers.get("X-RateLimit-Remaining", "?")
+            reset = response.headers.get("X-RateLimit-Reset", "0")
+            if remaining == "0":
+                wait = max(int(reset) - int(time.time()), 10)
+                print(f"  Rate limited, waiting {wait}s...")
+                time.sleep(wait)
+                return github_get(endpoint, params)
+            print(f"  Rate limited (remaining: {remaining})")
+            return None
+        if response.status_code == 404:
+            return None
+        response.raise_for_status()
+        return response.json()
+    except requests.RequestException as exc:
+        print(f"  API error: {exc}")
+        return None
+
+
+def get_link_next(response_headers: dict) -> str | None:
+    link = response_headers.get("Link", "")
+    for part in link.split(","):
+        if 'rel="next"' in part:
+            url = part.split(";")[0].strip().strip("<>")
+            return url
+    return None