#!/usr/bin/env python3
"""Evaluate MarkItDown conversion quality on a local document corpus.

The goal is not to prove that a converter is universally good. The goal is to
make conversion quality measurable for this knowledge-extraction product:
success rate, latency, information coverage, structure preservation, noise, and
optional similarity to reference Markdown.
"""
from __future__ import annotations

import argparse
import json
import re
import statistics
import sys
import time
from dataclasses import dataclass
from difflib import SequenceMatcher
from pathlib import Path
from typing import Any


SUPPORTED_EXTENSIONS = {
    ".txt",
    ".md",
    ".markdown",
    ".csv",
    ".json",
    ".xml",
    ".html",
    ".htm",
    ".pdf",
    ".docx",
    ".doc",
    ".pptx",
    ".ppt",
    ".xlsx",
    ".xls",
    ".zip",
    ".epub",
    ".jpg",
    ".jpeg",
    ".png",
    ".gif",
    ".wav",
    ".mp3",
}


@dataclass
class Case:
    case_id: str
    file_path: Path
    must_terms: list[str]
    forbidden_terms: list[str]
    expected_headings_min: int
    expected_tables_min: int
    expected_lists_min: int
    expected_links_min: int
    min_chars: int
    gold_markdown_path: Path | None
    notes: str


def load_manifest(path: Path | None) -> dict[str, Any]:
    if not path:
        return {"cases": []}
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict) or not isinstance(data.get("cases"), list):
        raise ValueError("Manifest must be a JSON object with a cases array")
    return data


def discover_cases(input_dir: Path, manifest: dict[str, Any]) -> list[Case]:
    by_file = {
        str(item.get("file", "")).strip(): item
        for item in manifest.get("cases", [])
        if isinstance(item, dict) and str(item.get("file", "")).strip()
    }
    if by_file:
        files = [input_dir / rel for rel in by_file]
    else:
        files = [
            p
            for p in sorted(input_dir.rglob("*"))
            if p.is_file() and p.suffix.lower() in SUPPORTED_EXTENSIONS
        ]

    cases: list[Case] = []
    for file_path in files:
        rel = str(file_path.relative_to(input_dir)) if file_path.is_relative_to(input_dir) else file_path.name
        cfg = by_file.get(rel, {})
        case_id = str(cfg.get("case_id") or file_path.with_suffix("").name)
        gold = cfg.get("gold_markdown")
        cases.append(
            Case(
                case_id=case_id,
                file_path=file_path,
                must_terms=[str(v) for v in cfg.get("must_terms", []) if str(v).strip()],
                forbidden_terms=[str(v) for v in cfg.get("forbidden_terms", []) if str(v).strip()],
                expected_headings_min=int(cfg.get("expected_headings_min") or 0),
                expected_tables_min=int(cfg.get("expected_tables_min") or 0),
                expected_lists_min=int(cfg.get("expected_lists_min") or 0),
                expected_links_min=int(cfg.get("expected_links_min") or 0),
                min_chars=int(cfg.get("min_chars") or 80),
                gold_markdown_path=(input_dir / str(gold)) if gold else None,
                notes=str(cfg.get("notes") or ""),
            )
        )
    return cases


def convert_with_markitdown(file_path: Path) -> str:
    try:
        from markitdown import MarkItDown
    except ImportError as exc:
        raise RuntimeError("MarkItDown is not installed. Run: pip install -r requirements.txt") from exc

    result = MarkItDown(enable_plugins=False).convert(str(file_path))
    text = getattr(result, "text_content", None) or getattr(result, "markdown", None) or ""
    return str(text).strip()


def count_patterns(markdown: str) -> dict[str, int]:
    lines = markdown.splitlines()
    return {
        "chars": len(markdown),
        "lines": len(lines),
        "headings": sum(1 for line in lines if re.match(r"^\s{0,3}#{1,6}\s+\S", line)),
        "table_rows": sum(1 for line in lines if line.count("|") >= 2),
        "list_items": sum(1 for line in lines if re.match(r"^\s*(?:[-*+]|\d+[.)])\s+\S", line)),
        "links": len(re.findall(r"\[[^\]]+\]\([^)]+\)|https?://\S+", markdown)),
        "replacement_chars": markdown.count("\ufffd"),
        "null_chars": markdown.count("\x00"),
        "html_tags": len(re.findall(r"</?[A-Za-z][^>]{0,200}>", markdown)),
        "long_lines": sum(1 for line in lines if len(line) > 500),
    }


