"""Customer QA benchmark parsing and deterministic SQL evaluation.""" from __future__ import annotations from dataclasses import dataclass import hashlib import json from pathlib import Path import re from typing import Any, Mapping class QaBenchmarkError(RuntimeError): """Raised when the QA benchmark source cannot be used safely.""" @dataclass(frozen=True) class QaQuestion: question_id: int | None question_code: str category: str title: str question_text: str source_document: str source_sheet: str source_row: int | None source_scenario: str sample_sql: str expected_focus: str baseline_sql: str baseline_answer: str support_level: str evaluation_rule: Mapping[str, Any] last_judgment_status: str = "" last_evaluated_at: str = "" @dataclass(frozen=True) class QaJudgment: status: str reason: str def question_fingerprint(question_text: str) -> str: normalized = " ".join(str(question_text or "").split()).casefold() return hashlib.sha256(normalized.encode("utf-8")).hexdigest() def _compact_text(value: object) -> str: return str(value or "").strip() def _string_list(value: object) -> tuple[str, ...]: if not isinstance(value, list): return () return tuple(_compact_text(item) for item in value if _compact_text(item)) def load_benchmark_questions(path: Path) -> tuple[QaQuestion, ...]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, ValueError) as exc: raise QaBenchmarkError(f"질답 기준 파일을 읽지 못했습니다: {path}") from exc rows = payload.get("scenarios") if isinstance(payload, Mapping) else None if not isinstance(rows, list): raise QaBenchmarkError("질답 기준 파일에 scenarios 배열이 필요합니다.") questions: list[QaQuestion] = [] seen_codes: set[str] = set() for row in rows: if not isinstance(row, Mapping): raise QaBenchmarkError("질답 기준의 각 시나리오는 객체여야 합니다.") source = row.get("source") if isinstance(row.get("source"), Mapping) else {} history = ( row.get("historical_answer") if isinstance(row.get("historical_answer"), Mapping) else {} ) code = _compact_text(row.get("case_id")).upper() question_text = _compact_text(row.get("question")) if not code or not question_text: raise QaBenchmarkError("각 질답 기준에는 case_id와 question이 필요합니다.") if code in seen_codes: raise QaBenchmarkError(f"중복된 질답 case_id입니다: {code}") evaluation_rule = row.get("evaluation_rule") if not isinstance(evaluation_rule, Mapping): evaluation_rule = {} questions.append( QaQuestion( question_id=None, question_code=code, category=_compact_text(row.get("category")) or "GENERAL", title=_compact_text(row.get("title")) or code, question_text=question_text, source_document=_compact_text(source.get("workbook")), source_sheet=_compact_text(source.get("sheet")), source_row=_number_or_none(source.get("excel_row")), source_scenario=_compact_text(source.get("scenario")), sample_sql=_compact_text(source.get("sample_query")), expected_focus=_compact_text(row.get("expected_focus")), baseline_sql=_compact_text(history.get("generated_sql")), baseline_answer=_compact_text(history.get("answer_text")), support_level=_compact_text(row.get("support_level")).upper() or "UNKNOWN", evaluation_rule={ "required_sql_terms": list( _string_list(evaluation_rule.get("required_sql_terms")) ), "recommended_sql_terms": list( _string_list(evaluation_rule.get("recommended_sql_terms")) ), }, ) ) seen_codes.add(code) return tuple(questions) def question_from_record(record: Mapping[str, Any]) -> QaQuestion: rule = record.get("evaluation_rule") if isinstance(rule, str): try: rule = json.loads(rule) except ValueError: rule = {} if not isinstance(rule, Mapping): rule = {} return QaQuestion( question_id=_number_or_none(record.get("question_id")), question_code=_compact_text(record.get("question_code")), category=_compact_text(record.get("category")) or "GENERAL", title=_compact_text(record.get("title")) or _compact_text(record.get("question_code")), question_text=_compact_text(record.get("question_text")), source_document=_compact_text(record.get("source_document")), source_sheet=_compact_text(record.get("source_sheet")), source_row=_number_or_none(record.get("source_row")), source_scenario=_compact_text(record.get("source_scenario")), sample_sql=_compact_text(record.get("sample_sql")), expected_focus=_compact_text(record.get("expected_focus")), baseline_sql=_compact_text(record.get("baseline_sql")), baseline_answer=_compact_text(record.get("baseline_answer")), support_level=_compact_text(record.get("support_level")).upper() or "UNKNOWN", evaluation_rule={ "required_sql_terms": list( _string_list(rule.get("required_sql_terms")) ), "recommended_sql_terms": list( _string_list(rule.get("recommended_sql_terms")) ), }, last_judgment_status=_compact_text(record.get("last_judgment_status")), last_evaluated_at=_compact_text(record.get("last_evaluated_at")), ) def _number_or_none(value: object) -> int | None: if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None def _append_issue(issues: list[str], condition: bool, message: str) -> None: if condition: issues.append(message) def evaluate_sql( question: QaQuestion | None, generated_sql: str, *, execution_succeeded: bool, error_text: str = "", ) -> QaJudgment: """Evaluate the generated SQL against the customer-approved benchmark rule.""" if question is None or not question.question_code: return QaJudgment( status="REVIEW", reason="자유 질의입니다. 고객 기준 정답 시나리오와 연결되지 않아 수동 검토가 필요합니다.", ) sql = _compact_text(generated_sql) upper_sql = sql.upper() lower_sql = sql.lower() execution_status = "PASS" if execution_succeeded else "FAIL_EXECUTION" issues: list[str] = [] failure_markers = ("could not be generated", "exception encountered", "invalid identifier", "ora-") has_failure_text = any(marker in lower_sql for marker in failure_markers) required = _string_list(question.evaluation_rule.get("required_sql_terms")) recommended = _string_list(question.evaluation_rule.get("recommended_sql_terms")) missing_required = [term for term in required if term.upper() not in upper_sql] missing_recommended = [term for term in recommended if term.upper() not in upper_sql] if not execution_succeeded: issues.append(f"실행 상태가 {execution_status}입니다.") if not sql: issues.append("생성 SQL이 없습니다.") if has_failure_text: issues.append("생성 SQL에 오류 또는 생성 실패 문구가 포함되어 있습니다.") if missing_required: issues.append("필수 SQL 요소 누락: " + ", ".join(missing_required)) if missing_recommended: issues.append("권장 SQL 요소 누락: " + ", ".join(missing_recommended)) _append_issue( issues, bool(re.search(r'_[A-Z0-9]*YN"\s*=\s*\'1\'', sql, flags=re.IGNORECASE)), "*_YN 컬럼은 샘플 메타데이터의 Y/N 값으로 비교해야 합니다.", ) _append_issue( issues, bool(re.search(r'_[A-Z0-9]*FLAG"\s*=\s*\'Y\'', sql, flags=re.IGNORECASE)), "*_FLAG 컬럼은 샘플 메타데이터의 0/1 값으로 비교해야 합니다.", ) support = question.support_level if support == "UNSUPPORTED": uses_alias_lookup = "COMN_GAME_ALIAS_BAS" in upper_sql substitutes_sample_game = "STOVE_CHAOSZERO" in upper_sql if not sql and any(marker in error_text.lower() for marker in failure_markers): return QaJudgment("PASS", "미지원 게임 질문이 실행 가능한 SQL로 변환되지 않았습니다. 기대한 안전 차단입니다.") if uses_alias_lookup and not substitutes_sample_game and execution_succeeded: return QaJudgment("PASS", "미지원 게임을 별칭 테이블로만 확인했고 샘플 게임 ID를 임의 대입하지 않았습니다.") return QaJudgment("FAIL", "미지원 게임이 안전한 별칭 조회로 제한되지 않았거나 실행에 실패했습니다.") if not execution_succeeded or not sql or has_failure_text or missing_required: return QaJudgment("FAIL", "\n".join(issues) or "필수 SQL 또는 실행 검증에 실패했습니다.") _apply_case_specific_rules(question.question_code, sql, upper_sql, issues) if any(issue.startswith("필수") or issue.startswith("월간") or issue.startswith("일별") or issue.startswith("주간") or issue.startswith("CZN-") for issue in issues): return QaJudgment("FAIL", "\n".join(issues)) if support == "PARTIAL": issues.append("지원 범위가 일부인 질문이므로 결과 범위를 함께 검토해야 합니다.") if issues: return QaJudgment("WARN", "\n".join(issues)) return QaJudgment("PASS", "필수 테이블·컬럼·집계 조건과 실행 결과를 확인했습니다.") def _apply_case_specific_rules(code: str, sql: str, upper_sql: str, issues: list[str]) -> None: if code == "STD-26": _append_issue(issues, "TRUNC(DATE" in upper_sql, "월간 NRU는 DATE 리터럴이 아니라 BASE_DT 기준 월 범위를 사용해야 합니다.") _append_issue(issues, "MAX" not in upper_sql or "BASE_DT" not in upper_sql, "월간 NRU는 최신 MAX(BASE_DT) 스냅샷을 기준으로 해야 합니다.") _append_issue(issues, "RAW_NRU_DT" in upper_sql and "TRUNC(" not in upper_sql, "월간 NRU의 RAW_NRU_DT 기간은 BASE_DT 기준으로 계산해야 합니다.") elif code == "STD-27": _append_issue(issues, "AU_FLAG" in upper_sql, "월간 AU는 AU_FLAG가 아니라 LAST_CONN_DT 전월 조건으로 계산해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("MAX", "BASE_DT", "ADD_MONTHS")), "월간 AU는 최신 BASE_DT의 전월을 기준으로 해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("STD_USER_YN", "EXPT_USER_YN", "LAST_CONN_DT")), "월간 AU에는 STD_USER_YN, EXPT_USER_YN, LAST_CONN_DT 조건이 필요합니다.") elif code == "STD-22": _append_issue(issues, "NRU_FLAG" not in upper_sql, "일별 NRU는 NRU_FLAG=1을 사용해야 합니다.") elif code == "STD-25": _append_issue(issues, "LAST_CONN_DT" not in upper_sql, "최근 7일 AU는 LAST_CONN_DT 기간 조건을 사용해야 합니다.") _append_issue(issues, "AU_FLAG" in upper_sql, "최근 7일 AU는 AU_FLAG로 제한하면 안 됩니다.") _append_issue(issues, any(term not in upper_sql for term in ("STD_USER_YN", "EXPT_USER_YN")), "최근 7일 AU에는 STD_USER_YN과 EXPT_USER_YN 조건이 필요합니다.") _append_issue(issues, "GROUP BY" in upper_sql, "최근 7일 AU는 일자별 목록이 아니라 단일 집계여야 합니다.") elif code == "STD-28": _append_issue(issues, any(term not in upper_sql for term in ("STD_USER_YN", "EXPT_USER_YN")), "전체 유저 수에는 STD_USER_YN과 EXPT_USER_YN 조건이 필요합니다.") _append_issue(issues, "RU_FLAG" in upper_sql, "이 기준의 전체 유저 수는 RU_FLAG가 아니라 STD_USER_YN으로 계산해야 합니다.") elif code in {"CZN-07", "CZN-08", "CZN-13", "CZN-17"}: _apply_goods_rules(code, sql, upper_sql, issues) def _apply_goods_rules(code: str, sql: str, upper_sql: str, issues: list[str]) -> None: dimension_name_terms = ("DIM_KR_NM", "DIM_EN_NM", "DIM_CTG", "CD_DESC", "CD_DTL_DESC") has_dimension_name = any(term in upper_sql for term in dimension_name_terms) if code in {"CZN-07", "CZN-08", "CZN-17"}: _append_issue(issues, "CZN_COMN_SVC_DIM_BAS" not in upper_sql or not has_dimension_name, f"{code}은 CZN_COMN_SVC_DIM_BAS의 이름/분류 컬럼으로 재화명을 해석해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("DIM_TYPE_DIV_CD", "DIM_CTG")) or "크리스탈" not in sql, f"{code}은 GOODS_AGG와 크리스탈 분류 조건을 사용해야 합니다.") _append_issue(issues, "HAVE_CNT" in upper_sql and "<>" not in upper_sql and "!=" not in upper_sql, f"{code}은 HAVE_CNT <> 0으로 0 보유량을 제외해야 합니다.") if code == "CZN-07": _append_issue(issues, any(term not in upper_sql for term in ("CZN_COMN_USER_MST", "EXPT_USER_YN")), "CZN-07은 유저 마스터를 조인하고 제외 유저를 필터링해야 합니다.") _append_issue(issues, "GROUP BY" not in upper_sql or "BASE_DT" not in upper_sql, "CZN-07은 기간별 일자 집계를 위해 BASE_DT GROUP BY가 필요합니다.") elif code == "CZN-08": _append_issue(issues, any(term not in upper_sql for term in ("AU_FLAG", "EXPT_USER_YN")), "CZN-08은 표준 AU와 제외 유저 조건이 필요합니다.") _append_issue(issues, "STD_USER_YN" in upper_sql, "CZN-08 기준에는 STD_USER_YN을 추가하면 모집단이 과도하게 좁아집니다.") _append_issue(issues, "AVG(" in upper_sql, "CZN-08 1인당 평균은 AVG(HAVE_CNT)가 아니라 합계/고유 유저 수여야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("SUM", "COUNT", "DISTINCT", "/")), "CZN-08 평균은 SUM(HAVE_CNT)/COUNT(DISTINCT GUID)로 계산해야 합니다.") elif code == "CZN-13": _append_issue(issues, "CZN_CUSTOM_GOODS_HAVE_TXN" in upper_sql, "CZN-13은 보유 스냅샷이 아니라 재화 변동 테이블을 사용해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("CZN_CUSTOM_GOODS_CHANGE_TXN", "GOODS_CHANGE_CNT")), "CZN-13은 GOODS_CHANGE_CNT를 사용해야 합니다.") _append_issue(issues, "CZN_COMN_SVC_DIM_BAS" not in upper_sql or not has_dimension_name, "CZN-13은 차원 테이블로 에테르를 해석해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("DIM_TYPE_DIV_CD", "DIM_CTG", "CHANGE_TYPE_CD", "'USE'")) or "에테르" not in sql, "CZN-13은 에테르 사용 분류와 CHANGE_TYPE_CD='USE' 조건이 필요합니다.") _append_issue(issues, any(term not in upper_sql for term in ("CZN_COMN_USER_MST", "EXPT_USER_YN", "COUNT", "DISTINCT", "GUID", "SUM")), "CZN-13은 대상 유저 조인, 고유 사용 유저 수, 사용량 합계가 필요합니다.") elif code == "CZN-17": _append_issue(issues, any(term not in upper_sql for term in ("CZN_CUSTOM_GOODS_HAVE_TXN", "HAVE_CNT")), "CZN-17은 재화 보유 스냅샷을 사용해야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("CZN_COMN_CHARACTER_MST", "AU_FLAG", "EXPT_USER_YN")), "CZN-17은 캐릭터 AU와 제외 유저 조건이 필요합니다.") _append_issue(issues, "AVG(" in upper_sql, "CZN-17 평균은 AVG(HAVE_CNT)가 아니라 합계/고유 캐릭터 수여야 합니다.") _append_issue(issues, any(term not in upper_sql for term in ("SUM", "COUNT", "DISTINCT", "/", "CUID")), "CZN-17 평균은 SUM(HAVE_CNT)/COUNT(DISTINCT CUID)로 계산해야 합니다.") _append_issue(issues, "GROUP BY" in upper_sql, "CZN-17은 기준일 단일 집계여야 하므로 상세 GROUP BY를 사용하면 안 됩니다.") _append_issue(issues, any(term in upper_sql for term in ("LEVEL_COL", "ACM_CONN_DCNT", "TDAY_PLAY_TIME")), "CZN-17에는 레벨·접속일·플레이타임 지표가 포함되면 안 됩니다.")