#!/usr/bin/env python3
"""
analyze_failure_report.py

Cross-reference CI test failures with test recommendations.

Pipeline:
  1. Scan each log file for failed tests using three methods:
     a. TIMINGS JSON block (machine-readable, from ci_utils.py)
     b. ci_utils.py "✗ FAILED:" summary section (structured text)
     c. pytest "short test summary info" block (for pytest-style logs)
  2. Read recommended_pytest_paths.txt
  3. Match: exact match + file-level match
  4. Generate a Markdown report

Usage:
  python analyze_failure_report.py --log-dir LOG_DIR --recommendations-file RECOMMENDED.txt [--output report.md]
"""

import argparse
import contextlib
import json
import sys
from pathlib import Path

import regex as re

# ============================================================
#  Utility: strip CI log noise
# ============================================================


def strip_ansi(text):
    """Remove ANSI color codes like \x1b[31m, \x1b[0m, etc."""
    return re.sub(r"\x1b\[[0-9;]*m", "", text)


def strip_timestamp(line):
    """Remove GitHub Actions timestamp prefix: YYYY-MM-DDTHH:MM:SS.fffffffZ"""
    return re.sub(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z\s+", "", line)


def clean_line(line):
    """Strip BOM, ANSI codes, and timestamp from one log line."""
    line = line.lstrip("\ufeff")  # UTF-8 BOM marker
    return strip_ansi(strip_timestamp(line)).strip()


# ============================================================
#  Step 1: Extract FAILED and ERROR tests from log files
# ============================================================


# Match pytest-style FAILED/ERROR lines with the test/ prefix (sglang convention).
FAILED_PATTERN = re.compile(r"^(?:FAILED|ERROR)\s+(test/\S+?\.py(?:::\S+?)?)\s")
SUMMARY_SEPARATOR_PATTERN = re.compile(r"^=+\s")
CPU_LOG_PATH_PATTERN = re.compile(r"(?:^|-)cpu-\d+card(?:-|$)", re.IGNORECASE)
CPU_FAILURE_LABEL = "cpu-ut"

# ci_utils.py summary: "✗ FAILED:" section lines like "  /path/to/test/registered/test_xxx.py (exit code 1)".
# Paths are absolute (from os.path.abspath in run_suite.py's glob).
CI_UTILS_FAILED_PATTERN = re.compile(r"^[✗X]\s*FAILED:\s*$")
CI_UTILS_FAILED_LINE_PATTERN = re.compile(r"^\s{2,}(\S+\.py)\s*\(")

# TIMINGS block: machine-readable JSON lines with "passed": false.
TIMINGS_BEGIN_PATTERN = re.compile(r"^=+\s*TIMINGS\s+BEGIN\s*=+")
TIMINGS_END_PATTERN = re.compile(r"^=+\s*TIMINGS\s+END\s*=+")


def _extract_from_timings(lines):
    """Extract failed test file paths from the TIMINGS JSON block (ci_utils.py)."""
    failed = []
    in_timings = False
    for line in lines:
        text = clean_line(line)
        if TIMINGS_BEGIN_PATTERN.search(text):
            in_timings = True
            continue
        if not in_timings:
            continue
        if TIMINGS_END_PATTERN.search(text):
            break
        try:
            entry = json.loads(text)
            if not entry.get("passed", True) and entry.get("file"):
                failed.append(entry["file"])
        except (json.JSONDecodeError, ValueError):
            continue
    return failed


def _extract_from_ci_utils_summary(lines):
    """Extract failed test file paths from ci_utils.py's '✗ FAILED:' summary section."""
    failed = []
    in_failed_section = False
    for line in lines:
        text = clean_line(line)
        if CI_UTILS_FAILED_PATTERN.match(text):
            in_failed_section = True
            continue
        if not in_failed_section:
            continue
        if SUMMARY_SEPARATOR_PATTERN.match(text):
            break
        match = CI_UTILS_FAILED_LINE_PATTERN.match(line)
        if match:
            # Paths in this section are absolute (e.g. "/__w/sglang/sglang/test/registered/test_xxx.py").
            # Strip everything up to and including the "/sglang/" marker to get
            # the repo-relative form (e.g. "test/registered/test_xxx.py").
            path = match.group(1)
            marker = "/sglang/"
            idx = path.rfind(marker)
            if idx >= 0:
                path = path[idx + len(marker) :]
            failed.append(path)
    return failed


def extract_failed_from_log(log_path):
    """Extract failed test paths from one log file.

    Tries three methods in order of reliability:
      1. TIMINGS JSON block (machine-readable, ci_utils.py)
      2. ci_utils.py '✗ FAILED:' summary section (human-readable but structured)
      3. pytest 'short test summary info' block (for pytest-style logs)
    """
    try:
        lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()
    except Exception as exc:
        print(f"::warning:: Cannot read {log_path}: {exc}")
        return []

    # Method 1: TIMINGS block (most reliable, machine-readable).
    failed = _extract_from_timings(lines)
    if failed:
        return failed

    # Method 2: ci_utils.py summary section.
    failed = _extract_from_ci_utils_summary(lines)
    if failed:
        return failed

    # Method 3: pytest-style "short test summary info" block.
    failed = []
    in_summary = False
    for line in lines:
        text = clean_line(line)

        if "short test summary info" in text:
            in_summary = True
            continue

        if not in_summary:
            continue

        if SUMMARY_SEPARATOR_PATTERN.match(text):
            in_summary = False
            continue

        match = FAILED_PATTERN.match(text)
        if match:
            failed.append(match.group(1))

    return failed


def is_cpu_log(log_path):
    """Return whether a log belongs to a CPU selected-test artifact."""
    if log_path.stem.lower().endswith("-cpu-ut"):
        return True

    return any(CPU_LOG_PATH_PATTERN.search(part) for part in log_path.parent.parts)


def extract_failed_from_logs(log_dir):
    """
    Scan CPU logs first and represent all CPU failures as one ``cpu-ut`` item.
    Scan all remaining logs with the existing pytest node-ID behavior.
    """
    base = Path(log_dir)
    if not base.is_dir():
        print(f"::warning:: Log directory not found: {log_dir}")
        return []

    # Scan .log files (from NPU test stages) and .txt files (legacy/mock).
    candidates = []
    candidates.extend(base.rglob("*.log"))
    candidates.extend(base.rglob("*.txt"))
    candidates = [
        candidate
        for candidate in sorted(candidates)
        if candidate.suffix != ".txt" or "run-selected-tests" in candidate.name
    ]

    cpu_candidates = []
    regular_candidates = []
    for candidate in candidates:
        target = cpu_candidates if is_cpu_log(candidate) else regular_candidates
        target.append(candidate)

    all_failed = []
    seen = set()

    cpu_failed = False
    for candidate in cpu_candidates:
        if extract_failed_from_log(candidate):
            cpu_failed = True

    if cpu_failed:
        seen.add(CPU_FAILURE_LABEL)
        all_failed.append(CPU_FAILURE_LABEL)

    for candidate in regular_candidates:
        for test_path in extract_failed_from_log(candidate):
            if test_path not in seen:
                seen.add(test_path)
                all_failed.append(test_path)

    return all_failed


# ============================================================
#  Step 2: Read recommendations
# ============================================================


def read_recommended(recommendations_file):
    """
    recommended_pytest_paths.txt contains one pytest path per line, e.g.:
        test/ops/test_matmul.py::test_bf16
        test/layers/test_attention.py
    """
    path = Path(recommendations_file)
    if not path.exists():
        print(f"::warning:: Recommendations file not found: {recommendations_file}")
        return []
    raw = path.read_text(encoding="utf-8").lstrip("\ufeff")
    return [
        line.strip()
        for line in raw.splitlines()
        if line.strip() and not line.startswith("ERROR")
    ]


# ============================================================
#  Step 3: Match
# ============================================================


def normalize_test_path(test_path):
    """Return a stable comparison key for a pytest path or node ID."""
    normalized = test_path.strip().replace("\\", "/").removeprefix("./")
    file_path, separator, test_name = normalized.partition("::")
    file_path = file_path.removesuffix(".py")
    if separator:
        test_name = test_name.partition("[")[0]
    return f"{file_path}{separator}{test_name}" if separator else file_path


def match_failed_vs_recommended(failed, recommended):
    """
    Two-level matching:
      Level 1 - File-level: recommended "test/foo.py" (no function)
                  matches failed "test/foo.py::anything"
      Level 2 - Exact: "test/foo.py::test_bar" in both lists

    Returns {"hit": [...], "miss": [...], "untested": [...]}
      hit:       failed AND recommended
      miss:      failed but NOT recommended
      untested:  recommended but NOT in failed list
    """
    recommended_files = {
        normalize_test_path(item) for item in recommended if "::" not in item
    }
    recommended_functions = {
        normalize_test_path(item) for item in recommended if "::" in item
    }

    hit = []
    miss = []

    normalized_failed = {item: normalize_test_path(item) for item in failed}
    for original, normalized in normalized_failed.items():
        failed_file = normalized.split("::", 1)[0]
        if failed_file in recommended_files or normalized in recommended_functions:
            hit.append(original)
        else:
            miss.append(original)

    # Recommended but not failed
    failed_functions = set(normalized_failed.values())
    failed_files = {item.split("::", 1)[0] for item in failed_functions}
    untested = []
    for item in recommended:
        normalized = normalize_test_path(item)
        has_failure = (
            normalized in failed_functions
            if "::" in item
            else normalized in failed_files
        )
        if not has_failure:
            untested.append(item)

    return {"hit": hit, "miss": miss, "untested": untested}


# ============================================================
#  Step 4: Generate Markdown report
# ============================================================


def generate_report(
    failed, recommended, matched, log_dir, recommendations_source="none"
):
    """Produce a Markdown summary table."""
    hit = matched["hit"]
    miss = matched["miss"]
    untested = matched["untested"]

    out = []
    out.append("# Test Failure vs Recommendation Report")
    out.append("")
    out.append(f"**Log source**: `{log_dir}`")
    out.append("")

    # Recommendation source indicator
    if recommendations_source == "output":
        out.append(
            "> **[Source: Workflow Output]** Recommended cases are passed from coverage recommendations outputs"
        )
    elif recommendations_source == "committed":
        out.append(
            "> **[Source: Local File]** Recommended test cases come from a txt file in the repository"
        )
    else:
        out.append("> **[Source: None]** No recommended test cases found")
    out.append("")

    # ================================================================
    #  Section 1: Full Failed Test List
    # ================================================================
    out.append("---")
    out.append("")
    out.append(f"## Failed Test Cases（ {len(failed)} total）")
    out.append("")
    if failed:
        for i, t in enumerate(failed, 1):
            tag = (
                " **[Matched Recommendation]**"
                if t in hit
                else " **[Not Matched Recommendation]**"
            )
            out.append(f"{i}. `{t}`{tag}")
        out.append("")
    else:
        out.append("> No failed test cases")
        out.append("")

    # ================================================================
    #  Section 2: Full Recommended Test List
    # ================================================================
    out.append("---")
    out.append("")
    out.append(f"## Recommended Test Cases（ {len(recommended)} total）")
    out.append("")
    if recommended:
        normalized_failed = {normalize_test_path(item) for item in failed}
        failed_file_set = {item.split("::", 1)[0] for item in normalized_failed}
        for i, item in enumerate(recommended, 1):
            normalized = normalize_test_path(item)
            has_failure = (
                normalized in normalized_failed
                if "::" in item
                else normalized in failed_file_set
            )
            tag = " **[Already Failed]**" if has_failure else ""
            out.append(f"{i}. `{item}`{tag}")
        out.append("")
    else:
        out.append("> No recommended test cases")
        out.append("")

    # ================================================================
    #  Section 3: Core Conclusion
    # ================================================================
    out.append("---")
    out.append("")
    out.append("## Core Conclusion")
    out.append("")
    if not failed:
        out.append(
            "> No failed cases in this CI run; no need to compare against the recommendation list."
        )
    elif len(miss) == 0:
        out.append("> **All failed test cases are within the recommended scope.**")
    else:
        total_failed = len(failed)
        out.append(
            f"> ** {len(miss)}/{total_failed} failed cases are outside the recommended scope.**"
        )
    out.append("")

    # ================================================================
    #  Section 4: Detail table
    # ================================================================
    out.append("| Category | Count |")
    out.append("|---|---|")
    out.append(f"| Failed & Matched Recommendation | {len(hit)} |")
    out.append(f"| Failed but Not Matched Recommendation | {len(miss)} |")
    out.append(f"| Recommended but Not Failed | {len(untested)} |")
    out.append("")

    if hit:
        out.append("## Failed & Matched Recommendation")
        out.append("")
        out.append("| # | Failed test |")
        out.append("|---|---|")
        for i, t in enumerate(hit, 1):
            out.append(f"| {i} | `{t}` |")
        out.append("")

    if miss:
        out.append("## Failed but Not Matched Recommendation")
        out.append("")
        out.append(
            "> Possible causes: uncovered modules, environment issues, flaky tests."
        )
        out.append("")
        for t in miss:
            out.append(f"- `{t}`")
        out.append("")

    if untested:
        out.append("## Recommended but Not Failed")
        out.append("")
        out.append(
            "> These test cases were recommended but did not fail this run (passed or not executed)."
        )
        out.append("")
        for t in untested:
            out.append(f"- `{t}`")
        out.append("")

    if not hit and not miss:
        out.append("## No failed cases")
        out.append("")

    out.append("---")
    out.append("*Generated by analyze_failure_report.py*")
    return "\n".join(out)


def main():
    parser = argparse.ArgumentParser(
        description="Cross-reference CI test failures with test recommendations"
    )
    parser.add_argument(
        "--log-dir", required=True, help="Directory containing CI .log files"
    )
    parser.add_argument(
        "--recommendations-file",
        help="Path to recommended_pytest_paths.txt",
    )
    parser.add_argument(
        "--output",
        default="failure_report.md",
        help="Output Markdown report path (default: failure_report.md)",
    )
    parser.add_argument(
        "--recommendations-source",
        default="none",
        choices=["committed", "output", "none"],
        help="Where recommendations came from",
    )
    args = parser.parse_args()

    if not args.recommendations_file:
        parser.error("--recommendations-file is required")

    # For Windows console: force UTF-8 if possible
    if sys.platform == "win32":
        with contextlib.suppress(Exception):
            sys.stdout.reconfigure(encoding="utf-8", errors="replace")

    print("=" * 50)
    print("Step 1: Extract failed tests from CI logs")
    print("=" * 50)
    failed = extract_failed_from_logs(args.log_dir)
    print(f"Failed: {len(failed)}")

    print()
    print("=" * 50)
    print("Step 2: Read recommendations")
    print("=" * 50)
    recommended = read_recommended(args.recommendations_file)
    print(f"Recommended: {len(recommended)}")

    print()
    print("=" * 50)
    print("Step 3: Match")
    print("=" * 50)
    matched = match_failed_vs_recommended(failed, recommended)
    print(f"Hit (failed + recommended): {len(matched['hit'])}")
    print(f"Miss (failed, not recommended): {len(matched['miss'])}")
    print(f"Untested (recommended, no failure): {len(matched['untested'])}")

    report = generate_report(
        failed, recommended, matched, args.log_dir, args.recommendations_source
    )
    output_path = Path(args.output)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(report, encoding="utf-8")
    print()
    print(f"Report => {output_path}")
    print()

    # Print report to stdout (safe fallback for Windows encoding)
    try:
        print(report)
    except UnicodeEncodeError:
        print(report.encode("ascii", errors="replace").decode("ascii"))


if __name__ == "__main__":
    main()