SimurgArray/scripts/jev/issue_triage.py

"""Triage of new Gitea issues: kind and area labels, missing information, and privacy flags.

Runs from .gitea/workflows/issue_triage.yml when an issue is opened or edited. Jev answers, in one request:

* the kind of issue (Choice: bug, feature, question, security, other) and the area of SimurgArray it concerns
  (Choice);
* three separate yes/no questions: does the text contain genetic data of identifiable people, contents of vendor
  files, or credentials (Noul each).

Code checks the template fields (version, command), applies the labels and, when the issue is opened, posts one
short comment. Low confidence answers get a ``needs-triage`` label instead of a guess. The issue title and text are
sent to the TypeSafe API; CONTRIBUTING.md says so.

    python scripts/jev/issue_triage.py --event event.json --dry-run     # print the planned actions only
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import urllib.request
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
from client import choice, noul, system_one

KINDS = {
    "bug": "Something fails, crashes, or gives a wrong or unexpected result",
    "feature": "A request for a new capability or a change in behaviour",
    "question": "A question about how to install, use or interpret SimurgArray",
    "security": "A vulnerability or a way to make SimurgArray leak data, run code or write outside its folders",
    "other": "Anything else, including spam",
}
AREAS = {
    "genotype": "IDAT reading, normalization, genotype calling, GTC files, sample summary",
    "qc": "Control probes, sample QC, contamination, relatedness, the HTML QC report",
    "vcf_export": "VCF, BCF, PLINK, bedGraph, final reports and other exports",
    "cn": "Copy number, loss of heterozygosity and mosaicism",
    "pgx": "Pharmacogenomics: targeted copy number, star alleles, phenotypes, the PGx database",
    "train": "ArrayTrain cluster file training",
    "methylation": "Methylation arrays and their QC",
    "compare": "The simurg compare differential testing tool and nightly runs",
    "install": "Installation, packaging, Docker, conda, the Rust kernels, platforms",
    "docs": "Documentation, examples, the website or blog",
    "other": "None of these areas",
}
PRIVACY = {
    "genetic_data": "Does `issue` contain genetic data of identifiable people, such as genotype calls, sequences, or "
                    "sample identifiers linked to real people? Public reference samples (1000 Genomes, GIAB, demo "
                    "data) do not count.",
    "vendor_content": "Does `issue` paste the contents of vendor files, such as manifest rows, cluster file "
                      "records or IDAT bytes, beyond a few locus names?",
    "credentials": "Does `issue` contain credentials: passwords, API keys, tokens or private keys?",
}  # fmt: skip
LABELS = {  # name: (colour, description)
    "kind/bug": ("C8453A", "Something is wrong"), "kind/feature": ("2F8F5B", "New capability"),
    "kind/question": ("A3A7C9", "Question"), "kind/security": ("8C6500", "Security report"),
    "kind/other": ("5A5E7E", "Other"), "needs-info": ("D4A017", "Version or command missing"),
    "needs-triage": ("1E2240", "Automatic triage was unsure"),
    "privacy-check": ("C8453A", "May contain personal data, vendor content or credentials"),
}  # fmt: skip
for _a in AREAS:
    if _a != "other":
        LABELS[f"area/{_a}"] = ("1E2240", AREAS[_a])
CONFIDENT, FLAG = 0.6, 0.5


def missing_info(body: str, kind: str) -> list[str]:
    """Template fields of a bug report that were left empty."""
    if kind != "bug":
        return []
    out = []
    for field, label in (("SimurgArray version", "the SimurgArray version"), ("Command", "the exact command")):
        m = re.search(rf"###\s*{field}\s*\n+(.*?)(?=\n###|\Z)", body, re.S)
        empty = m is None or not m.group(1).strip() or "_No response_" in m.group(1)
        version_elsewhere = field == "SimurgArray version" and re.search(r"\b\d+\.\d+\.\d+\b", body)
        if empty and not version_elsewhere:
            out.append(label)
    return out


def decide(title: str, body: str) -> dict[str, Any]:
    qs = {"kind": choice("What kind of issue is `issue`?", KINDS),
          "area": choice("Which part of SimurgArray does `issue` concern?", AREAS)}  # fmt: skip
    qs.update({k: noul(v) for k, v in PRIVACY.items()})
    qs["clinical"] = noul(
        "Is the author using SimurgArray results to make or check decisions about a patient's diagnosis or treatment?"
    )
    ans = system_one({"project": "SimurgArray, open source genotyping array analysis",
                      "issue": {"title": title, "body": body[:6000]}}, qs)  # fmt: skip
    kind, area = ans["kind"], ans["area"]
    labels: list[str] = []
    if float(kind["confidence"]) >= CONFIDENT:
        labels.append(f"kind/{kind['choice']}")
    if float(area["confidence"]) >= CONFIDENT and area["choice"] != "other":
        labels.append(f"area/{area['choice']}")
    if len(labels) < 2:
        labels.append("needs-triage")
    flags = [k for k in PRIVACY if float(ans[k]["noul"]) >= FLAG]
    if flags:
        labels.append("privacy-check")
    clinical = float(ans["clinical"]["noul"]) >= FLAG
    missing = missing_info(body, str(kind["choice"]))
    if missing:
        labels.append("needs-info")
    return {"kind": kind["choice"], "kind_confidence": round(float(kind["confidence"]), 2), "area": area["choice"],
            "area_confidence": round(float(area["confidence"]), 2), "labels": labels, "privacy": flags,
            "clinical": clinical, "missing": missing}  # fmt: skip


def comment_text(d: dict[str, Any]) -> str:
    parts = ["Thank you for the report. An automatic first triage added the labels "
             + ", ".join(f"`{x}`" for x in d["labels"]) + "; a maintainer will check them."]  # fmt: skip
    if d["kind"] == "security":
        parts.append("This looks like a security report. Please do not discuss details in public: write to "
                     "[email protected] and consider editing this issue to remove them.")  # fmt: skip
    if d["privacy"]:
        what = {"genetic_data": "genetic data of real people", "vendor_content": "contents of vendor files",
                "credentials": "credentials"}  # fmt: skip
        parts.append("The text may contain " + " and ".join(what[f] for f in d["privacy"]) + ". Please edit the "
                     "issue to remove it; issues are public.")  # fmt: skip
    if d.get("clinical"):
        parts.append(
            "A reminder: SimurgArray is a research tool. Its results, including pharmacogenomics, must not be used "
            "for diagnosis or treatment decisions; please rely on a validated clinical test."
        )
    if d["missing"]:
        parts.append("To reproduce the problem we need " + " and ".join(d["missing"]) + " (`simurg --version` "
                     "prints the version; the `run_manifest.json` of the run helps most).")  # fmt: skip
    parts.append("_Triage uses TypeSafe Jev on the issue text; see CONTRIBUTING.md._")
    return "\n\n".join(parts)


class Gitea:
    def __init__(self, api: str, repo: str, token: str) -> None:
        self.base, self.token = f"{api.rstrip('/')}/repos/{repo}", token

    def req(self, method: str, path: str, data: Any = None) -> Any:
        body = json.dumps(data).encode() if data is not None else None
        headers = {"Authorization": f"token {self.token}", "Content-Type": "application/json", "User-Agent": "triage"}
        r = urllib.request.Request(self.base + path, data=body, method=method, headers=headers)
        with urllib.request.urlopen(r, timeout=60) as resp:
            raw = resp.read()
        return json.loads(raw) if raw else None

    def label_ids(self, names: list[str]) -> list[int]:
        have = {lb["name"]: lb["id"] for lb in self.req("GET", "/labels?limit=100")}
        for n in names:
            if n not in have:
                col, desc = LABELS.get(n, ("5A5E7E", ""))
                have[n] = self.req("POST", "/labels", {"name": n, "color": f"#{col}", "description": desc})["id"]
        return [have[n] for n in names]


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--event", type=Path, default=Path(os.environ.get("GITHUB_EVENT_PATH", "event.json")))
    ap.add_argument("--dry-run", action="store_true")
    a = ap.parse_args(argv)
    ev = json.loads(a.event.read_text())
    issue, action = ev["issue"], ev.get("action", "opened")
    if issue.get("pull_request"):
        return 0
    d = decide(issue.get("title", ""), issue.get("body") or "")
    print(json.dumps({"issue": issue["number"], "action": action, **d}))
    if a.dry_run:
        print(comment_text(d))
        return 0
    g = Gitea(os.environ["GITEA_API"], os.environ["REPO"], os.environ["GITEA_TOKEN"])
    g.req("POST", f"/issues/{issue['number']}/labels", {"labels": g.label_ids(d["labels"])})
    if action == "opened":
        g.req("POST", f"/issues/{issue['number']}/comments", {"body": comment_text(d)})
    return 0


if __name__ == "__main__":
    sys.exit(main())