arm-ka-docs/scripts/regression/parser_bank.py

"""Прогон tabular-parser (CLI) по банку первички doc-bank и сравнение с прошлым прогоном (regress.sh, этап bank).

Берёт из doc-bank/index/index.csv файлы с route = «парсер» (xls/xlsx/xlsb), разбирает каждый CLI
`python -m src.parser.cli <файл>` в api/ сервиса (6 процессов, таймаут на файл) и пишет:
  results.jsonl — строка на файл: код возврата, время, статус файла, отпечатки документов (тип, статус, счёт, период,
                  число строк, сумма денежных колонок, коды предупреждений);
  summary.md    — счётчики статусов и переходы против базы: прошлый прогон (--baseline DIR с results.jsonl) или,
                  если его нет, колонка parse_status индекса (прогон 20.10.2026, main 6c81199).
Переходы «лучше → хуже» (done → needs_check/ошибка, needs_check → ошибка) — кандидаты в регрессию, их надо
просмотреть; изменения отпечатков при том же статусе перечислены отдельно. Код возврата 0 всегда, кроме сбоя самого
скрипта: решение «регрессия или исправление» за человеком.

  python3 -I parser_bank.py --parser-api ../tabular-parser/api --bank ../doc-bank --out DIR [--baseline DIR] [--jobs 6]
"""

import argparse
import csv
import hashlib
import json
import os
import subprocess
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal, InvalidOperation
from pathlib import Path

MONEY = ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit", "closing_debit", "closing_credit")
RANK = {"done": 0, "needs_check": 1, "failed": 2, "error": 2, "timeout": 3}
INDEX_STATUS = {"done": "done", "needs_check": "needs_check", "ошибка": "error", "ложный done": "done"}


def money_sum(rows: list[dict]) -> str:
    total = Decimal(0)
    for r in rows:
        for k in MONEY:
            v = r.get(k)
            if v not in (None, ""):
                try:
                    total += Decimal(str(v))
                except InvalidOperation:
                    pass
    return str(total)


def fingerprint(doc: dict) -> dict:
    h = doc.get("header") or {}
    rows = doc.get("rows") or []
    lines = doc.get("lines") or []
    fp = {
        "type": doc.get("document_type"),
        "status": doc.get("status"),
        "locator": doc.get("locator"),
        "account": h.get("account_code_raw"),
        "period": [h.get("period_year"), h.get("period_quarter"), h.get("is_cumulative")],
        "rows": len(rows) or len(lines),
        "sum": money_sum(rows),
        "warnings": sorted({w.get("code") for w in doc.get("warnings") or []}),
    }
    if lines:
        fp["lines_hash"] = hashlib.sha1(json.dumps(lines, sort_keys=True, default=str).encode()).hexdigest()[:12]
    return fp