def term_coverage(markdown: str, terms: list[str]) -> tuple[float, list[str]]:
    if not terms:
        return 1.0, []
    haystack = markdown.lower()
    missing = [term for term in terms if term.lower() not in haystack]
    return (len(terms) - len(missing)) / len(terms), missing


def forbidden_hits(markdown: str, terms: list[str]) -> list[str]:
    haystack = markdown.lower()
    return [term for term in terms if term.lower() in haystack]


def min_ratio(actual: int, expected: int) -> float:
    if expected <= 0:
        return 1.0
    return min(actual / expected, 1.0)


def score_case(metrics: dict[str, Any]) -> float:
    content_score = min(metrics["chars"] / max(metrics["min_chars"], 1), 1.0)
    structure_score = statistics.mean(
        [
            metrics["heading_score"],
            metrics["table_score"],
            metrics["list_score"],
            metrics["link_score"],
        ]
    )
    noise_penalty = min(
        1.0,
        metrics["replacement_chars"] * 0.08
        + metrics["null_chars"] * 0.2
        + metrics["long_lines"] * 0.03
        + metrics["forbidden_hit_count"] * 0.12,
    )
    gold_similarity = metrics.get("gold_similarity")
    if gold_similarity is None:
        score = (
            0.35 * metrics["must_term_coverage"]
            + 0.25 * content_score
            + 0.25 * structure_score
            + 0.15 * (1.0 - noise_penalty)
        )
    else:
        score = (
            0.30 * metrics["must_term_coverage"]
            + 0.20 * content_score
            + 0.20 * structure_score
            + 0.15 * (1.0 - noise_penalty)
            + 0.15 * gold_similarity
        )
    return round(max(0.0, min(score, 1.0)), 4)


def evaluate_case(case: Case, output_dir: Path) -> dict[str, Any]:
    started = time.perf_counter()
    output_path = output_dir / "converted" / f"{case.case_id}.markitdown.md"
    output_path.parent.mkdir(parents=True, exist_ok=True)
    result: dict[str, Any] = {
        "case_id": case.case_id,
        "file": str(case.file_path),
        "notes": case.notes,
        "success": False,
    }

    try:
        markdown = convert_with_markitdown(case.file_path)
        output_path.write_text(markdown + "\n", encoding="utf-8")
        counts = count_patterns(markdown)
        coverage, missing = term_coverage(markdown, case.must_terms)
        forbidden = forbidden_hits(markdown, case.forbidden_terms)
        metrics: dict[str, Any] = {
            **counts,
            "min_chars": case.min_chars,
            "must_terms": case.must_terms,
            "must_term_coverage": round(coverage, 4),
            "missing_terms": missing,
            "forbidden_terms": case.forbidden_terms,
            "forbidden_hits": forbidden,
            "forbidden_hit_count": len(forbidden),
            "expected_headings_min": case.expected_headings_min,
            "expected_tables_min": case.expected_tables_min,
            "expected_lists_min": case.expected_lists_min,
            "expected_links_min": case.expected_links_min,
            "heading_score": min_ratio(counts["headings"], case.expected_headings_min),
            "table_score": min_ratio(counts["table_rows"], case.expected_tables_min),
            "list_score": min_ratio(counts["list_items"], case.expected_lists_min),
            "link_score": min_ratio(counts["links"], case.expected_links_min),
        }
        if case.gold_markdown_path and case.gold_markdown_path.exists():
            gold = case.gold_markdown_path.read_text(encoding="utf-8")
            metrics["gold_similarity"] = round(SequenceMatcher(None, gold, markdown).ratio(), 4)
        else:
            metrics["gold_similarity"] = None

        result.update(
            {
                "success": True,
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "markdown_path": str(output_path),
                "metrics": metrics,
                "quality_score": score_case(metrics),
            }
        )
    except Exception as exc:  # noqa: BLE001
        result.update(
            {
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "error": str(exc),
                "quality_score": 0.0,
            }
        )
    return result


