CustomNPC-Plus/tools/migration/batch_migration_diff.py

"""
batch_migration_diff.py — Batch Migration Diff
================================================

Runs migration-diff logic on ALL migrated files in core/ and produces
one unified JSON with every file's verdict (PASS/WARN/FAIL/NO_COUNTERPART).

Usage:
    python tools/migration/batch_migration_diff.py
    python tools/migration/batch_migration_diff.py --json
    python tools/migration/batch_migration_diff.py --output tools/migration/_batch_diffs.json
    python tools/migration/batch_migration_diff.py --mc-root src/main/java --core-root core/src/main/java
    python tools/migration/batch_migration_diff.py --classifications tools/migration/mc_type_classifications_original.json
    python tools/migration/batch_migration_diff.py --project-root /path/to/project
"""

import sys
import os
import json
import argparse
import difflib
import re
from pathlib import Path
from datetime import datetime, timezone

# Add script directory to path so we can import analyze.py and pa_rules.py
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))


def auto_detect_project_root():
    """Auto-detect project root from script location (tools/migration/ -> root)."""
    return SCRIPT_DIR.parent.parent


def _try_import_analyze():
    """Try to import functions from analyze.py. Returns a dict of functions or None."""
    try:
        from analyze import (
            setup_treesitter_parser,
            _load_classifications,
            _extract_method_bodies,
            _normalized_signature_key,
            normalize_text,
            _scan_unmigrated_types,
        )

        return {
            "setup_treesitter_parser": setup_treesitter_parser,
            "_load_classifications": _load_classifications,
            "_extract_method_bodies": _extract_method_bodies,
            "_normalized_signature_key": _normalized_signature_key,
            "normalize_text": normalize_text,
            "_scan_unmigrated_types": _scan_unmigrated_types,
        }
    except ImportError as e:
        print(f"WARNING: Could not import from analyze.py: {e}", file=sys.stderr)
        return None


def _normalize_classifications(data):
    """Normalize both mc_type_mappings.json and mc_type_classifications_original.json formats."""
    types = data.get("types") or data.get("mappings") or {}
    normalized = {}
    for type_name, info in types.items():
        if type_name.startswith("_"):
            normalized[type_name] = info
            continue
        entry = dict(info)
        if "pa_name" in entry and "pa" not in entry:
            entry["pa"] = entry["pa_name"]
        if "pa_package" in entry and "package" not in entry:
            entry["package"] = entry["pa_package"]
        if "service_name" in entry and "service" not in entry:
            entry["service"] = entry["service_name"]
        normalized[type_name] = entry
    return normalized


def _load_classifications_fallback(project_root, classifications_path):
    cls_file = Path(project_root) / classifications_path
    if not cls_file.exists():
        raise FileNotFoundError(f"Classifications file not found: {cls_file}")
    with open(cls_file, "r", encoding="utf-8") as f:
        data = json.load(f)
    return _normalize_classifications(data)


def _scan_unmigrated_types_fallback(file_text, classifications):
    """Scan file text for MC type names that should have been replaced. Standalone fallback."""
    results = []
    lines = file_text.split("\n")
    for type_name, info in classifications.items():
        if type_name.startswith("_"):
            continue
        category = info.get("category", "")
        if category not in ("EXISTING", "INTERFACE", "SERVICE"):
            continue
        pa_name = info.get("pa") or info.get("service")
        if not pa_name:
            continue
        if len(type_name) < 4:
            continue
        pattern = re.compile(r"\b" + re.escape(type_name) + r"\b")
        found_lines = []
        has_import = False
        for i, line in enumerate(lines, 1):
            if pattern.search(line):
                if line.strip().startswith("import "):
                    has_import = True
                found_lines.append(i)
        if found_lines:
            results.append(
                {
                    "mc_name": type_name,
                    "pa_name": pa_name,
                    "lines": found_lines,
                    "in_import": has_import,
                }
            )
    return results