def parse_one(api: Path, file: Path, timeout: int) -> dict:
    t0 = time.time()
    env = dict(os.environ, ENV_FILE=".env.test", APP_ENV="test")
    try:
        p = subprocess.run(
            [str(api / ".venv/bin/python"), "-m", "src.parser.cli", str(file)],
            cwd=api, env=env, capture_output=True, timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return {"rc": "timeout", "sec": timeout, "status": "timeout", "docs": []}
    res = {"rc": p.returncode, "sec": round(time.time() - t0, 2)}
    try:
        out = json.loads(p.stdout)
    except ValueError:
        res |= {"status": "error", "error": (p.stderr.decode(errors="replace").strip().splitlines() or ["?"])[-1][:300], "docs": []}
        return res
    res["status"] = out.get("status")
    if out.get("error"):
        res["error"] = str(out["error"])[:300]
    res["docs"] = [fingerprint(d) for d in out.get("documents") or []]
    return res


def norm(status: str | None) -> str:
    return {"failed": "error"}.get(status or "", status or "")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--parser-api", required=True, type=Path)
    ap.add_argument("--bank", required=True, type=Path)
    ap.add_argument("--out", required=True, type=Path)
    ap.add_argument("--baseline", type=Path)
    ap.add_argument("--jobs", type=int, default=6)
    ap.add_argument("--timeout", type=int, default=300)
    a = ap.parse_args()
    a.out.mkdir(parents=True, exist_ok=True)
    index = [r for r in csv.DictReader((a.bank / "index/index.csv").open()) if r["route"] == "парсер"]
    files = sorted(index, key=lambda r: -int(r["size"] or 0))  # большие первыми — короче хвост
    t0 = time.time()
    results: dict[str, dict] = {}

    def work(r: dict) -> tuple[str, dict]:
        return r["file"], parse_one(a.parser_api.resolve(), (a.bank / r["file"]).resolve(), a.timeout)

    with ThreadPoolExecutor(a.jobs) as ex:
        for i, (name, res) in enumerate(ex.map(work, files), 1):
            results[name] = res
            if i % 200 == 0:
                print(f"{i}/{len(files)} за {int(time.time() - t0)} с", flush=True)
    with (a.out / "results.jsonl").open("w") as f:
        for r in index:
            f.write(json.dumps({"file": r["file"], "name": r["name"]} | results[r["file"]], ensure_ascii=False) + "\n")

    base: dict[str, dict] = {}
    base_name = ""
    if a.baseline and (a.baseline / "results.jsonl").exists():
        base_name = f"прошлый прогон `{a.baseline}`"
        for ln in (a.baseline / "results.jsonl").read_text().splitlines():
            x = json.loads(ln)
            base[x["file"]] = x
    else:
        base_name = "индекс doc-bank (parse_status, прогон 20.10.2026)"
        for r in index:
            if r["parse_status"] in INDEX_STATUS:
                base[r["file"]] = {"status": INDEX_STATUS[r["parse_status"]], "docs": None}

    names = {r["file"]: r["name"] for r in index}
    cur = Counter(norm(x["status"]) for x in results.values())
    worse, better, changed = [], [], []
    for file, res in results.items():
        b = base.get(file)
        if not b:
            continue
        s0, s1 = norm(b["status"]), norm(res["status"])
        if RANK.get(s1, 9) > RANK.get(s0, 9):
            worse.append((file, s0, s1, res.get("error", "")))
        elif RANK.get(s1, 9) < RANK.get(s0, 9):
            better.append((file, s0, s1))
        elif b.get("docs") is not None and b["docs"] != res["docs"]:
            changed.append((file, s1))
    md = [
        "## Парсер по банку первички (doc-bank)",
        "",
        f"Файлов: {len(results)} за {int(time.time() - t0)} с; статусы: " + ", ".join(f"{k} {v}" for k, v in cur.most_common()),
        f"База сравнения: {base_name}.",
        "",
        f"- хуже, чем в базе: **{len(worse)}** (кандидаты в регрессию — просмотреть)",
        f"- лучше: {len(better)}",
        f"- тот же статус, другой разбор (отпечаток документов): {len(changed)}",
    ]
    if worse:
        md += ["", "| Файл | Было | Стало | Ошибка |", "|---|---|---|---|"]
        md += [f"| {names[f]} | {s0} | {s1} | {e.replace('|', '/')[:120]} |" for f, s0, s1, e in sorted(worse, key=lambda x: names[x[0]])[:100]]
        if len(worse) > 100:
            md.append(f"| … ещё {len(worse) - 100} | | | |")
    tr = Counter(f"{s0} → {s1}" for _, s0, s1 in better)
    if tr:
        md += ["", "Переходы к лучшему: " + ", ".join(f"{k}: {v}" for k, v in tr.most_common())]
    (a.out / "summary.md").write_text("\n".join(md) + "\n")
    (a.out / "transitions.json").write_text(
        json.dumps({"worse": worse, "better": better, "changed": changed}, ensure_ascii=False, indent=1)
    )
    print("\n".join(md))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())