shared_resources/tree_duplicate_identity_byteshark/analyze_duplicate_identity.py

#!/usr/bin/env python3
"""Summarize duplicate-row identity probe output from job logs."""

from __future__ import annotations

import argparse
import re
from pathlib import Path


LINE_RE = re.compile(
    r"\[tree-dup-id\]\s+stage=(?P<stage>\S+)\s+checks=(?P<checks>\d+)\s+"
    r"pairs=(?P<pairs>\d+)\s+bad=(?P<bad>\d+)\s+"
    r"last_checked=(?P<last_checked>\d+)\s+last_bad=(?P<last_bad>\d+)\s+"
    r"examples=(?P<examples>.*)$"
)


def parse(path: Path) -> dict[str, dict[str, object]]:
    stages: dict[str, dict[str, object]] = {}
    for line in path.read_text(errors="replace").splitlines():
        match = LINE_RE.search(line)
        if not match:
            continue
        stage = match.group("stage")
        stages[stage] = {
            "checks": int(match.group("checks")),
            "pairs": int(match.group("pairs")),
            "bad": int(match.group("bad")),
            "last_checked": int(match.group("last_checked")),
            "last_bad": int(match.group("last_bad")),
            "examples": match.group("examples"),
        }
    return stages


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("job_logs", type=Path)
    args = parser.parse_args()

    stages = parse(args.job_logs)
    if not stages:
        print("no tree-dup-id probe lines found")
        return 1

    for stage in sorted(stages):
        data = stages[stage]
        pairs = int(data["pairs"])
        bad = int(data["bad"])
        rate = bad / pairs if pairs else 0.0
        print(
            f"{stage}: checks={data['checks']} pairs={pairs} "
            f"bad={bad} bad_rate={rate:.6f} examples={data['examples']}"
        )

    if any(int(data["bad"]) for data in stages.values()):
        print("verdict: FAIL duplicate-row identity")
        return 2
    print("verdict: PASS duplicate-row identity")
    return 0


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