scripts/22_eval300.py

"""
평가셋 300개 기반 검색엔진 품질 평가
=====================================
실행:
    .venv/Scripts/python.exe scripts/22_eval300.py
    .venv/Scripts/python.exe scripts/22_eval300.py --no-hyde
    .venv/Scripts/python.exe scripts/22_eval300.py --save-xlsx  # 결과 엑셀 저장
"""
import sys
import re
import json
import argparse
from pathlib import Path
from collections import defaultdict

sys.path.insert(0, str(Path(__file__).parent.parent))

from core.search_engine import MenuSearchEngine

EVAL_PATH    = Path(__file__).parent.parent / "data" / "eval_queries_300.json"
RESULT_PATH  = Path(__file__).parent.parent / "data" / "eval_results_300.json"
XLSX_PATH    = Path(__file__).parent.parent / "data" / "eval_results_300.xlsx"


def normalize_path(path: str) -> str:
    if not path:
        return ""
    path = re.sub(r"\s*>\s*", ">", str(path))
    path = re.sub(r"\(.*?\)", "", path)
    return path.strip().lower()


def matches(pred: str, ans: str) -> bool:
    if not pred or not ans:
        return False
    if pred == ans:
        return True
    pred_end = pred.split(">")[-1].strip()
    ans_end  = ans.split(">")[-1].strip()
    if pred_end and ans_end and pred_end == ans_end:
        pp = pred.split(">")
        ap = ans.split(">")
        return sum(1 for p in pp if p in ap) >= min(2, len(ap))
    return False


