src/qa_judgment.py

"""
KCSC QA Judgment engine (rule-based, domain-knowledge driven).

Produces draft judgments and missing-condition lists for construction-review
queries. Used by:

* `generate_qa_candidates.py` (offline evaluation against the verified golden set)
* `mcp_server.py` `qa_judgment` tool (live MCP queries)

Design rule (per `docs/qa_eval_plan.md`):
This module MUST NOT read `expected.codes`, `expected.judgment`, or
`expected.missing_conditions` from the golden set. Rules are written as a
codification of standard Korean structural-engineering review practice
(checklists a senior reviewer would intuit from work_type / check_item /
member alone, plus measurement-style terms found in retrieved KDS/KCS text).

The engine is intentionally conservative: it never emits `pass` or `fail`.
Those labels require explicit numerical/spec input which is rarely present in
free-text review queries; the safe default is `conditional_pass` or
`needs_more_info`. A future LLM-backed engine can extend this to the full
4-label space.
"""

from __future__ import annotations

import re
from typing import Any, Iterable

# ---------------------------------------------------------------------------
# Judgment label heuristics (carried over from generate_qa_candidates.py)
# ---------------------------------------------------------------------------

CONCRETE_NUMBER_PATTERN = re.compile(r"\d+(\.\d+)?\s*(mpa|kn|mm|m\b|%|°c|c)", re.I)

CONDITIONAL_PASS_SIGNALS: tuple[str, ...] = (
    "검측", "품질 확인", "품질관리", "품질 판정", "품질 검사",
    "공통 확인사항", "시공 전 검토", "시공 및 계측", "재하시험",
    "시험조정", "품질시험", "가설 계획", "추락방지", "안전시설",
    "일반사항", "공장제작", "고장력볼트",
)

NEEDS_MORE_INFO_SIGNALS: tuple[str, ...] = (
    "적합한지", "충분한지", "판단", "설계 검토", "설계기준",
    "적용 여부", "지진하중", "내진", "한계상태", "접합부",
    "지지력", "배합강도", "안전한지",
)


# ---------------------------------------------------------------------------
# Missing-condition taxonomy
# ---------------------------------------------------------------------------
# Each entry: (required_keywords_tuple, conditions). The first rule whose
# keywords ALL appear in the case text wins. Order from most specific to most
# general. Conditions reflect typical engineering-review checklists.
# ---------------------------------------------------------------------------

