| 1 | #!/usr/bin/env python3 |
| 2 | """Score crawled repositories on topic relevance for SquadScope. |
| 3 | |
| 4 | Reads raw crawl JSON, applies scoring criteria from squadscope.topic.yml, |
| 5 | and outputs a ranked list of repos with relevance_score field. |
| 6 | |
| 7 | Usage: |
| 8 | python scripts/score_repos.py [--config squadscope.topic.yml] \ |
| 9 | [--input data/raw/ai-ml/2026-W21.json] [--output scored.json] [--topic ai-ml] |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import json |
| 16 | import math |
| 17 | import sys |
| 18 | from datetime import UTC, datetime |
| 19 | from pathlib import Path |
| 20 | from typing import Any |
| 21 | |
| 22 | import yaml |
| 23 | |
| 24 | from scripts.topic_paths import load_topic_id, raw_dir |
| 25 | |
| 26 | |
| 27 | def load_config(config_path: str | Path) -> dict[str, Any]: |
| 28 | """Load and return the full config from a YAML file.""" |
| 29 | path = Path(config_path) |
| 30 | if not path.exists(): |
| 31 | raise FileNotFoundError(f"Config file not found: {config_path}") |
| 32 | with open(path, encoding="utf-8") as f: |
| 33 | return yaml.safe_load(f) or {} |
| 34 | |
| 35 | |
| 36 | def get_scoring_config(config: dict[str, Any]) -> dict[str, Any]: |
| 37 | """Extract scoring section with defaults.""" |
| 38 | defaults = { |
| 39 | "min_stars": 20, |
| 40 | "min_stars_gained": 10, |
| 41 | "max_age_days": 365, |
| 42 | "min_relevance_score": 40, |
| 43 | "language_boost": {}, |
| 44 | "topic_relevance": [], |
| 45 | } |
| 46 | scoring = config.get("scoring", {}) |
| 47 | return {**defaults, **scoring} |
| 48 | |
| 49 | |
| 50 | def find_latest_raw_json(topic_id: str | None) -> Path | None: |
| 51 | """Find the most recent raw JSON file for a topic.""" |
| 52 | directory = raw_dir(topic_id) |
| 53 | if not directory.exists(): |
| 54 | return None |
| 55 | json_files = sorted(directory.glob("*.json"), reverse=True) |
| 56 | return json_files[0] if json_files else None |
| 57 | |
| 58 | |
| 59 | def score_stars(stars: int) -> float: |
| 60 | """Score based on star count with diminishing returns (0-25 points).""" |
| 61 | if stars <= 0: |
| 62 | return 0.0 |
| 63 | # Log scale: 100 stars ≈ 12.5, 1000 stars ≈ 18.7, 10000 stars ≈ 25 |
| 64 | return min(25.0, 25.0 * math.log10(stars) / math.log10(10000)) |
| 65 | |
| 66 | |
| 67 | def score_stars_gained(stars_gained: int) -> float: |
| 68 | """Score based on stars gained velocity (0-25 points).""" |
| 69 | if stars_gained <= 0: |
| 70 | return 0.0 |
| 71 | # Log scale with faster saturation |
| 72 | return min(25.0, 25.0 * math.log10(1 + stars_gained) / math.log10(1000)) |
| 73 | |
| 74 | |
| 75 | def score_language(language: str | None, language_boost: dict[str, float]) -> float: |
| 76 | """Score based on language match (0-15 points).""" |
| 77 | if not language or not language_boost: |
| 78 | return 7.5 # neutral score when no language info or no config |
| 79 | multiplier = language_boost.get(language, 1.0) |
| 80 | return min(15.0, 7.5 * multiplier) |
| 81 | |
| 82 | |
| 83 | def score_topics(repo_topics: list[str], topic_relevance: list[str]) -> float: |
| 84 | """Score based on topic overlap (0-25 points).""" |
| 85 | if not topic_relevance or not repo_topics: |
| 86 | return 0.0 |
| 87 | repo_set = set(t.lower() for t in repo_topics) |
| 88 | relevance_set = set(t.lower() for t in topic_relevance) |
| 89 | matches = len(repo_set & relevance_set) |
| 90 | max_possible = min(len(relevance_set), 3) # cap at 3 matches for full score |
| 91 | if max_possible == 0: |
| 92 | return 0.0 |
| 93 | return min(25.0, 25.0 * matches / max_possible) |
| 94 | |
| 95 | |
| 96 | def score_age(created_at: str | None, max_age_days: int) -> float: |
| 97 | """Score based on repo age (0-10 points). Newer repos get a boost.""" |
| 98 | if not created_at: |
| 99 | return 5.0 # neutral when unknown |
| 100 | try: |
| 101 | created = datetime.fromisoformat(created_at.replace("Z", "+00:00")) |
| 102 | except (ValueError, TypeError): |
| 103 | return 5.0 |
| 104 | now = datetime.now(UTC) |
| 105 | age_days = (now - created).days |
| 106 | if age_days < 0: |
| 107 | age_days = 0 |
| 108 | if age_days > max_age_days: |
| 109 | # Penalty for old repos: scale down from 5 to 0 |
| 110 | penalty_ratio = min((age_days - max_age_days) / max_age_days, 1.0) |
| 111 | return max(0.0, 5.0 * (1.0 - penalty_ratio)) |
| 112 | # Boost for newer repos: 0 days = 10, max_age_days = 5 |
| 113 | freshness = 1.0 - (age_days / max_age_days) |
| 114 | return 5.0 + 5.0 * freshness |
| 115 | |
| 116 | |
| 117 | def compute_relevance_score(repo: dict[str, Any], scoring_config: dict[str, Any]) -> float: |
| 118 | """Compute a 0-100 relevance score for a single repo.""" |
| 119 | stars = repo.get("stars", 0) or 0 |
| 120 | stars_gained = repo.get("stars_gained", 0) or 0 |
| 121 | language = repo.get("language") |
| 122 | topics = repo.get("topics", []) or [] |
| 123 | created_at = repo.get("created_at") |
| 124 | |
| 125 | s_stars = score_stars(stars) |
| 126 | s_gained = score_stars_gained(stars_gained) |
| 127 | s_lang = score_language(language, scoring_config.get("language_boost", {})) |
| 128 | s_topics = score_topics(topics, scoring_config.get("topic_relevance", [])) |
| 129 | s_age = score_age(created_at, scoring_config.get("max_age_days", 365)) |
| 130 | |
| 131 | raw_score = s_stars + s_gained + s_lang + s_topics + s_age |
| 132 | return round(min(100.0, max(0.0, raw_score)), 1) |
| 133 | |
| 134 | |
| 135 | def score_repos( |
| 136 | repos: list[dict[str, Any]], scoring_config: dict[str, Any] |
| 137 | ) -> list[dict[str, Any]]: |
| 138 | """Score and filter a list of repos. Returns sorted list with relevance_score.""" |
| 139 | min_score = scoring_config.get("min_relevance_score", 40) |
| 140 | scored = [] |
| 141 | for repo in repos: |
| 142 | score = compute_relevance_score(repo, scoring_config) |
| 143 | if score >= min_score: |
| 144 | scored.append({**repo, "relevance_score": score}) |
| 145 | scored.sort(key=lambda r: r["relevance_score"], reverse=True) |
| 146 | return scored |
| 147 | |
| 148 | |
| 149 | def main(argv: list[str] | None = None) -> int: |
| 150 | parser = argparse.ArgumentParser(description="Score repos on topic relevance") |
| 151 | parser.add_argument( |
| 152 | "--config", default="squadscope.topic.yml", help="Path to topic config YAML" |
| 153 | ) |
| 154 | parser.add_argument("--input", default=None, help="Path to raw crawl JSON file") |
| 155 | parser.add_argument("--output", default=None, help="Output file path (default: stdout)") |
| 156 | parser.add_argument("--topic", default=None, help="Topic ID override") |
| 157 | args = parser.parse_args(argv) |
| 158 | |
| 159 | config = load_config(args.config) |
| 160 | scoring_config = get_scoring_config(config) |
| 161 | |
| 162 | topic_id = args.topic or load_topic_id(args.config) |
| 163 | |
| 164 | if args.input: |
| 165 | input_path = Path(args.input) |
| 166 | else: |
| 167 | input_path = find_latest_raw_json(topic_id) |
| 168 | if input_path is None: |
| 169 | print(f"Error: No raw JSON found for topic '{topic_id}'", file=sys.stderr) |
| 170 | return 1 |
| 171 | |
| 172 | if not input_path.exists(): |
| 173 | print(f"Error: Input file not found: {input_path}", file=sys.stderr) |
| 174 | return 1 |
| 175 | |
| 176 | with open(input_path, encoding="utf-8") as f: |
| 177 | repos = json.load(f) |
| 178 | |
| 179 | if not isinstance(repos, list): |
| 180 | print("Error: Input JSON must be a list of repo objects", file=sys.stderr) |
| 181 | return 1 |
| 182 | |
| 183 | scored = score_repos(repos, scoring_config) |
| 184 | |
| 185 | output_json = json.dumps(scored, indent=2, ensure_ascii=False) |
| 186 | if args.output: |
| 187 | Path(args.output).parent.mkdir(parents=True, exist_ok=True) |
| 188 | with open(args.output, "w", encoding="utf-8") as f: |
| 189 | f.write(output_json + "\n") |
| 190 | else: |
| 191 | print(output_json) |
| 192 | |
| 193 | return 0 |
| 194 | |
| 195 | |
| 196 | if __name__ == "__main__": |
| 197 | sys.exit(main()) |