scripts/eval_export.py

"""
검색 품질 평가 결과 → 엑셀 출력 (자체평가 이유 포함)
실행: python scripts/eval_export.py
"""
import sys, json
from pathlib import Path
from collections import defaultdict
from datetime import datetime

sys.path.insert(0, str(Path(__file__).parent.parent))
sys.stdout.reconfigure(encoding='utf-8')

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter

from core.search_engine import MenuSearchEngine

EVAL_PATH = Path(__file__).parent.parent / "data" / "eval_queries.json"
TOP_N = 10   # 정답 메뉴 Dense/BM25 정보 수집을 위해 10개 검색

# ── 색상 ─────────────────────────────────────────────────────────────────────
C_HEADER = "1F4E79"
C_SUB    = "2E75B6"
C_GREEN  = "E2EFDA"
C_YELLOW = "FFF2CC"
C_ORANGE = "FCE4D6"
C_RED    = "FFDCE0"
C_WHITE  = "FFFFFF"

def thin_border():
    s = Side(style="thin", color="CCCCCC")
    return Border(left=s, right=s, top=s, bottom=s)

def header_cell(ws, row, col, value, bg=C_HEADER, size=10):
    c = ws.cell(row=row, column=col, value=value)
    c.font = Font(name="맑은 고딕", size=size, bold=True,
                  color="FFFFFF" if bg in (C_HEADER, C_SUB) else "000000")
    c.fill = PatternFill("solid", fgColor=bg)
    c.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
    c.border = thin_border()
    return c

def data_cell(ws, row, col, value, bg=C_WHITE, bold=False,
              align="left", wrap=False, num_fmt=None):
    c = ws.cell(row=row, column=col, value=value)
    c.font = Font(name="맑은 고딕", size=9, bold=bold)
    c.fill = PatternFill("solid", fgColor=bg)
    c.alignment = Alignment(horizontal=align, vertical="center", wrap_text=wrap)
    c.border = thin_border()
    if num_fmt:
        c.number_format = num_fmt
    return c


# ── 자체평가 이유 생성 ────────────────────────────────────────────────────────
# 자주 오매핑되는 메뉴의 알려진 원인
_KNOWN_ISSUE = {
    "SCR_1050": "호가(국내 현재가, sc=9.1M)가 '주식 가격/얼마' 의미로 Dense 상위 점령",
    "SCR_1300": "국내잔고(sc=14.9M) search_count boost가 관련 없는 쿼리까지 끌어올림",
    "SCR_1301": "미체결이 '주문/취소' 키워드에서 BM25 상위 → 주문창 대신 미체결 1위",
    "SCR_3032": "해외잔고가 해외주식 쿼리 전반에서 Dense 상위 점령",
    "SCR_3010": "해외주식 호가가 '주가/가격' 쿼리에서 Dense 오매핑",
    "SCR_1303": "예수금(sc=3.4M)이 '계좌/현금 얼마' 쿼리에서 Dense 1위",
    "SCR_1307": "수익률현황이 '수익/얼마' 의미와 유사하게 Dense 매핑",
    "SCR_5055": "매도담보대출이 '대출' BM25에서 담보대출보다 앞서 나옴",
    "SCR_5000": "입출금이 '출금/이체' 쿼리에서 정답보다 앞서 나옴",
    "SCR_6000": "알림센터가 알림 세부 메뉴보다 먼저 1위",
    "SCR_2152": "모의매매가 '정규전/대회' 보다 앞서 나옴 → embedding 미구분",
    "SCR_3050": "해외주식 종목차트가 국내주식 종목차트와 혼재",
}

