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