def save_xlsx(detail_rows: list[dict], summary: dict, xlsx_path: Path):
    """평가 결과를 엑셀로 저장 (상세 + 요약 시트)"""
    import pandas as pd
    from openpyxl import load_workbook
    from openpyxl.styles import PatternFill, Font, Alignment

    df = pd.DataFrame(detail_rows)
    df = df[["id", "rank", "source", "결과", "menu_id", "menu_name",
             "query", "정답경로", "예측1위", "예측2위", "예측3위", "예측4위", "예측5위", "mrr"]]

    with pd.ExcelWriter(xlsx_path, engine="openpyxl") as writer:
        df.to_excel(writer, index=False, sheet_name="상세결과")

        # 요약 시트
        rows_summary = [
            ["지표", "값"],
            ["평가셋 (n)", summary["n"]],
            ["HyDE", summary["hyde"]],
            ["Acc@1", f"{summary['acc1']:.1%}"],
            ["Acc@3", f"{summary['acc3']:.1%}"],
            ["Acc@5", f"{summary['acc5']:.1%}"],
            ["MRR@5", f"{summary['mrr']:.3f}"],
            [],
            ["[소스별]", "Acc@1", "Acc@3", "Acc@5", "n"],
        ]
        for src_row in summary.get("by_source", []):
            rows_summary.append([src_row["source"], f"{src_row['acc1']:.1%}",
                                  f"{src_row['acc3']:.1%}", f"{src_row['acc5']:.1%}", src_row["n"]])
        rows_summary.append([])
        rows_summary.append(["[순위구간별]", "Acc@1", "Acc@5", "n"])
        for rk_row in summary.get("by_rank", []):
            rows_summary.append([rk_row["group"], f"{rk_row['acc1']:.1%}",
                                  f"{rk_row['acc5']:.1%}", rk_row["n"]])

        ws_s = writer.book.create_sheet("요약")
        for r in rows_summary:
            ws_s.append(r)

        # 상세 시트 서식
        ws = writer.sheets["상세결과"]
        col_widths = [5, 8, 10, 7, 14, 20, 38, 42, 42, 42, 42, 42, 42, 8]
        for i, w in enumerate(col_widths, 1):
            from openpyxl.utils import get_column_letter
            ws.column_dimensions[get_column_letter(i)].width = w

        green = PatternFill("solid", start_color="C6EFCE")
        yellow = PatternFill("solid", start_color="FFEB9C")
        red   = PatternFill("solid", start_color="FFC7CE")

        for row in ws.iter_rows(min_row=2, max_row=ws.max_row):
            label = row[3].value  # 결과 컬럼
            fill = green if label == "O" else (yellow if label == "~" else red)
            row[3].fill = fill

    print(f"\n엑셀 저장: {xlsx_path}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--no-hyde",    action="store_true", help="HyDE 비활성화")
    parser.add_argument("--save-xlsx",  action="store_true", help="결과 엑셀 저장")
    args = parser.parse_args()
    use_hyde = not args.no_hyde

    with open(EVAL_PATH, encoding="utf-8") as f:
        eval_set = json.load(f)

    print(f"평가셋: {len(eval_set)}개  |  HyDE: {'ON' if use_hyde else 'OFF'}")
    print("검색엔진 초기화 중...")
    engine = MenuSearchEngine.get_instance()
    print("준비 완료\n")

    acc1, acc3, acc5, mrr_list = [], [], [], []
    wrong_cases = []
    detail_rows  = []
    by_source = defaultdict(lambda: {"acc1": [], "acc3": [], "acc5": []})

    for item in eval_set:
        query     = item["query"]
        menu_path = item["menu_path"]
        source    = item["source"]

        hits  = engine.search(query, top_n=5, threshold=0.0, use_hyde=use_hyde)
        paths = [normalize_path(h.get("menu_path", "")) for h in hits]
        names = [h.get("menu_name", "") for h in hits]
        ans   = normalize_path(menu_path)

        hit1 = matches(paths[0], ans) if paths else False
        hit3 = any(matches(p, ans) for p in paths[:3])
        hit5 = any(matches(p, ans) for p in paths[:5])

        # MRR
        rr = 0.0
        for k, p in enumerate(paths[:5], 1):
            if matches(p, ans):
                rr = 1.0 / k
                break
        mrr_list.append(rr)

        acc1.append(hit1)
        acc3.append(hit3)
        acc5.append(hit5)
        by_source[source]["acc1"].append(hit1)
        by_source[source]["acc3"].append(hit3)
        by_source[source]["acc5"].append(hit5)

        label = "O" if hit1 else ("~" if hit3 else "X")
        print(f"  [{label}] [{item['id']:3d}] {query[:30]:<30} -> {paths[0] if paths else '(없음)'}")

        if not hit5:
            wrong_cases.append({"id": item["id"], "query": query,
                                 "expected": menu_path, "got": paths[0] if paths else ""})

        # 상세 행
        pad = lambda lst, n: lst + [""] * (n - len(lst))
        p5 = pad(paths, 5)
        detail_rows.append({
            "id": item["id"], "rank": item["rank"], "source": source,
            "결과": label,
            "menu_id": item["menu_id"], "menu_name": item["menu_name"],
            "query": query,
            "정답경로": menu_path,
            "예측1위": p5[0], "예측2위": p5[1], "예측3위": p5[2],
            "예측4위": p5[3], "예측5위": p5[4],
            "mrr": round(rr, 4),
        })

    n = len(acc1)
    print(f"\n{'='*62}")
    print(f"== 평가 결과 (n={n})  HyDE={'ON' if use_hyde else 'OFF'} ==")
    print(f"{'='*62}")
    print(f"  Acc@1  : {sum(acc1)}/{n} = {sum(acc1)/n:.1%}")
    print(f"  Acc@3  : {sum(acc3)}/{n} = {sum(acc3)/n:.1%}")
    print(f"  Acc@5  : {sum(acc5)}/{n} = {sum(acc5)/n:.1%}")
    print(f"  MRR@5  : {sum(mrr_list)/n:.3f}")
    print(f"{'='*62}")

    # 소스별 분석
    print(f"\n[소스별]")
    src_rows = []
    for src in ["human", "llm"]:
        d = by_source[src]
        if not d["acc1"]:
            continue
        ns = len(d["acc1"])
        print(f"  {src:8s} (n={ns:3d})  Acc@1={sum(d['acc1'])/ns:.1%}  "
              f"Acc@3={sum(d['acc3'])/ns:.1%}  Acc@5={sum(d['acc5'])/ns:.1%}")
        src_rows.append({"source": src, "n": ns,
                          "acc1": sum(d["acc1"])/ns,
                          "acc3": sum(d["acc3"])/ns,
                          "acc5": sum(d["acc5"])/ns})

    # top200 vs 201+ 분석
    by_rank = defaultdict(lambda: {"acc1": [], "acc5": []})
    for item, a1, a5 in zip(eval_set, acc1, acc5):
        grp = "top200" if item["rank"] <= 200 else "201+"
        by_rank[grp]["acc1"].append(a1)
        by_rank[grp]["acc5"].append(a5)

    print(f"\n[순위 구간별]")
    rk_rows = []
    for grp in ["top200", "201+"]:
        d = by_rank[grp]
        if not d["acc1"]:
            continue
        ng = len(d["acc1"])
        print(f"  {grp:8s} (n={ng:3d})  Acc@1={sum(d['acc1'])/ng:.1%}  Acc@5={sum(d['acc5'])/ng:.1%}")
        rk_rows.append({"group": grp, "n": ng,
                         "acc1": sum(d["acc1"])/ng,
                         "acc5": sum(d["acc5"])/ng})

    # Top5 완전 오답
    print(f"\n[Top5 오답 {len(wrong_cases)}개]")
    for w in wrong_cases[:10]:
        print(f"  [{w['id']:3d}] {w['query'][:28]:<28}")
        print(f"        정답: {w['expected']}")
        print(f"        예측: {w['got']}")

    # 결과 저장
    summary = {
        "n": n, "hyde": "ON" if use_hyde else "OFF",
        "acc1": sum(acc1)/n, "acc3": sum(acc3)/n,
        "acc5": sum(acc5)/n, "mrr": sum(mrr_list)/n,
        "by_source": src_rows, "by_rank": rk_rows,
    }
    with open(RESULT_PATH, "w", encoding="utf-8") as f:
        json.dump({"summary": summary, "detail": detail_rows}, f, ensure_ascii=False, indent=2)
    print(f"\n결과 JSON 저장: {RESULT_PATH}")

    if args.save_xlsx:
        save_xlsx(detail_rows, summary, XLSX_PATH)


if __name__ == "__main__":
    main()