def _make_reason(r):
    """케이스별 자체평가 이유 문자열 생성"""
    exp_id   = r["exp_id"]
    exp_name = r["exp_name"]
    top1_id  = r["top1_id"]
    top1_nm  = r["top1_name"]
    rank     = r["rank"]   # None이면 5위 밖
    ed       = r.get("exp_data")  # 정답 메뉴의 Dense/BM25 정보 (있으면)

    rank_str = f"정답 {rank}위" if rank else "정답 5위 밖(miss)"

    # ── 정답 1위 ────────────────────────────────────────────────────────
    if r["hit1"]:
        if ed:
            dr = ed.get("_dense_rank", "?")
            br = ed.get("_bm25_rank", 1000)
            b_str = f"BM25 {br}위" if br < 1000 else "BM25 미매칭(키워드 미포함)"
            sc = r.get("exp_sc_norm", 0)
            sc_str = f" + sc_boost({sc:.2f})" if sc > 0.7 else ""
            return f"✅ Dense {dr}위 · {b_str}{sc_str} → RRF 합산 1위"
        return "✅ 정상 매칭"

    # ── 정답이 상위 5개 안에 있지만 1위 아님 ─────────────────────────────
    known_top1 = _KNOWN_ISSUE.get(top1_id, "")
    if ed:
        dr = ed.get("_dense_rank", "?")
        br = ed.get("_bm25_rank", 1000)
        b_str = f"BM25 {br}위" if br < 1000 else "BM25 미매칭"
        exp_rrf = f"정답 Dense {dr}위 · {b_str}"
    else:
        exp_rrf = "정답 메뉴 Dense/BM25 모두 낮음"

    if known_top1:
        return f"❌ {known_top1}\n→ {rank_str} / {exp_rrf}"

    # ── 정답 메뉴가 아예 5위 밖 ────────────────────────────────────────
    cat = r["category"].split("-")[0] if "-" in r["category"] else r["category"]
    cat_reasons = {
        "랭킹/영웅전": "랭킹 메뉴 embedding 특화 부족 (구어체 쿼리 미포함)",
        "파생상품":    "파생상품 메뉴 embedding과 일반 쿼리 간 의미적 거리 큼",
        "환전":       "환전 관련 쿼리 BM25/Dense 매핑 약함",
        "주식더모으기": "소수 메뉴로 embedding 다양성 낮음",
        "생활/혜택":   "생활/혜택 카테고리 embedding 특화 부족",
    }
    cat_note = cat_reasons.get(cat, "")

    return (
        f"❌ '{top1_nm}'이 1위 / {rank_str}\n"
        f"{exp_rrf}"
        + (f"\n[카테고리 특성] {cat_note}" if cat_note else "")
    )


# ── 검색 실행 + 결과 수집 ────────────────────────────────────────────────────
def collect_results():
    queries = json.loads(EVAL_PATH.read_text(encoding="utf-8"))
    engine  = MenuSearchEngine.get_instance()
    # ChromaDB에서 sc_norm 조회용 캐시
    vs = engine.vectorstore

    rows = []
    cat_stats = defaultdict(lambda: {"total":0,"hit1":0,"hit3":0,"hit5":0,"rr":0.0})

    for item in queries:
        res = engine.search(item["query"], top_n=TOP_N, use_reranker=False)

        rank = None
        top_results = []
        exp_data = None
        for i, r in enumerate(res, 1):
            if i <= 5:
                top_results.append(f"{i}. {r['menu_name']} ({r['menu_id']})")
            if r["menu_id"] == item["expected_id"]:
                rank = i
                exp_data = {k: v for k, v in r.items()}

        hit1 = rank == 1
        hit3 = rank is not None and rank <= 3
        hit5 = rank is not None and rank <= 5

        # 정답 메뉴 sc_norm (ChromaDB에서 직접)
        try:
            meta_res = vs.collection.get(ids=[item["expected_id"]], include=["metadatas"])
            exp_sc_norm = meta_res["metadatas"][0].get("search_count_norm", 0) if meta_res["metadatas"] else 0
        except Exception:
            exp_sc_norm = 0

        row = {
            "id":          item["id"],
            "query":       item["query"],
            "exp_id":      item["expected_id"],
            "exp_name":    item["expected_name"],
            "category":    item["category"],
            "rank":        rank,
            "hit1":        hit1,
            "hit3":        hit3,
            "hit5":        hit5,
            "rr":          1/rank if rank else 0,
            "top1_name":   res[0]["menu_name"] if res else "",
            "top1_id":     res[0]["menu_id"]   if res else "",
            "top_list":    "\n".join(top_results),
            "exp_data":    exp_data,
            "exp_sc_norm": exp_sc_norm,
        }
        row["reason"] = _make_reason(row)
        rows.append(row)

        cat = item["category"].split("-")[0]
        s = cat_stats[cat]
        s["total"] += 1
        if hit1: s["hit1"] += 1
        if hit3: s["hit3"] += 1
        if hit5: s["hit5"] += 1
        s["rr"] += (1/rank if rank else 0)

    return rows, cat_stats