def aggregate(results: list[dict[str, Any]]) -> dict[str, Any]:
    scores = [float(r.get("quality_score") or 0) for r in results]
    successes = [r for r in results if r.get("success")]
    latencies = [float(r.get("latency_ms") or 0) for r in successes]
    return {
        "cases": len(results),
        "successes": len(successes),
        "failures": len(results) - len(successes),
        "success_rate": round(len(successes) / len(results), 4) if results else 0,
        "mean_quality_score": round(statistics.mean(scores), 4) if scores else 0,
        "median_quality_score": round(statistics.median(scores), 4) if scores else 0,
        "mean_latency_ms": round(statistics.mean(latencies)) if latencies else 0,
    }


def write_markdown_report(report: dict[str, Any], path: Path) -> None:
    summary = report["summary"]
    lines = [
        "# MarkItDown Conversion Evaluation",
        "",
        "## Summary",
        "",
        f"- Cases: {summary['cases']}",
        f"- Success rate: {summary['success_rate']:.2%}",
        f"- Mean quality score: {summary['mean_quality_score']:.3f}",
        f"- Median quality score: {summary['median_quality_score']:.3f}",
        f"- Mean latency: {summary['mean_latency_ms']} ms",
        "",
        "## Cases",
        "",
        "| Case | Success | Score | Chars | Must-term coverage | Missing terms | Latency |",
        "| --- | --- | ---: | ---: | ---: | --- | ---: |",
    ]
    for item in report["results"]:
        metrics = item.get("metrics") or {}
        missing = ", ".join(metrics.get("missing_terms") or [])
        lines.append(
            "| {case} | {success} | {score:.3f} | {chars} | {coverage:.2%} | {missing} | {latency} ms |".format(
                case=item.get("case_id"),
                success="yes" if item.get("success") else "no",
                score=float(item.get("quality_score") or 0),
                chars=metrics.get("chars", 0),
                coverage=float(metrics.get("must_term_coverage") or 0),
                missing=missing.replace("|", "\\|") or "-",
                latency=item.get("latency_ms", 0),
            )
        )
    lines.append("")
    lines.append("## Interpretation")
    lines.append("")
    lines.append("- Score >= 0.85: suitable for normal extraction after spot check.")
    lines.append("- 0.70 <= score < 0.85: usable, but inspect missing terms or structure loss.")
    lines.append("- Score < 0.70: do not trust automatic extraction without fallback/OCR/manual correction.")
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input-dir", default="data/markitdown_eval/input")
    parser.add_argument("--manifest", default="data/markitdown_eval/manifest.json")
    parser.add_argument("--output-dir", default="outputs/markitdown_eval")
    parser.add_argument("--fail-under", type=float, default=0.70)
    parser.add_argument("--allow-missing-manifest", action="store_true")
    args = parser.parse_args()

    input_dir = Path(args.input_dir)
    manifest_path = Path(args.manifest)
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    if not input_dir.exists():
        print(f"Input directory not found: {input_dir}", file=sys.stderr)
        return 2
    if not manifest_path.exists() and not args.allow_missing_manifest:
        print(
            f"Manifest not found: {manifest_path}. Pass --allow-missing-manifest to auto-discover files.",
            file=sys.stderr,
        )
        return 2

    manifest = load_manifest(manifest_path if manifest_path.exists() else None)
    cases = discover_cases(input_dir, manifest)
    if not cases:
        print(f"No supported files found in {input_dir}", file=sys.stderr)
        return 2

    results = [evaluate_case(case, output_dir) for case in cases]
    report = {
        "input_dir": str(input_dir),
        "manifest": str(manifest_path) if manifest_path.exists() else None,
        "summary": aggregate(results),
        "results": results,
    }
    json_path = output_dir / "markitdown_eval_report.json"
    md_path = output_dir / "markitdown_eval_report.md"
    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    write_markdown_report(report, md_path)

    print(f"Wrote {json_path}")
    print(f"Wrote {md_path}")
    mean_score = report["summary"]["mean_quality_score"]
    if mean_score < args.fail_under:
        print(f"Mean quality score {mean_score:.3f} is below fail-under {args.fail_under:.3f}", file=sys.stderr)
        return 1
    return 0


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