forked from bots-garden/ori
| 🎉 Begin a project. | 1 | #!/usr/bin/env python3 |
| 2 | """Measure project code quality with qlty and emit a report, a history line, and a gate verdict. | |
| 3 | ||
| 4 | Read-only with respect to the source tree: `qlty check` runs with --no-fix so a | |
| 5 | measurement never silently rewrites the code it is measuring. | |
| 6 | ||
| 7 | Exit codes: 0 gate passed, 1 gate failed, 2 tooling/setup error. | |
| 8 | """ | |
| 9 | ||
| 10 | from __future__ import annotations | |
| 11 | ||
| 12 | import argparse | |
| 13 | import json | |
| 14 | import re | |
| 15 | import shutil | |
| 16 | ||
| 17 | # nosec B404 - running qlty *is* this script's job. Every command it launches is a | |
| 18 | # literal argv list built in this file; none is assembled from user input, and none | |
| 19 | # goes through a shell. See the matching justification on subprocess.run below. | |
| 20 | import subprocess # nosec B404 | |
| 21 | import sys | |
| 22 | from datetime import datetime, timezone | |
| 23 | from pathlib import Path | |
| 24 | ||
| 25 | ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") | |
| 26 | ||
| 27 | # Overridable per project by writing .quality/gate.json with any subset of these | |
| 28 | # keys. Kept strict on purpose: the gate is what the refactoring loop drives to | |
| 29 | # zero. A null value disables that check. | |
| 30 | DEFAULT_GATE = { | |
| 31 | "max_error": 0, | |
| 32 | "max_warning": 0, | |
| 33 | "max_note": None, | |
| 34 | "max_smells": 0, | |
| 35 | "max_file_complexity": None, | |
| 36 | "max_total_complexity": None, | |
| 37 | } | |
| 38 | ||
| 39 | QLTY_COMMON = ["--no-upgrade-check"] | |
| 40 | ||
| 41 | ||
| 42 | def run(cmd: list[str], cwd: Path, timeout: int = 1800) -> tuple[int, str, str]: | |
| 43 | try: | |
| 44 | # nosec B603 - cmd is always a literal argv list from this file (qlty or git | |
| 45 | # with fixed subcommands), never a string and never shell-interpreted, so | |
| 46 | # there is no injection surface. shell=False is the default and is what makes | |
| 47 | # this the safe form rather than the risky one. | |
| 48 | p = subprocess.run( # nosec B603 | |
| 49 | cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout | |
| 50 | ) | |
| 51 | except FileNotFoundError: | |
| 52 | return 127, "", f"command not found: {cmd[0]}" | |
| 53 | except subprocess.TimeoutExpired: | |
| 54 | return 124, "", f"timed out after {timeout}s: {' '.join(cmd)}" | |
| 55 | return p.returncode, p.stdout, p.stderr | |
| 56 | ||
| 57 | ||
| 58 | def git(args: list[str], cwd: Path) -> str: | |
| 59 | code, out, _ = run(["git"] + args, cwd, timeout=60) | |
| 60 | return out.strip() if code == 0 else "" | |
| 61 | ||
| 62 | ||
| 63 | def ensure_setup(ws: Path, allow_init: bool) -> None: | |
| 64 | """Verify qlty is usable here, running `qlty init` when the project has no config.""" | |
| 65 | if shutil.which("qlty") is None: | |
| 66 | die( | |
| 67 | "qlty is not installed. This project expects the `dev-toolkit` sbx kit " | |
| 68 | "(see kits/dev-toolkit/README.md)." | |
| 69 | ) | |
| 70 | if not (ws / ".git").exists() and not git(["rev-parse", "--git-dir"], ws): | |
| 71 | die(f"{ws} is not a git repository — qlty needs git to scope its analysis.") | |
| 72 | ||
| 73 | if (ws / ".qlty" / "qlty.toml").is_file(): | |
| 74 | return | |
| 75 | if not allow_init: | |
| 76 | die("no .qlty/qlty.toml found and --no-init was passed; run `qlty init --yes`.") | |
| 77 | ||
| 78 | log("no .qlty/qlty.toml found — running `qlty init --yes` (first-time setup)") | |
| 79 | code, out, err = run(["qlty", "init", "--yes"] + QLTY_COMMON, ws, timeout=1800) | |
| 80 | if code != 0 or not (ws / ".qlty" / "qlty.toml").is_file(): | |
| 81 | die(f"`qlty init` failed (exit {code}).\n{(out + err).strip()[-2000:]}") | |
| 82 | log("qlty initialised: .qlty/qlty.toml written") | |
| 83 | ||
| 84 | ||
| 85 | def sarif_location(res: dict) -> tuple[str, int | None]: | |
| 86 | """Return the file and start line of a SARIF result, or ("", None) if it has none. | |
| 87 | ||
| 88 | Every level of the structure is optional in the spec and plugins do omit them, | |
| 89 | hence the `or {}` at each step rather than a single chained access. | |
| 90 | """ | |
| 91 | locs = res.get("locations") or [] | |
| 92 | if not locs: | |
| 93 | return "", None | |
| 94 | phys = (locs[0] or {}).get("physicalLocation") or {} | |
| 95 | uri = ((phys.get("artifactLocation") or {}).get("uri")) or "" | |
| 96 | return uri, (phys.get("region") or {}).get("startLine") | |
| 97 | ||
| 98 | ||
| 99 | def sarif_finding(res: dict) -> dict: | |
| 100 | """Flatten one SARIF result into the shape the rest of this script uses.""" | |
| 101 | uri, line = sarif_location(res) | |
| 102 | return { | |
| 103 | "rule": res.get("ruleId") or "unknown", | |
| 104 | # SARIF omits `level` when it equals the rule default; treat an absent | |
| 105 | # level as "warning" rather than dropping the finding. | |
| 106 | "level": (res.get("level") or "warning").lower(), | |
| 107 | "message": ((res.get("message") or {}).get("text") or "").strip(), | |
| 108 | "file": uri, | |
| 109 | "line": line, | |
| 110 | } | |
| 111 | ||
| 112 | ||
| 113 | def sarif_results(payload: str) -> list[dict]: | |
| 114 | """Flatten SARIF runs into a list of result dicts, tolerating partial output.""" | |
| 115 | try: | |
| 116 | doc = json.loads(payload) | |
| 117 | except json.JSONDecodeError: | |
| 118 | return [] | |
| 119 | return [ | |
| 120 | sarif_finding(res) | |
| 121 | for run_ in doc.get("runs") or [] | |
| 122 | for res in run_.get("results") or [] | |
| 123 | ] | |
| 124 | ||
| 125 | ||
| 126 | def collect_check(ws: Path) -> tuple[list[dict], str]: | |
| 127 | # --no-fail / --no-error: always emit a report instead of exiting on findings. | |
| 128 | # --no-fix: measurement must not mutate the tree. | |
| 129 | cmd = [ | |
| 130 | "qlty", "check", "--all", "--no-fix", "--no-fail", "--no-error", | |
| 131 | "--no-progress", "--sarif", | |
| 132 | ] + QLTY_COMMON | |
| 133 | code, out, err = run(cmd, ws) | |
| 134 | results = sarif_results(out) | |
| 135 | note = "" if (results or code == 0) else f"`qlty check` exit {code}: {err.strip()[-500:]}" | |
| 136 | return results, note | |
| 137 | ||
| 138 | ||
| 139 | def collect_smells(ws: Path) -> tuple[list[dict], str]: | |
| 140 | cmd = ["qlty", "smells", "--all", "--quiet", "--sarif"] + QLTY_COMMON | |
| 141 | code, out, err = run(cmd, ws) | |
| 142 | results = sarif_results(out) | |
| 143 | note = "" if (results or code == 0) else f"`qlty smells` exit {code}: {err.strip()[-500:]}" | |
| 144 | return results, note | |
| 145 | ||
| 146 | ||
| 147 | def table_cells(text: str) -> list[list[str]]: | |
| 148 | """Split a pipe table into rows of cells, dropping rules and blank lines. | |
| 149 | ||
| 150 | qlty renders an ANSI-coloured table and ignores NO_COLOR, so the escape codes | |
| 151 | are stripped here rather than by the caller. | |
| 152 | """ | |
| 153 | rows: list[list[str]] = [] | |
| 154 | for raw in text.splitlines(): | |
| 155 | line = ANSI.sub("", raw).strip() | |
| 156 | if "|" not in line or set(line) <= set("-+| "): | |
| 157 | continue | |
| 158 | rows.append([c.strip() for c in line.split("|")]) | |
| 159 | return rows | |
| 160 | ||
| 161 | ||
| 162 | def metric_row(header: list[str], cells: list[str]) -> dict: | |
| 163 | """Pair a data row with its header, keeping `name` textual and the rest numeric.""" | |
| 164 | row: dict = {} | |
| 165 | for key, val in zip(header, cells): | |
| 166 | if key == "name": | |
| 167 | row["name"] = val | |
| 168 | continue | |
| 169 | try: | |
| 170 | row[key] = int(val) | |
| 171 | except ValueError: | |
| 172 | row[key] = val | |
| 173 | return row | |
| 174 | ||
| 175 | ||
| 176 | def parse_metrics_table(text: str) -> tuple[dict, list[dict]]: | |
| 177 | """Split qlty's metrics table into its TOTAL row and its per-file rows. | |
| 178 | ||
| 179 | A row whose width does not match the header is skipped rather than guessed at: | |
| 180 | qlty occasionally wraps a long path, and inventing a value would silently | |
| 181 | corrupt the numbers this whole report rests on. | |
| 182 | """ | |
| 183 | header: list[str] = [] | |
| 184 | total: dict = {} | |
| 185 | rows: list[dict] = [] | |
| 186 | for cells in table_cells(text): | |
| 187 | if not header: | |
| 188 | header = [c.lower() for c in cells] | |
| 189 | continue | |
| 190 | if len(cells) != len(header): | |
| 191 | continue | |
| 192 | row = metric_row(header, cells) | |
| 193 | if row.get("name", "").upper() == "TOTAL": | |
| 194 | total = {k: v for k, v in row.items() if k != "name"} | |
| 195 | else: | |
| 196 | rows.append(row) | |
| 197 | return total, rows | |
| 198 | ||
| 199 | ||
| 200 | def collect_metrics(ws: Path) -> tuple[dict, list[dict], str]: | |
| 201 | """Run `qlty metrics` and return its TOTAL row, its per-file rows, and any note.""" | |
| 202 | cmd = ["qlty", "metrics", "--all", "--quiet"] + QLTY_COMMON | |
| 203 | code, out, err = run(cmd, ws) | |
| 204 | if code != 0 and not out.strip(): | |
| 205 | return {}, [], f"`qlty metrics` exit {code}: {err.strip()[-500:]}" | |
| 206 | ||
| 207 | total, rows = parse_metrics_table(out) | |
| 208 | return total, rows, "" | |
| 209 | ||
| 210 | ||
| 211 | def load_gate(ws: Path, override: Path | None) -> dict: | |
| 212 | gate = dict(DEFAULT_GATE) | |
| 213 | path = override or (ws / ".quality" / "gate.json") | |
| 214 | if path.is_file(): | |
| 215 | try: | |
| 216 | gate.update(json.loads(path.read_text())) | |
| 217 | except (json.JSONDecodeError, OSError) as exc: | |
| 218 | die(f"could not read gate file {path}: {exc}") | |
| 219 | return gate | |
| 220 | ||
| 221 | ||
| 222 | def evaluate(gate: dict, counts: dict, smells: int, metrics: dict, files: list[dict]) -> list[str]: | |
| 223 | """Return one human-readable violation string per breached threshold.""" | |
| 224 | breaches: list[str] = [] | |
| 225 | ||
| 226 | def check(limit_key: str, actual: int, label: str) -> None: | |
| 227 | limit = gate.get(limit_key) | |
| 228 | if limit is not None and actual > limit: | |
| 229 | breaches.append(f"{label}: {actual} (max {limit})") | |
| 230 | ||
| 231 | check("max_error", counts.get("error", 0), "error-level issues") | |
| 232 | check("max_warning", counts.get("warning", 0), "warning-level issues") | |
| 233 | check("max_note", counts.get("note", 0), "note-level issues") | |
| 234 | check("max_smells", smells, "code smells") | |
| 235 | check("max_total_complexity", metrics.get("complex", 0), "total complexity") | |
| 236 | ||
| 237 | limit = gate.get("max_file_complexity") | |
| 238 | if limit is not None: | |
| 239 | over = [f for f in files if isinstance(f.get("complex"), int) and f["complex"] > limit] | |
| 240 | for f in sorted(over, key=lambda r: -r["complex"])[:10]: | |
| 241 | breaches.append(f"{f['name']} complexity {f['complex']} (max {limit})") | |
| 242 | return breaches | |
| 243 | ||
| 244 | ||
| 245 | def tally(results: list[dict], key: str) -> dict[str, int]: | |
| 246 | counts: dict[str, int] = {} | |
| 247 | for r in results: | |
| 248 | counts[r[key]] = counts.get(r[key], 0) + 1 | |
| 249 | return counts | |
| 250 | ||
| 251 | ||
| 252 | def log(msg: str) -> None: | |
| 253 | print(f"[quality] {msg}", file=sys.stderr) | |
| 254 | ||
| 255 | ||
| 256 | def die(msg: str) -> None: | |
| 257 | print(f"[quality] error: {msg}", file=sys.stderr) | |
| 258 | sys.exit(2) | |
| 259 | ||
| 260 | ||
| 261 | def md_table(headers: list[str], rows: list[list[str]]) -> str: | |
| 262 | if not rows: | |
| 263 | return "_none_\n" | |
| 264 | ||
| 265 | out = ["| " + " | ".join(headers) + " |", "|" + "|".join(["---"] * len(headers)) + "|"] | |
| 266 | out += ["| " + " | ".join(rows_) + " |" for rows_ in (map(str, r) for r in rows)] | |
| 267 | return "\n".join(out) + "\n" | |
| 268 | ||
| 269 | ||
| 270 | def delta(cur: int, prev: int | None) -> str: | |
| 271 | if prev is None: | |
| 272 | return "—" | |
| 273 | d = cur - prev | |
| 274 | if d == 0: | |
| 275 | return "±0" | |
| 276 | return f"{d:+d}" | |
| 277 | ||
| 278 | ||
| 279 | def report_header(entry: dict, prev: dict | None, breaches: list[str]) -> list[str]: | |
| 280 | verdict = "✅ **PASS**" if entry["gate_passed"] else "❌ **FAIL**" | |
| 281 | lines = [ | |
| 282 | f"# Quality report — {entry['timestamp']}", | |
| 283 | "", | |
| 284 | f"- **Gate**: {verdict}", | |
| 285 | f"- **Commit**: `{entry['commit'] or 'n/a'}` on `{entry['branch'] or 'n/a'}`", | |
| 286 | f"- **qlty**: {entry['qlty_version']}", | |
| 287 | f"- **Run**: #{entry['run']}" | |
| 288 | + (f" (previous: {prev['timestamp']})" if prev else " (first recorded run)"), | |
| 289 | "", | |
| 290 | ] | |
| 291 | if breaches: | |
| 292 | lines += ["## Gate violations", ""] | |
| 293 | lines += [f"- {b}" for b in breaches] | |
| 294 | lines.append("") | |
| 295 | return lines | |
| 296 | ||
| 297 | ||
| 298 | def report_issues(entry: dict, prev: dict | None, check_res: list[dict]) -> list[str]: | |
| 299 | """The lint section: counts by level, then the two "where to start" tables.""" | |
| 300 | c = entry["counts"] | |
| 301 | p_counts = (prev or {}).get("counts") or {} | |
| 302 | ||
| 303 | lines = ["## Lint issues (`qlty check`)", ""] | |
| 304 | lines.append( | |
| 305 | md_table( | |
| 306 | ["level", "count", "vs previous"], | |
| 307 | [[lvl, c.get(lvl, 0), delta(c.get(lvl, 0), p_counts.get(lvl))] | |
| 308 | for lvl in ("error", "warning", "note", "none") | |
| 309 | if c.get(lvl) or p_counts.get(lvl)], | |
| 310 | ) | |
| 311 | ) | |
| 312 | ||
| 313 | by_rule = sorted(tally(check_res, "rule").items(), key=lambda kv: -kv[1])[:15] | |
| 314 | lines += ["### Top rules", "", md_table(["rule", "count"], [[k, v] for k, v in by_rule])] | |
| 315 | ||
| 316 | by_file = sorted(tally([r for r in check_res if r["file"]], "file").items(), | |
| 317 | key=lambda kv: -kv[1])[:15] | |
| 318 | lines += ["### Most affected files", "", | |
| 319 | md_table(["file", "issues"], [[k, v] for k, v in by_file])] | |
| 320 | return lines | |
| 321 | ||
| 322 | ||
| 323 | def report_smells(entry: dict, prev: dict | None, smell_res: list[dict]) -> list[str]: | |
| 324 | lines = ["## Code smells (`qlty smells`)", "", | |
| 325 | f"Total: **{entry['smells']}** (vs previous: " | |
| 326 | f"{delta(entry['smells'], (prev or {}).get('smells'))})", ""] | |
| 327 | smell_rows = [[r["rule"], r["file"] or "—", r["line"] or "—", r["message"][:110]] | |
| 328 | for r in smell_res[:20]] | |
| 329 | lines.append(md_table(["smell", "file", "line", "detail"], smell_rows)) | |
| 330 | return lines | |
| 331 | ||
| 332 | ||
| 333 | def report_metrics(entry: dict, prev: dict | None, files: list[dict]) -> list[str]: | |
| 334 | m = entry["metrics"] | |
| 335 | p_metrics = (prev or {}).get("metrics") or {} | |
| 336 | ||
| 337 | lines = ["## Metrics (`qlty metrics`)", ""] | |
| 338 | metric_rows = [ | |
| 339 | [k, m.get(k, "—"), delta(m[k], p_metrics.get(k)) if isinstance(m.get(k), int) else "—"] | |
| 340 | for k in ("funcs", "classes", "fields", "cyclo", "complex", "lcom", "lines", "loc") | |
| 341 | if k in m | |
| 342 | ] | |
| 343 | lines.append(md_table(["metric", "total", "vs previous"], metric_rows)) | |
| 344 | ||
| 345 | worst = sorted([f for f in files if isinstance(f.get("complex"), int)], | |
| 346 | key=lambda r: -r["complex"])[:15] | |
| 347 | lines += ["### Most complex files", "", | |
| 348 | md_table(["file", "complex", "cyclo", "loc"], | |
| 349 | [[f["name"], f.get("complex", "—"), f.get("cyclo", "—"), | |
| 350 | f.get("loc", "—")] for f in worst])] | |
| 351 | return lines | |
| 352 | ||
| 353 | ||
| 354 | def report_trend(entry: dict) -> list[str]: | |
| 355 | return ["## Trend", "", md_table( | |
| 356 | ["run", "timestamp", "error", "warning", "smells", "complex", "gate"], | |
| 357 | [[h["run"], h["timestamp"], (h.get("counts") or {}).get("error", 0), | |
| 358 | (h.get("counts") or {}).get("warning", 0), h.get("smells", 0), | |
| 359 | (h.get("metrics") or {}).get("complex", "—"), | |
| 360 | "PASS" if h.get("gate_passed") else "FAIL"] | |
| 361 | for h in entry["_trend"]], | |
| 362 | )] | |
| 363 | ||
| 364 | ||
| 365 | def build_report(entry: dict, prev: dict | None, check_res: list[dict], | |
| 366 | smell_res: list[dict], files: list[dict], breaches: list[str], | |
| 367 | notes: list[str]) -> str: | |
| 368 | """Assemble the Markdown report, in the order a reader works through it.""" | |
| 369 | lines = report_header(entry, prev, breaches) | |
| 370 | lines += report_issues(entry, prev, check_res) | |
| 371 | lines += report_smells(entry, prev, smell_res) | |
| 372 | lines += report_metrics(entry, prev, files) | |
| 373 | ||
| 374 | if notes: | |
| 375 | lines += ["## Tooling notes", ""] + [f"- {n}" for n in notes] + [""] | |
| 376 | ||
| 377 | lines += report_trend(entry) | |
| 378 | return "\n".join(lines) | |
| 379 | ||
| 380 | ||
| 381 | def main() -> int: | |
| 382 | ap = argparse.ArgumentParser(description=__doc__) | |
| 383 | ap.add_argument("--workspace", default=".", help="project root (default: cwd)") | |
| 384 | ap.add_argument("--no-init", action="store_true", | |
| 385 | help="fail instead of running `qlty init` when unconfigured") | |
| 386 | ap.add_argument("--gate", type=Path, help="gate JSON file (default: .quality/gate.json)") | |
| 387 | ap.add_argument("--json", action="store_true", help="print the history entry to stdout") | |
| 388 | args = ap.parse_args() | |
| 389 | ||
| 390 | ws = Path(args.workspace).resolve() | |
| 391 | if not ws.is_dir(): | |
| 392 | die(f"workspace {ws} does not exist") | |
| 393 | ||
| 394 | ensure_setup(ws, allow_init=not args.no_init) | |
| 395 | ||
| 396 | notes: list[str] = [] | |
| 397 | log("running qlty check / smells / metrics (this can take several minutes on a cold cache)") | |
| 398 | check_res, n1 = collect_check(ws) | |
| 399 | smell_res, n2 = collect_smells(ws) | |
| 400 | metrics, files, n3 = collect_metrics(ws) | |
| 401 | notes += [n for n in (n1, n2, n3) if n] | |
| 402 | ||
| 403 | counts = tally(check_res, "level") | |
| 404 | gate = load_gate(ws, args.gate) | |
| 405 | breaches = evaluate(gate, counts, len(smell_res), metrics, files) | |
| 406 | ||
| 407 | _, ver_out, _ = run(["qlty", "--version"], ws, timeout=60) | |
| 408 | outdir = ws / ".quality" | |
| 409 | outdir.mkdir(parents=True, exist_ok=True) | |
| 410 | history = outdir / "history.jsonl" | |
| 411 | ||
| 412 | past: list[dict] = [] | |
| 413 | if history.is_file(): | |
| 414 | for line in history.read_text().splitlines(): | |
| 415 | line = line.strip() | |
| 416 | if line: | |
| 417 | try: | |
| 418 | past.append(json.loads(line)) | |
| 419 | except json.JSONDecodeError: | |
| 420 | continue | |
| 421 | ||
| 422 | now = datetime.now(timezone.utc) | |
| 423 | entry = { | |
| 424 | "run": len(past) + 1, | |
| 425 | "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), | |
| 426 | "commit": git(["rev-parse", "--short", "HEAD"], ws), | |
| 427 | "branch": git(["rev-parse", "--abbrev-ref", "HEAD"], ws), | |
| 428 | "qlty_version": ANSI.sub("", ver_out).strip() or "unknown", | |
| 429 | "counts": counts, | |
| 430 | "smells": len(smell_res), | |
| 431 | "metrics": metrics, | |
| 432 | "gate": {k: v for k, v in gate.items() if v is not None}, | |
| 433 | "gate_passed": not breaches, | |
| 434 | "breaches": breaches, | |
| 435 | } | |
| 436 | ||
| 437 | with history.open("a") as fh: | |
| 438 | fh.write(json.dumps(entry, sort_keys=True) + "\n") | |
| 439 | ||
| 440 | entry["_trend"] = (past + [entry])[-10:] | |
| 441 | report = build_report(entry, past[-1] if past else None, | |
| 442 | check_res, smell_res, files, breaches, notes) | |
| 443 | del entry["_trend"] | |
| 444 | ||
| 445 | stamp = now.strftime("%Y%m%dT%H%M%SZ") | |
| 446 | (outdir / f"report-{stamp}.md").write_text(report) | |
| 447 | (outdir / "report-latest.md").write_text(report) | |
| 448 | ||
| 449 | print(report) | |
| 450 | log(f"report written to .quality/report-{stamp}.md (and report-latest.md)") | |
| 451 | log(f"history appended to .quality/history.jsonl (run #{entry['run']})") | |
| 452 | if args.json: | |
| 453 | print(json.dumps(entry, indent=2, sort_keys=True)) | |
| 454 | ||
| 455 | if breaches: | |
| 456 | log(f"GATE FAILED — {len(breaches)} violation(s)") | |
| 457 | return 1 | |
| 458 | log("GATE PASSED") | |
| 459 | return 0 | |
| 460 | ||
| 461 | ||
| 462 | if __name__ == "__main__": | |
| 463 | sys.exit(main()) |