# ── 시트 1: 요약 ─────────────────────────────────────────────────────────────
def write_summary(wb, rows, cat_stats):
    ws = wb.active
    ws.title = "요약"
    ws.sheet_view.showGridLines = False
    for col, w in zip("ABCDE", [24, 12, 10, 10, 38]):
        ws.column_dimensions[col].width = w

    n = len(rows)
    acc1 = sum(r["hit1"] for r in rows) / n
    acc3 = sum(r["hit3"] for r in rows) / n
    acc5 = sum(r["hit5"] for r in rows) / n
    mrr  = sum(r["rr"]   for r in rows) / n

    # 제목
    ws.merge_cells("A1:E1")
    c = ws.cell(row=1, column=1, value="영웅문 S# AI 메뉴 검색 품질 평가 결과")
    c.font      = Font(name="맑은 고딕", size=14, bold=True, color="FFFFFF")
    c.fill      = PatternFill("solid", fgColor=C_HEADER)
    c.alignment = Alignment(horizontal="center", vertical="center")
    ws.row_dimensions[1].height = 34

    ws.merge_cells("A2:E2")
    c = ws.cell(row=2, column=1,
                value=f"평가일: {datetime.today().strftime('%Y-%m-%d')}  |  총 {n}개 쿼리  |  검색 모드: RRF 하이브리드(Dense+BM25)")
    c.font = Font(name="맑은 고딕", size=9, color="FFFFFF")
    c.fill = PatternFill("solid", fgColor=C_SUB)
    c.alignment = Alignment(horizontal="center", vertical="center")
    ws.row_dimensions[2].height = 18

    # 전체 지표 헤더
    ws.row_dimensions[3].height = 8
    for col, txt in enumerate(["지표", "점수", "정답 수", "전체", "설명"], 1):
        header_cell(ws, 4, col, txt, bg=C_SUB)
    ws.row_dimensions[4].height = 22

    for i, (label, val, hit, total, note) in enumerate([
        ("Acc@1  1위 정확도",   acc1, sum(r['hit1'] for r in rows), n, "1위 결과가 정답인 비율 (가장 중요)"),
        ("Acc@3  3위 내 정확도", acc3, sum(r['hit3'] for r in rows), n, "상위 3개 안에 정답이 포함된 비율"),
        ("Acc@5  5위 내 정확도", acc5, sum(r['hit5'] for r in rows), n, "상위 5개 안에 정답이 포함된 비율"),
        ("MRR    평균 역순위",   mrr,  "—",                         "—", "정답이 몇 위에 나왔는지 역수 평균 (높을수록 좋음)"),
    ], 5):
        bg = C_GREEN if val >= 0.6 else (C_YELLOW if val >= 0.4 else C_RED)
        data_cell(ws, i, 1, label, bold=True)
        c = data_cell(ws, i, 2, val, bg=bg, bold=True, align="center")
        c.number_format = "0.0%"
        data_cell(ws, i, 3, hit,   align="center")
        data_cell(ws, i, 4, total, align="center")
        data_cell(ws, i, 5, note)
        ws.row_dimensions[i].height = 20

    # 카테고리별
    ws.row_dimensions[9].height = 8
    for col, txt in enumerate(["카테고리", "Acc@1", "Acc@3", "Acc@5", "건수"], 1):
        header_cell(ws, 10, col, txt, bg=C_SUB)
    ws.row_dimensions[10].height = 22

    for i, (cat, s) in enumerate(sorted(cat_stats.items()), 11):
        t  = s["total"]
        a1 = s["hit1"] / t
        a3 = s["hit3"] / t
        a5 = s["hit5"] / t
        bg1 = C_GREEN if a1 >= 0.6 else (C_YELLOW if a1 >= 0.4 else C_RED)
        data_cell(ws, i, 1, cat, bold=True)
        c = data_cell(ws, i, 2, a1, bg=bg1, align="center"); c.number_format = "0%"
        c = data_cell(ws, i, 3, a3, align="center");         c.number_format = "0%"
        c = data_cell(ws, i, 4, a5, align="center");         c.number_format = "0%"
        data_cell(ws, i, 5, t, align="center")
        ws.row_dimensions[i].height = 20