SPEC_TERM_RULES: tuple[tuple[tuple[str, ...], list[str]], ...] = (
    # ── Concrete (specific work types first) ─────────────────────────────
    (("매스", "콘크리트"), ["부재 치수", "타설 온도", "양생 계획", "균열 제어 계획"]),
    (("팽창", "콘크리트"), ["팽창재 종류", "배합 조건", "양생 조건"]),
    (("수중", "콘크리트"), ["타설 방법", "수심", "유속", "배합 조건"]),
    (("일반콘크리트", "타설"), ["배합표", "슬럼프", "공기량", "타설 온도"]),
    (("타설", "품질"), ["배합표", "슬럼프", "공기량", "타설 온도"]),
    (("철근공사", "배근"), ["철근 규격", "배근 간격", "피복 두께", "도면 요구사항"]),
    (("철근공사", "검측"), ["철근 규격", "배근 간격", "피복 두께", "도면 요구사항"]),
    (("정착", "철근"), ["부재 종류", "철근 지름", "콘크리트 강도", "정착 위치"]),
    (("동바리",), ["동바리 높이", "하중 조건", "설치 간격", "해체 시기"]),
    (("거푸집",), ["동바리 높이", "하중 조건", "설치 간격", "해체 시기"]),
    (("콘크리트공사", "공통"), ["공사 종류", "적용 시방서", "품질 기준"]),

    # ── Steel ────────────────────────────────────────────────────────────
    (("강구조공사", "일반"), ["공사 범위", "적용 재료", "제작 및 설치 계획"]),
    (("공장제작",), ["제작 도면", "재료 성적서", "제작 허용오차"]),
    (("강재",), ["강재 규격", "성적서", "검사 항목"]),
    (("한계상태",), ["부재 종류", "하중 조건", "설계 방법"]),
    (("고장력볼트",), ["볼트 등급", "체결 축력", "검사 방법"]),
    (("볼트", "접합"), ["볼트 종류", "접합 형식", "체결 방법", "검사 기준"]),
    (("용접",), ["용접 종류", "용접 절차서", "검사 등급", "비파괴검사 여부"]),

    # ── Ground / Foundation ──────────────────────────────────────────────
    (("지반조사",), ["조사 위치", "시추 수량", "시험 결과", "지층 조건"]),
    (("흙막이", "구조"), ["적용 시설", "지반정수", "하중 조합", "굴착 단계"]),
    (("흙막이", "시공"), ["계측 계획", "관리 기준치", "시공 단계"]),
    (("흙막이",), ["굴착 깊이", "지하수위", "흙막이 형식", "주변 구조물"]),
    (("얕은기초",), ["지지층", "허용지지력", "기초 형식", "침하 검토"]),
    # Pile QC judgment uses test results, not the test method itself — order matters
    (("말뚝", "품질"), ["시험 결과", "침하 기준", "설계 지지력"]),
    (("말뚝", "재하시험"), ["시험 방법", "하중 단계", "침하량", "판정 기준"]),
    (("연약지반",), ["개량 공법", "지반 조건", "품질 관리 항목"]),

    # ── Quality / Tests ──────────────────────────────────────────────────
    (("품질관리계획",), ["공사 규모", "시험 계획", "품질관리 조직"]),
    (("상수도",), ["관종", "시험 압력", "시험 시간", "누수 여부"]),
    (("시험조정",), ["시험 항목", "성능 기준", "측정 결과"]),
    (("농업생산기반",), ["시설 종류", "계측 항목", "관리 기준"]),
    (("품질시험",), ["시험 종목", "빈도", "공종", "시료 채취 방법"]),
    (("시험", "기준"), ["시험 항목", "재료 종류", "적용 공종"]),

    # ── Safety / Temporary works ─────────────────────────────────────────
    (("안전관리계획",), ["공사 규모", "위험 작업", "안전관리 조직", "점검 계획"]),
    (("하수급",), ["하수급인 작업 범위", "위험성평가", "교육 및 점검 기록"]),
    (("가설공사",), ["가설 종류", "하중 조건", "설치 위치"]),
    (("현장가설시설물",), ["설비 종류", "용량", "배치 계획", "안전 조치"]),
    (("추락방지",), ["시설 종류", "설치 위치", "작업 높이"]),
    (("안전시설", "전문시방"), ["시설 형식", "설치 구간", "검측 항목"]),
    (("철도", "안전"), ["작업 종류", "철도 운행 영향", "안전 조치"]),

    # ── Seismic ──────────────────────────────────────────────────────────
    (("내진설계", "적용"), ["구조물 종류", "내진등급", "지역계수", "중요도"]),
    (("내진설계", "지진하중"), ["지반분류", "위험도", "응답수정계수", "중요도계수"]),
    (("콘크리트구조", "내진상세"), ["부재 종류", "내진상세 등급", "철근 배근", "연성 요구"]),
    (("강구조", "내진"), ["내진시스템", "접합 형식", "하중 조합", "연성 요구"]),
    (("합성구조", "내진"), ["합성부재 형식", "내진시스템", "접합 상세", "하중 조건"]),
    (("기초", "내진"), ["기초 형식", "지반분류", "액상화 여부", "상부구조 반력"]),
    (("비탈면", "내진"), ["비탈면 높이", "지반정수", "지진계수", "배수 조건"]),

    # ── Composite structures ─────────────────────────────────────────────
    (("합성콘크리트", "전단"), ["부재 형식", "연결재 상세", "하중 조건", "재료 강도"]),
    # Steel-composite QC inspection precedes generic composite-construction rule
    (("강합성", "검사"), ["시공 단계", "검사 항목", "부재 형식"]),
    (("합성구조", "시공"), ["합성구조 형식", "타설 계획", "품질시험 항목"]),
    (("합성보",), ["합성보 단면", "전단연결재", "하중 조합", "시공 단계"]),
    (("합성기둥",), ["부재 형식", "축력", "모멘트", "재료 강도"]),
    (("강-콘크리트", "접합"), ["접합부 상세", "하중 전달 경로", "재료 강도"]),
    (("강합성교",), ["교량 형식", "경간", "하중 조건", "시공 단계"]),
)