def _resolve_classifications_path(args):
    """Resolve which classifications file to use."""
    if args.classifications:
        return args.classifications
    # Prefer mc_type_mappings.json if it exists
    project_root = (
        Path(args.project_root) if args.project_root else auto_detect_project_root()
    )
    mappings_path = project_root / "tools" / "migration" / "mc_type_mappings.json"
    if mappings_path.exists():
        return "tools/migration/mc_type_mappings.json"
    # Fallback to original classifications
    original_path = (
        project_root / "tools" / "migration" / "mc_type_classifications_original.json"
    )
    if original_path.exists():
        return "tools/migration/mc_type_classifications_original.json"
    # Default
    return "tools/migration/mc_type_classifications.json"


def run_single_diff(old_file, new_file, parser, classifications, analyze_fns):
    """
    Run migration-diff logic on a single file pair.
    Returns a result dict with verdict, unmigrated, missing_methods, semantic_body_diffs.
    """
    result = {
        "unmigrated": [],
        "missing_methods": [],
        "semantic_body_diffs": [],
    }

    # Step 1: Scan for unmigrated types in new file
    new_text = new_file.read_text(encoding="utf-8", errors="replace")
    if analyze_fns:
        unmigrated = analyze_fns["_scan_unmigrated_types"](new_text, classifications)
    else:
        unmigrated = _scan_unmigrated_types_fallback(new_text, classifications)
    result["unmigrated"] = unmigrated

    # Step 2 & 3: Method-level analysis (requires tree-sitter)
    if parser is not None and analyze_fns:
        try:
            _extract = analyze_fns["_extract_method_bodies"]
            _norm_key = analyze_fns["_normalized_signature_key"]
            _norm_text = analyze_fns["normalize_text"]

            old_methods, _ = _extract(old_file, parser)
            new_methods, _ = _extract(new_file, parser)

            # Build maps by normalized key
            old_method_map = {}
            for m in old_methods:
                key = _norm_key(m, classifications)
                old_method_map[key] = m

            new_method_map = {}
            for m in new_methods:
                key = _norm_key(m, classifications)
                new_method_map[key] = m

            old_keys = set(old_method_map.keys())
            new_keys = set(new_method_map.keys())

            # Missing methods
            for key in sorted(old_keys - new_keys):
                m = old_method_map[key]
                result["missing_methods"].append(
                    {
                        "name": m["name"],
                        "signature": m["signature_text"],
                        "line": m["line"],
                    }
                )

            # Semantic body diffs
            common_keys = old_keys & new_keys
            for key in sorted(common_keys):
                old_m = old_method_map[key]
                new_m = new_method_map[key]

                old_body = old_m.get("body")
                new_body = new_m.get("body")
                if old_body is None or new_body is None:
                    continue

                norm_old = _norm_text(old_body, classifications)
                norm_new = _norm_text(new_body, classifications)

                norm_old_lines = norm_old.strip().splitlines()
                norm_new_lines = norm_new.strip().splitlines()

                matcher = difflib.SequenceMatcher(None, norm_old_lines, norm_new_lines)
                ratio = matcher.ratio()

                diff_lines = list(
                    difflib.unified_diff(
                        norm_old_lines,
                        norm_new_lines,
                        fromfile="old (normalized)",
                        tofile="new (normalized)",
                        lineterm="",
                    )
                )
                change_lines = [
                    l
                    for l in diff_lines
                    if (l.startswith("+") or l.startswith("-"))
                    and not l.startswith("+++")
                    and not l.startswith("---")
                ]

                if ratio < 0.85 or len(change_lines) > 0:
                    result["semantic_body_diffs"].append(
                        {
                            "name": old_m["name"],
                            "signature": old_m["signature_text"],
                            "similarity_ratio": round(ratio, 3),
                            "diff_line_count": len(change_lines),
                            "diff_preview": diff_lines[:10],
                        }
                    )
        except Exception as e:
            result["note"] = f"tree-sitter analysis error: {str(e)}"
    else:
        result["note"] = "tree-sitter not available"

    # Verdict
    if result["missing_methods"]:
        result["verdict"] = "FAIL"
        result["verdict_reason"] = f"{len(result['missing_methods'])} missing method(s)"
    elif result["unmigrated"] or result["semantic_body_diffs"]:
        parts = []
        if result["unmigrated"]:
            parts.append(f"{len(result['unmigrated'])} unmigrated type(s)")
        if result["semantic_body_diffs"]:
            parts.append(f"{len(result['semantic_body_diffs'])} body diff(s)")
        result["verdict"] = "WARN"
        result["verdict_reason"] = ", ".join(parts)
    else:
        result["verdict"] = "PASS"
        result["verdict_reason"] = (
            "all methods present, no unmigrated types, no body drift"
        )

    return result


