| 1 | #!/usr/bin/env python3 |
| 2 | """Quality threshold enforcement for SquadScope (warn-only). |
| 3 | |
| 4 | Checks whether scored repos meet minimum coverage thresholds defined in |
| 5 | the topic config. Emits GitHub Actions warning annotations but never |
| 6 | blocks the pipeline (always exits 0). |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import argparse |
| 12 | import json |
| 13 | import sys |
| 14 | from datetime import UTC, datetime |
| 15 | from pathlib import Path |
| 16 | from typing import Any |
| 17 | |
| 18 | try: # pragma: no cover |
| 19 | import yaml |
| 20 | except ImportError: # pragma: no cover |
| 21 | yaml = None |
| 22 | |
| 23 | from scripts.topic_paths import load_topic_id, metrics_dir |
| 24 | |
| 25 | DEFAULT_QUALITY = { |
| 26 | "min_repos_per_week": 5, |
| 27 | "max_repos_per_week": 30, |
| 28 | "min_quality_score": 60, |
| 29 | } |
| 30 | |
| 31 | |
| 32 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 33 | parser = argparse.ArgumentParser(description="Quality threshold gate (warn-only).") |
| 34 | parser.add_argument("--input", default=None, type=Path, help="Path to scored repos JSON file.") |
| 35 | parser.add_argument( |
| 36 | "--config", default="squadscope.topic.yml", type=Path, help="Path to topic config YAML." |
| 37 | ) |
| 38 | parser.add_argument("--topic", default=None, help="Topic ID override.") |
| 39 | return parser.parse_args(argv) |
| 40 | |
| 41 | |
| 42 | def load_config(path: Path) -> dict[str, Any]: |
| 43 | """Load topic config YAML. Returns empty dict on failure.""" |
| 44 | if not path.exists(): |
| 45 | return {} |
| 46 | text = path.read_text(encoding="utf-8") |
| 47 | if yaml is not None: |
| 48 | try: |
| 49 | return yaml.safe_load(text) or {} |
| 50 | except Exception: |
| 51 | return {} |
| 52 | # Minimal fallback: not needed in practice since yaml is available in CI |
| 53 | return {} # pragma: no cover |
| 54 | |
| 55 | |
| 56 | def get_quality_config(config: dict[str, Any]) -> dict[str, Any]: |
| 57 | """Extract quality section with defaults.""" |
| 58 | quality = config.get("quality", {}) |
| 59 | return {**DEFAULT_QUALITY, **quality} |
| 60 | |
| 61 | |
| 62 | def get_scoring_config(config: dict[str, Any]) -> dict[str, Any]: |
| 63 | """Extract scoring section.""" |
| 64 | return config.get("scoring", {}) |
| 65 | |
| 66 | |
| 67 | def load_scored_repos(path: Path) -> list[dict[str, Any]]: |
| 68 | """Load scored repos from JSON file.""" |
| 69 | if not path.exists(): |
| 70 | return [] |
| 71 | try: |
| 72 | data = json.loads(path.read_text(encoding="utf-8")) |
| 73 | if isinstance(data, list): |
| 74 | return data |
| 75 | return [] |
| 76 | except (json.JSONDecodeError, OSError): |
| 77 | return [] |
| 78 | |
| 79 | |
| 80 | def week_slug(dt: datetime | None = None) -> str: |
| 81 | """Return current ISO week slug like '2026-W21'.""" |
| 82 | dt = dt or datetime.now(tz=UTC) |
| 83 | year, week, _ = dt.isocalendar() |
| 84 | return f"{year}-W{week:02d}" |
| 85 | |
| 86 | |
| 87 | def check_quality( |
| 88 | scored_repos: list[dict[str, Any]], |
| 89 | quality_config: dict[str, Any], |
| 90 | scoring_config: dict[str, Any], |
| 91 | ) -> dict[str, Any]: |
| 92 | """Evaluate quality thresholds. Returns metric dict.""" |
| 93 | min_repos = quality_config.get("min_repos_per_week", 5) |
| 94 | max_repos = quality_config.get("max_repos_per_week", 30) |
| 95 | min_score = scoring_config.get("min_relevance_score", 40) |
| 96 | |
| 97 | # Count repos passing the relevance score threshold |
| 98 | repos_passing = sum(1 for r in scored_repos if r.get("relevance_score", 0) >= min_score) |
| 99 | repos_scored = len(scored_repos) |
| 100 | |
| 101 | warnings: list[str] = [] |
| 102 | status = "ok" |
| 103 | |
| 104 | if repos_passing < min_repos: |
| 105 | status = "below_threshold" |
| 106 | warnings.append( |
| 107 | f"Only {repos_passing} repos pass min_relevance_score ({min_score}), " |
| 108 | f"threshold is {min_repos}." |
| 109 | ) |
| 110 | |
| 111 | if repos_passing > max_repos: |
| 112 | status = "above_maximum" if status == "ok" else status |
| 113 | warnings.append( |
| 114 | f"{repos_passing} repos pass min_relevance_score ({min_score}), " |
| 115 | f"exceeds max_repos_per_week ({max_repos}). Potential noise." |
| 116 | ) |
| 117 | |
| 118 | return { |
| 119 | "repos_scored": repos_scored, |
| 120 | "repos_passing": repos_passing, |
| 121 | "threshold": min_repos, |
| 122 | "status": status, |
| 123 | "warnings": warnings, |
| 124 | } |
| 125 | |
| 126 | |
| 127 | def emit_warnings(warnings: list[str]) -> None: |
| 128 | """Print GitHub Actions warning annotations.""" |
| 129 | for warning in warnings: |
| 130 | print(f"::warning::{warning}") |
| 131 | |
| 132 | |
| 133 | def write_metric(topic_id: str, metric: dict[str, Any], week: str) -> Path: |
| 134 | """Write quality metric JSON to the metrics directory.""" |
| 135 | out_dir = metrics_dir(topic_id) |
| 136 | out_dir.mkdir(parents=True, exist_ok=True) |
| 137 | filename = f"quality-{week}.json" |
| 138 | out_path = out_dir / filename |
| 139 | payload = { |
| 140 | "week": week, |
| 141 | "topic": topic_id, |
| 142 | **{k: v for k, v in metric.items() if k != "warnings"}, |
| 143 | } |
| 144 | out_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") |
| 145 | return out_path |
| 146 | |
| 147 | |
| 148 | def main(argv: list[str] | None = None) -> int: |
| 149 | args = parse_args(argv) |
| 150 | |
| 151 | config = load_config(args.config) |
| 152 | quality_config = get_quality_config(config) |
| 153 | scoring_config = get_scoring_config(config) |
| 154 | topic_id = args.topic or load_topic_id(args.config) |
| 155 | |
| 156 | if args.input: |
| 157 | scored_repos = load_scored_repos(args.input) |
| 158 | else: |
| 159 | # Try to find scored output in analyzed dir |
| 160 | scored_repos = [] |
| 161 | print("::warning::No --input provided and no scored repos found.", file=sys.stderr) |
| 162 | |
| 163 | metric = check_quality(scored_repos, quality_config, scoring_config) |
| 164 | week = week_slug() |
| 165 | |
| 166 | emit_warnings(metric["warnings"]) |
| 167 | write_metric(topic_id, metric, week) |
| 168 | |
| 169 | if metric["status"] == "ok": |
| 170 | print( |
| 171 | f"✅ Quality gate passed: {metric['repos_passing']}/{metric['repos_scored']} repos meet threshold." |
| 172 | ) |
| 173 | else: |
| 174 | print( |
| 175 | f"⚠️ Quality gate warning: {metric['status']} ({metric['repos_passing']}/{metric['repos_scored']} repos)." |
| 176 | ) |
| 177 | |
| 178 | return 0 |
| 179 | |
| 180 | |
| 181 | if __name__ == "__main__": |
| 182 | raise SystemExit(main()) |