# Domain default fallback (used when no SPEC_TERM_RULES matches).
DOMAIN_DEFAULT_MISSING: dict[str, list[str]] = {
    "concrete_rebar_formwork": ["부재 종류", "재료 강도", "시공 위치", "도면 요구사항"],
    "ground_foundation_earth_retaining": ["지반 조건", "하중 조건", "시공 단계", "계측 결과"],
    "steel_connection_bolts": ["부재 규격", "접합 형식", "재료 등급", "검사 기준"],
    "quality_test_inspection_tolerance": ["시험 항목", "시험 빈도", "허용 기준", "검사 기록"],
    "safety_temporary_construction_management": ["현장 조건", "위험 요인", "관리 계획", "점검 기록"],
    "seismic_design_detailing": ["내진 등급", "지반 조건", "하중 산정 조건", "상세 도면"],
    "composite_structure_criteria": ["부재 형식", "재료 강도", "연결 상세", "하중 조건"],
}


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def build_case_text(case: dict[str, Any]) -> str:
    """Concatenate query + structured field values for keyword matching."""
    structured = case.get("input", {}).get("structured", {})
    parts = [case.get("input", {}).get("query", "")]
    parts.extend(str(v) for v in structured.values() if v)
    return " ".join(parts)


def has_concrete_conditions(case: dict[str, Any]) -> bool:
    """Detect that the user actually supplied measurable conditions."""
    structured = case.get("input", {}).get("structured", {})
    if structured.get("conditions"):
        return True
    query = case.get("input", {}).get("query", "")
    return bool(CONCRETE_NUMBER_PATTERN.search(query))


def infer_judgment(case: dict[str, Any]) -> str:
    text = build_case_text(case)
    if has_concrete_conditions(case):
        return "conditional_pass"
    if any(sig in text for sig in NEEDS_MORE_INFO_SIGNALS):
        return "needs_more_info"
    if any(sig in text for sig in CONDITIONAL_PASS_SIGNALS):
        return "conditional_pass"
    return "needs_more_info"


def _all_in(text: str, terms: Iterable[str]) -> bool:
    return all(term in text for term in terms)


def infer_missing_conditions(case: dict[str, Any]) -> list[str]:
    """Return up to 4 missing conditions following review-checklist rules."""
    text = build_case_text(case)

    for terms, conditions in SPEC_TERM_RULES:
        if _all_in(text, terms):
            return list(conditions)

    structured = case.get("input", {}).get("structured", {})
    work_type = structured.get("work_type")
    check_item = structured.get("check_item")
    inferred: list[str] = []
    if work_type:
        inferred.append(f"{work_type} 적용 범위")
    if check_item:
        inferred.append(f"{check_item} 판정 기준")
    inferred.extend(
        DOMAIN_DEFAULT_MISSING.get(case.get("domain"), ["현장 조건", "설계도서 요구사항"])
    )

    deduped: list[str] = []
    for value in inferred:
        if value and value not in deduped:
            deduped.append(value)
    return deduped[:4]


def judge_case(case: dict[str, Any]) -> dict[str, Any]:
    """Produce the full judgment payload for a case."""
    return {
        "judgment": infer_judgment(case),
        "missing_conditions": infer_missing_conditions(case),
        "generator": "rule_based_taxonomy_v2",
    }