#!/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())