def main():
    parser = argparse.ArgumentParser(
        description="Batch migration-diff: run migration-diff on ALL core/ files"
    )
    parser.add_argument(
        "--mc-root",
        default="src/main/java",
        help="MC 1.7.10 source root (default: src/main/java)",
    )
    parser.add_argument(
        "--core-root",
        default="core/src/main/java",
        help="Core source root (default: core/src/main/java)",
    )
    parser.add_argument(
        "--classifications",
        help="Path to classifications JSON (relative to project root)",
    )
    parser.add_argument(
        "--output",
        default="tools/migration/_batch_diffs.json",
        help="Output JSON file path (relative to project root)",
    )
    parser.add_argument(
        "--json",
        dest="json_output",
        action="store_true",
        help="Print JSON to stdout instead of human-readable output",
    )
    parser.add_argument(
        "--project-root",
        help="Project root (auto-detected if omitted)",
    )

    args = parser.parse_args()

    project_root = (
        Path(args.project_root) if args.project_root else auto_detect_project_root()
    )

    # Resolve classifications path
    classifications_path = _resolve_classifications_path(args)

    # Try to import from analyze.py
    analyze_fns = _try_import_analyze()

    # Load classifications
    if analyze_fns:
        try:
            classifications = analyze_fns["_load_classifications"](
                project_root, classifications_path
            )
            if not classifications:
                classifications = _load_classifications_fallback(
                    project_root, classifications_path
                )
        except (FileNotFoundError, KeyError):
            classifications = _load_classifications_fallback(
                project_root, classifications_path
            )
    else:
        classifications = _load_classifications_fallback(
            project_root, classifications_path
        )

    # Setup tree-sitter parser
    ts_parser = None
    if analyze_fns:
        try:
            ts_parser, _ = analyze_fns["setup_treesitter_parser"]()
        except Exception:
            pass

    # Enumerate all .java files in core-root
    core_root = project_root / args.core_root
    mc_root = project_root / args.mc_root

    if not core_root.exists():
        print(f"ERROR: Core root not found: {core_root}", file=sys.stderr)
        return 1

    core_files = sorted(core_root.rglob("*.java"))
    total_files = len(core_files)

    results = []
    summary = {"PASS": [], "WARN": [], "FAIL": [], "NO_COUNTERPART": []}
    files_with_counterpart = 0
    files_without_counterpart = 0

    for core_file in core_files:
        rel_path = core_file.relative_to(core_root)
        # Normalize to forward slashes for consistency
        rel_path_str = str(rel_path).replace("\\", "/")

        mc_file = mc_root / rel_path

        if mc_file.exists():
            files_with_counterpart += 1
            diff_result = run_single_diff(
                mc_file, core_file, ts_parser, classifications, analyze_fns
            )

            entry = {
                "relative_path": rel_path_str,
                "old_file": str(Path(args.mc_root) / rel_path).replace("\\", "/"),
                "new_file": str(Path(args.core_root) / rel_path).replace("\\", "/"),
                "verdict": diff_result["verdict"],
                "verdict_reason": diff_result["verdict_reason"],
                "unmigrated_count": len(diff_result["unmigrated"]),
                "missing_methods_count": len(diff_result["missing_methods"]),
                "semantic_diffs_count": len(diff_result["semantic_body_diffs"]),
                "unmigrated": diff_result["unmigrated"],
                "missing_methods": diff_result["missing_methods"],
                "semantic_body_diffs": diff_result["semantic_body_diffs"],
            }
            if "note" in diff_result:
                entry["note"] = diff_result["note"]

            results.append(entry)
            summary[diff_result["verdict"]].append(rel_path_str)
        else:
            files_without_counterpart += 1
            entry = {
                "relative_path": rel_path_str,
                "verdict": "NO_COUNTERPART",
                "verdict_reason": "No mc1710 original found",
                "new_file": str(Path(args.core_root) / rel_path).replace("\\", "/"),
            }
            results.append(entry)
            summary["NO_COUNTERPART"].append(rel_path_str)

    # Build output
    output = {
        "_meta": {
            "generated": datetime.now(timezone.utc).isoformat(),
            "mc_root": args.mc_root,
            "core_root": args.core_root,
            "classifications_source": classifications_path,
            "total_files_in_core": total_files,
            "files_with_counterpart": files_with_counterpart,
            "files_without_counterpart": files_without_counterpart,
            "passed": len(summary["PASS"]),
            "warned": len(summary["WARN"]),
            "failed": len(summary["FAIL"]),
        },
        "results": results,
        "summary": summary,
    }

    if args.json_output:
        json.dump(output, sys.stdout, indent=2)
        print()
        return 0

    # Human-readable output
    print()
    print(f"Batch Migration Diff — {total_files} core files analyzed")
    print(f"  With counterpart:    {files_with_counterpart}")
    print(f"  Without counterpart: {files_without_counterpart}")
    print()
    print(f"  PASS: {len(summary['PASS']):>4}")
    print(f"  WARN: {len(summary['WARN']):>4}")
    print(f"  FAIL: {len(summary['FAIL']):>4}")
    print()

    if summary["FAIL"]:
        print("FAILED FILES:")
        for rel in summary["FAIL"]:
            entry = next(r for r in results if r["relative_path"] == rel)
            print(f"  {rel}")
            if entry.get("missing_methods"):
                methods = ", ".join(m["name"] for m in entry["missing_methods"])
                print(
                    f"    - {len(entry['missing_methods'])} missing methods: {methods}"
                )
            if entry.get("unmigrated"):
                types = ", ".join(u["mc_name"] for u in entry["unmigrated"])
                print(f"    - {len(entry['unmigrated'])} unmigrated types: {types}")
        print()

    if summary["WARN"]:
        shown = min(5, len(summary["WARN"]))
        print(f"WARNED FILES (showing first {shown}):")
        for rel in summary["WARN"][:shown]:
            entry = next(r for r in results if r["relative_path"] == rel)
            parts = []
            if entry.get("semantic_body_diffs"):
                sims = [
                    str(round(d["similarity_ratio"] * 100)) + "%"
                    for d in entry["semantic_body_diffs"]
                ]
                parts.append(
                    f"{len(entry['semantic_body_diffs'])} body diffs (similarity: {', '.join(sims)})"
                )
            if entry.get("unmigrated"):
                parts.append(f"{len(entry['unmigrated'])} unmigrated types")
            print(f"  {rel}")
            for part in parts:
                print(f"    ~ {part}")
        if len(summary["WARN"]) > shown:
            print(f"  ... and {len(summary['WARN']) - shown} more")
        print()

    # Write JSON output file
    output_path = project_root / args.output
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(output, f, indent=2)
    print(f"Results written to: {args.output}")
    print()

    return 0


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