"""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 = "", game_plan_status: 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": plan_status = _compact_text(game_plan_status).upper() safe_empty_result = bool( re.search(r"\bFROM\s+DUAL\b", upper_sql) and re.search(r"\bWHERE\s+1\s*=\s*0\b", upper_sql) ) if plan_status in {"UNAVAILABLE", "UNMATCHED"} and execution_succeeded and safe_empty_result: return QaJudgment( "PASS", "게임 계획이 데이터 미지원 또는 미매칭으로 판정됐고, 임의 객체 선택 없이 빈 결과를 반환했습니다.", ) if not sql and any(marker in error_text.lower() for marker in failure_markers): return QaJudgment("PASS", "미지원 게임 질문이 실행 가능한 SQL로 변환되지 않았습니다. 기대한 안전 차단입니다.") 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 또는 실행 검증에 실패했습니다.") if any(issue.startswith("필수") 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", "고객 기준의 필수 SQL 요소와 실행 결과를 확인했습니다.")