# ── 시트 2: 전체 결과 (이유 포함) ────────────────────────────────────────────
def write_detail(wb, rows):
    ws = wb.create_sheet("전체 결과")
    ws.sheet_view.showGridLines = False
    ws.freeze_panes = "A2"

    cols = [
        ("No",      4),
        ("카테고리", 18),
        ("질문",    35),
        ("정답 메뉴\n(기대값)", 16),
        ("1위 메뉴\n(실제값)", 16),
        ("정답\n순위",  8),
        ("판정",  7),
        ("상위 5개 검색 결과", 42),
        ("자체 평가 이유\n(왜 맞았나 / 왜 틀렸나)", 52),
    ]
    for i, (title, width) in enumerate(cols, 1):
        header_cell(ws, 1, i, title)
        ws.column_dimensions[get_column_letter(i)].width = width
    ws.row_dimensions[1].height = 30

    for r in rows:
        row = r["id"] + 1
        if r["hit1"]:
            bg, mark = C_GREEN,  "✅ 1위"
        elif r["hit3"]:
            bg, mark = C_YELLOW, f"🔶 {r['rank']}위"
        elif r["hit5"]:
            bg, mark = C_ORANGE, f"🔸 {r['rank']}위"
        else:
            rank_label = "miss" if not r["rank"] else f"{r['rank']}위"
            bg, mark = C_RED, f"❌ {rank_label}"

        rank_val = r["rank"] if r["rank"] else "miss"
        same     = r["top1_id"] == r["exp_id"]

        data_cell(ws, row, 1, r["id"],      align="center")
        data_cell(ws, row, 2, r["category"])
        data_cell(ws, row, 3, r["query"])
        data_cell(ws, row, 4, f"{r['exp_name']}\n({r['exp_id']})", wrap=True)
        data_cell(ws, row, 5, r["top1_name"],
                  bg=C_GREEN if same else (C_RED if not same and not r["hit1"] else C_WHITE))
        data_cell(ws, row, 6, rank_val, bg=bg, align="center", bold=True)
        data_cell(ws, row, 7, mark,     bg=bg, align="center")
        data_cell(ws, row, 8, r["top_list"], wrap=True)
        data_cell(ws, row, 9, r["reason"],   wrap=True,
                  bg=C_GREEN if r["hit1"] else (C_YELLOW if r["hit3"] else (C_ORANGE if r["hit5"] else C_RED)))
        ws.row_dimensions[row].height = 50


# ── 시트 3: 틀린 케이스 (이유 포함) ─────────────────────────────────────────
def write_missed(wb, rows):
    ws = wb.create_sheet("개선 필요 (1위 불일치)")
    ws.sheet_view.showGridLines = False
    ws.freeze_panes = "A2"

    cols = [
        ("No", 4), ("카테고리", 18), ("질문", 35),
        ("정답 메뉴", 16), ("1위 메뉴", 16), ("정답 순위", 8),
        ("상위 5개 결과", 42), ("자체 평가 이유", 52),
    ]
    for i, (title, width) in enumerate(cols, 1):
        header_cell(ws, 1, i, title)
        ws.column_dimensions[get_column_letter(i)].width = width
    ws.row_dimensions[1].height = 22

    missed = [r for r in rows if not r["hit1"]]
    for idx, r in enumerate(missed, 2):
        bg       = C_YELLOW if r["hit3"] else (C_ORANGE if r["hit5"] else C_RED)
        rank_val = r["rank"] if r["rank"] else "miss"
        data_cell(ws, idx, 1, r["id"],      align="center")
        data_cell(ws, idx, 2, r["category"])
        data_cell(ws, idx, 3, r["query"])
        data_cell(ws, idx, 4, f"{r['exp_name']} ({r['exp_id']})")
        data_cell(ws, idx, 5, r["top1_name"], bg=C_RED)
        data_cell(ws, idx, 6, rank_val, bg=bg, align="center", bold=True)
        data_cell(ws, idx, 7, r["top_list"], wrap=True)
        data_cell(ws, idx, 8, r["reason"],   wrap=True, bg=bg)
        ws.row_dimensions[idx].height = 52


# ── 메인 ─────────────────────────────────────────────────────────────────────
def main():
    print("검색 실행 중 (100개 쿼리)...")
    rows, cat_stats = collect_results()

    n    = len(rows)
    acc1 = sum(r["hit1"] for r in rows) / n * 100
    acc3 = sum(r["hit3"] for r in rows) / n * 100
    acc5 = sum(r["hit5"] for r in rows) / n * 100
    mrr  = sum(r["rr"]   for r in rows) / n * 100
    print(f"  Acc@1={acc1:.1f}%  Acc@3={acc3:.1f}%  Acc@5={acc5:.1f}%  MRR={mrr:.1f}%")

    wb = openpyxl.Workbook()
    write_summary(wb, rows, cat_stats)
    write_detail(wb, rows)
    write_missed(wb, rows)

    out_path = (
        Path(__file__).parent.parent / "data" / "generated" /
        f"search_eval_{datetime.today().strftime('%Y%m%d')}.xlsx"
    )
    out_path.parent.mkdir(parents=True, exist_ok=True)
    wb.save(out_path)
    print(f"\n✅ 저장 완료: {out_path}")


if __name__ == "__main__":
    main()