Macbook
627 lines
28 KiB
Python
627 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""간편조회 기능 다양 질문 테스트 → test.csv
|
|
|
|
- domain-query: 간편조회 UI와 동일 경로 (다수 케이스, 빠름)
|
|
- /ask: 실제 고객 자연어 질문 (기능당 소수)
|
|
|
|
사용:
|
|
PYTHONUNBUFFERED=1 python3 scripts/quick_inquiry_nlu_test.py \\
|
|
--base http://127.0.0.1:8083/web --out test.csv
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import re
|
|
import ssl
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections import Counter
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
# 운영 HTTPS 등에서 로컬 CA 이슈 시 사용
|
|
_SSL_CTX = ssl.create_default_context()
|
|
|
|
|
|
def one_line(text: str, limit: int = 280) -> str:
|
|
t = re.sub(r"\s+", " ", (text or "").strip())
|
|
return (t[: limit - 1] + "…") if len(t) > limit else t
|
|
|
|
|
|
def post_json(url: str, body: Dict[str, Any], timeout: float) -> Tuple[Optional[int], Any, float]:
|
|
t0 = time.time()
|
|
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
url, data=data, headers={"Content-Type": "application/json; charset=utf-8"}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout, context=_SSL_CTX) as resp:
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|
try:
|
|
return resp.status, json.loads(raw), time.time() - t0
|
|
except json.JSONDecodeError:
|
|
return resp.status, raw, time.time() - t0
|
|
except urllib.error.HTTPError as e:
|
|
try:
|
|
payload = e.read().decode("utf-8", errors="replace")
|
|
try:
|
|
return e.code, json.loads(payload), time.time() - t0
|
|
except Exception:
|
|
return e.code, payload, time.time() - t0
|
|
except Exception:
|
|
return e.code, str(e), time.time() - t0
|
|
except Exception as e:
|
|
return None, str(e), time.time() - t0
|
|
|
|
|
|
def get_json(url: str, timeout: float) -> Tuple[Optional[int], Any, float]:
|
|
t0 = time.time()
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout, context=_SSL_CTX) as resp:
|
|
return resp.status, json.loads(resp.read().decode("utf-8", errors="replace")), time.time() - t0
|
|
except Exception as e:
|
|
return None, str(e), time.time() - t0
|
|
|
|
|
|
def domain_cases_prod_light() -> List[Dict[str, Any]]:
|
|
"""운영용 — 기능당 1~2건만 (부하 최소화)."""
|
|
light: List[Tuple[str, str, str, Dict[str, str]]] = [
|
|
("미납조회", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?", {"carNo": "12가3456"}),
|
|
("환불조회", "FARE_REFUND", "잘못 낸 톨비 환불받을 수 있어? 12가3456", {"carNo": "12가3456"}),
|
|
("구간통행요금", "FARE_SEARCH", "서울에서 부산까지 톨비 얼마 나와?", {"fromIc": "서울", "toIc": "부산"}),
|
|
("구간통행요금", "FARE_SEARCH", "판교에서 신갈 가면 요금이 어떻게 돼?", {"fromIc": "판교", "toIc": "신갈"}),
|
|
("구간통행시간", "TROAD_TRAVEL_TIME", "서울에서 부산까지 지금 가면 얼마나 걸려?",
|
|
{"fromIc": "서울", "toIc": "부산", "carType": "1"}),
|
|
("구간통행시간", "TROAD_TRAVEL_TIME", "판교에서 신갈 지금 막혀? 몇 분 걸려?",
|
|
{"fromIc": "판교", "toIc": "신갈", "carType": "1"}),
|
|
("차단계획", "CONSTRUCTION_PLAN", "경부선 오늘 공사해서 막히는 데 있어?", {"routeName": "경부선"}),
|
|
("차단계획", "CONSTRUCTION_PLAN", "영동선 차로 막는 일정 있나?", {"routeName": "영동선"}),
|
|
("영업소정보", "IC_TEL", "판교 톨게이트 전화번호 좀", {"icName": "판교"}),
|
|
("영업소정보", "IC_TEL", "기흥 톨게이트 주소랑 전화 좀", {"icName": "기흥"}),
|
|
("긴급구조기관", "EMERGENCY_HELP_ORG", "경부선에서 사고 나면 어디로 연락해?",
|
|
{"listMode": "true", "routeName": "경부선"}),
|
|
("휴게소음식", "REST_AREA_FOOD_LIST", "죽전휴게소에서 뭐 먹을 거 있어?", {"restAreaName": "죽전"}),
|
|
("휴게소브랜드매장", "BRAND_SHOP", "죽전에 편의점이나 카페 있어?", {"restAreaName": "죽전"}),
|
|
("휴게소시설", "REST_AREA_FACILITY", "죽전휴게소에 수유실 있나?", {"restAreaName": "죽전"}),
|
|
("휴게소주유", "GASSTATION", "죽전휴게소 기름값 얼마야?", {"restAreaName": "죽전"}),
|
|
("휴게소주유", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?",
|
|
{"compareMode": "cheapest", "fuelType": "휘발유", "routeName": "경부선"}),
|
|
("휴게소충전소", "REST_AREA_CHARGING", "죽전에서 전기차 충전돼?",
|
|
{"restAreaName": "죽전", "chargingType": "all"}),
|
|
("휴게소충전소", "REST_AREA_CHARGING", "강릉휴게소에 충전소 있어?",
|
|
{"restAreaName": "강릉", "chargingType": "all"}),
|
|
("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", "죽전휴게소에 청년들 장사하는 가게 있어?",
|
|
{"restAreaName": "죽전"}),
|
|
("가까운휴게소", "REST_AREA_NEAREST", "판교 근처 휴게소 어디 있어?", {"icName": "판교"}),
|
|
("노선혼잡", "TROAD_CONGESTION", "경부선 지금 막히는 구간 있어?", {"routeName": "경부선"}),
|
|
]
|
|
return [
|
|
{
|
|
"mode": "domain-query",
|
|
"feature": f,
|
|
"expected_intent": intent,
|
|
"question": q,
|
|
"params": params,
|
|
"note": "운영 경량 domain-query",
|
|
}
|
|
for f, intent, q, params in light
|
|
]
|
|
|
|
|
|
def extract_answer(body: Any) -> str:
|
|
if not isinstance(body, dict):
|
|
return one_line(str(body))
|
|
for key in ("answer", "message", "text", "statusMsg", "llmSummary", "formattedAnswer"):
|
|
val = body.get(key)
|
|
if isinstance(val, str) and val.strip():
|
|
return one_line(val)
|
|
for nest_key in ("domainData", "data", "rawData"):
|
|
nest = body.get(nest_key)
|
|
if isinstance(nest, dict):
|
|
for key in ("llmSummary", "statusMsg", "answer", "message"):
|
|
val = nest.get(key)
|
|
if isinstance(val, str) and val.strip():
|
|
return one_line(val)
|
|
# list summary
|
|
lst = nest.get("list") or nest.get("items")
|
|
if isinstance(lst, list) and lst:
|
|
return one_line(f"건수={len(lst)} 예시={json.dumps(lst[0], ensure_ascii=False)}")
|
|
if body.get("needsRestAreaForm") or body.get("needsIcTelForm"):
|
|
return one_line(str(body.get("statusMsg") or "폼 유도 응답"))
|
|
return one_line(json.dumps(body, ensure_ascii=False))
|
|
|
|
|
|
def intent_of(body: Any, fallback: str = "") -> str:
|
|
if not isinstance(body, dict):
|
|
return fallback
|
|
return str(body.get("intentType") or fallback)
|
|
|
|
|
|
def status_of(body: Any) -> str:
|
|
if not isinstance(body, dict):
|
|
return ""
|
|
if "status" in body:
|
|
return str(body.get("status"))
|
|
dd = body.get("domainData")
|
|
if isinstance(dd, dict) and "status" in dd:
|
|
return str(dd.get("status"))
|
|
return ""
|
|
|
|
|
|
INTENT_OK = {
|
|
"FARE_UNPAID": {"FARE_UNPAID"},
|
|
"FARE_REFUND": {"FARE_REFUND"},
|
|
"FARE_SEARCH": {"FARE_SEARCH"},
|
|
"TROAD_TRAVEL_TIME": {"TROAD_TRAVEL_TIME"},
|
|
"TROAD_BREAKING_NEWS": {"TROAD_BREAKING_NEWS", "TRAFFIC_NEWS"},
|
|
"TRAFFIC_NEWS": {"TRAFFIC_NEWS", "TROAD_BREAKING_NEWS"},
|
|
"CONSTRUCTION_PLAN": {"CONSTRUCTION_PLAN"},
|
|
"IC_TEL": {"IC_TEL"},
|
|
"EMERGENCY_HELP_ORG": {"EMERGENCY_HELP_ORG"},
|
|
"REST_AREA_FOOD_LIST": {"REST_AREA_FOOD_LIST"},
|
|
"BRAND_SHOP": {"BRAND_SHOP"},
|
|
"REST_AREA_FACILITY": {"REST_AREA_FACILITY"},
|
|
"GASSTATION": {"GASSTATION"},
|
|
"REST_AREA_CHARGING": {"REST_AREA_CHARGING"},
|
|
"REST_AREA_YOUTH_SHOP": {"REST_AREA_YOUTH_SHOP"},
|
|
"REST_AREA_NEAREST": {"REST_AREA_NEAREST"},
|
|
"TROAD_CONGESTION": {"TROAD_CONGESTION"},
|
|
}
|
|
|
|
|
|
def judge(expected: str, actual: str, status: str, answer: str, http_st: Optional[int]) -> Tuple[str, str]:
|
|
if http_st != 200:
|
|
return "FAIL", f"HTTP {http_st}"
|
|
natural = len(answer or "") >= 6 and "Traceback" not in (answer or "")
|
|
intent_ok = (actual or "") in INTENT_OK.get(expected, {expected}) or actual == expected
|
|
st_true = status in ("True", "true", "1")
|
|
infra = any(
|
|
h in (answer or "")
|
|
for h in ("일시적인 장애", "Failed to obtain", "Listener refused", "연결")
|
|
)
|
|
if intent_ok and st_true and natural:
|
|
return "PASS", "의도·응답 정상"
|
|
if intent_ok and natural and infra:
|
|
return "PARTIAL", "의도 일치, 인프라/DB 제약 안내"
|
|
if intent_ok and natural:
|
|
return "PARTIAL", "의도 일치, 데이터없음/안내성 응답"
|
|
if natural and actual:
|
|
return "PARTIAL", f"의도상이 expected={expected} actual={actual}"
|
|
if natural:
|
|
return "PARTIAL", "응답만 존재"
|
|
return "FAIL", f"의도/응답 불량 expected={expected} actual={actual or '-'}"
|
|
|
|
|
|
# ---- case builders ----
|
|
|
|
def domain_cases() -> List[Dict[str, Any]]:
|
|
"""간편조회 domain-query — 질문은 구어체, 호출은 폼/직조회 params."""
|
|
rows: List[Dict[str, Any]] = []
|
|
|
|
def add(feature: str, intent: str, question: str, params: Dict[str, str], note: str = "") -> None:
|
|
rows.append(
|
|
{
|
|
"mode": "domain-query",
|
|
"feature": feature,
|
|
"expected_intent": intent,
|
|
"question": question,
|
|
"params": params,
|
|
"note": note,
|
|
}
|
|
)
|
|
|
|
# 미납/환불
|
|
add("미납조회", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?", {"carNo": "12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("미납조회", "FARE_UNPAID", "하이패스 미납금 좀 봐줘 서울12가3456", {"carNo": "서울12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("미납조회", "FARE_UNPAID", "톨비 밀린 거 있어요? 차량번호 123가4567요", {"carNo": "123가4567"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("미납조회", "FARE_UNPAID", "98나1234 미납통행료 나와요?", {"carNo": "98나1234"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("환불조회", "FARE_REFUND", "12가3456으로 환불받을 돈 있어요?", {"carNo": "12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("환불조회", "FARE_REFUND", "잘못 낸 톨비 환불 내역 좀 확인해줘 서울12가3456", {"carNo": "서울12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("환불조회", "FARE_REFUND", "과오납 환불금 나왔나? 123가4567", {"carNo": "123가4567"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
add("환불조회", "FARE_REFUND", "하이패스 환불받을 거 있는지 봐줘 98나1234", {"carNo": "98나1234"}, "Oracle 미연결 시 PARTIAL 가능")
|
|
|
|
# 통행요금
|
|
fare_qs = [
|
|
("서울에서 부산까지 톨비 얼마 나와?", "서울", "부산"),
|
|
("판교에서 신갈 가면 요금이 어떻게 돼?", "판교", "신갈"),
|
|
("판교IC에서 신갈IC까지 하이패스 얼마야?", "판교IC", "신갈IC"),
|
|
("판교 영업소에서 수원까지 통행료 좀", "판교영업소", "수원"),
|
|
("대전에서 대구 가는 데 톨비 얼마쯤 해?", "대전", "대구"),
|
|
("천안에서 남이IC까지 요금 알려줄래?", "천안", "남이"),
|
|
("서울 톨게이트에서 기흥까지 얼마야?", "서울영업소", "기흥"),
|
|
("김포에서 인천 쪽 고속도로 요금 궁금해", "김포", "인천"),
|
|
]
|
|
for q, a, b in fare_qs:
|
|
add("구간통행요금", "FARE_SEARCH", q, {"fromIc": a, "toIc": b})
|
|
|
|
# 통행시간
|
|
time_qs = [
|
|
("서울에서 부산까지 지금 가면 얼마나 걸려?", "서울", "부산"),
|
|
("판교에서 신갈 지금 막혀? 몇 분 걸려?", "판교", "신갈"),
|
|
("대전에서 대구까지 지금 소요시간 어때?", "대전", "대구"),
|
|
("서울IC에서 부산IC 지금 가면 몇 시간이야?", "서울IC", "부산IC"),
|
|
("기흥에서 천안까지 지금 얼마나 걸릴까?", "기흥", "천안"),
|
|
]
|
|
for q, a, b in time_qs:
|
|
add("구간통행시간", "TROAD_TRAVEL_TIME", q, {"fromIc": a, "toIc": b, "carType": "1"})
|
|
|
|
# 차단계획
|
|
for q, route in [
|
|
("경부선 오늘 공사해서 막히는 데 있어?", "경부선"),
|
|
("영동선 차로 막는 일정 있나?", "영동선"),
|
|
("서해안선 공사구간 좀 알려줘", "서해안선"),
|
|
("호남선 이번 주 차단계획 있어?", "호남선"),
|
|
("경부고속도로 공사한다던데 어디야?", "경부고속도로"),
|
|
("중부내륙선 차로통제 있어?", "중부내륙선"),
|
|
]:
|
|
add("차단계획", "CONSTRUCTION_PLAN", q, {"routeName": route})
|
|
|
|
# 영업소
|
|
for q, ic in [
|
|
("판교 톨게이트 전화번호 좀", "판교"),
|
|
("판교IC 전화 어떻게 걸어?", "판교IC"),
|
|
("판교 영업소 연락처 알려줘", "판교영업소"),
|
|
("서울 영업소 번호 뭐야?", "서울"),
|
|
("기흥 톨게이트 주소랑 전화 좀", "기흥"),
|
|
("부산IC 영업소 정보 있어?", "부산IC"),
|
|
("신갈 영업소 전화 받나?", "신갈"),
|
|
("경부선 판교 쪽 영업소 번호 알려줘", "경부선판교"),
|
|
]:
|
|
add("영업소정보", "IC_TEL", q, {"icName": ic})
|
|
|
|
# 긴급구조
|
|
for q, route in [
|
|
("경부선에서 사고 나면 어디로 연락해?", "경부선"),
|
|
("영동선 긴급구조 가능한 데 어디야?", "영동선"),
|
|
("서해안선 긴급구조기관 좀 알려줘", "서해안선"),
|
|
]:
|
|
add("긴급구조기관", "EMERGENCY_HELP_ORG", q, {"listMode": "true", "routeName": route})
|
|
add("긴급구조기관", "EMERGENCY_HELP_ORG", "판교 근처 고속도로 긴급구조대 있어?", {"icName": "판교"})
|
|
|
|
# 휴게소
|
|
rest_sets = [
|
|
("죽전", "죽전휴게소에서 뭐 먹을 거 있어?", "죽전에 편의점이나 카페 있어?",
|
|
"죽전휴게소에 수유실 있나?", "죽전휴게소 기름값 얼마야?",
|
|
"죽전에서 전기차 충전돼?", "죽전휴게소에 청년들 장사하는 가게 있어?"),
|
|
("안성", "안성휴게소 밥 뭐 파나?", "안성휴게소에 브랜드 매장 뭐 있어?",
|
|
"안성휴게소 샤워실 쓸 수 있어?", "안성 휘발유 얼마야?",
|
|
"안성휴게소 전기차 충전소 있어?", "안성에 청년창업 매장 있어?"),
|
|
("김천", "김천휴게소에서 뭐 먹지?", "김천휴게소 입점 매장 뭐 있어?",
|
|
"김천휴게소 편의시설 뭐 있어?", "김천 경유값 어때?",
|
|
"김천휴게소에 수소차 충전해?", "김천에 청년몰 있어?"),
|
|
("금강", "금강휴게소 메뉴 추천해줘", "금강휴게소에 카페 있어?",
|
|
"금강휴게소 시설 뭐 있어?", "금강휴게소 주유소 있어?",
|
|
"금강에서 전기차 충전 가능해?", "금강휴게소 청년창업매장 있나?"),
|
|
("죽전휴게소", "죽전 휴게소 식당 뭐 있어?", "죽전 휴게소에 쇼핑할 데 있어?",
|
|
"죽전 휴게소에 애완동물 시설 있어?", "죽전 휴게소 기름 넣어?",
|
|
"죽전 휴게소 EV 충전돼?", "죽전 휴게소 청년가게 있어?"),
|
|
("강릉", "강릉휴게소에서 뭐 먹을까?", "강릉휴게소 매장 뭐 있어?",
|
|
"강릉휴게소 편의시설 알려줘", "강릉휴게소 주유소 유가 궁금해",
|
|
"강릉휴게소 충전소 있어?", "강릉에 청년창업 매장 있어?"),
|
|
]
|
|
for name, q_food, q_brand, q_fac, q_gas, q_chg, q_youth in rest_sets:
|
|
add("휴게소음식", "REST_AREA_FOOD_LIST", q_food, {"restAreaName": name})
|
|
add("휴게소브랜드매장", "BRAND_SHOP", q_brand, {"restAreaName": name})
|
|
add("휴게소시설", "REST_AREA_FACILITY", q_fac, {"restAreaName": name})
|
|
add("휴게소주유", "GASSTATION", q_gas, {"restAreaName": name})
|
|
add("휴게소충전소", "REST_AREA_CHARGING", q_chg, {"restAreaName": name, "chargingType": "all"})
|
|
add("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", q_youth, {"restAreaName": name})
|
|
|
|
add("휴게소주유", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?",
|
|
{"compareMode": "cheapest", "fuelType": "휘발유", "routeName": "경부선"})
|
|
add("휴게소주유", "GASSTATION", "고속도로에서 경유 제일 싼 곳 알려줘",
|
|
{"compareMode": "cheapest", "fuelType": "경유"})
|
|
add("휴게소충전소", "REST_AREA_CHARGING", "경부선에 전기차 충전되는 휴게소 어디 있어?",
|
|
{"listMode": "true", "routeName": "경부선", "chargingType": "electric"})
|
|
add("휴게소충전소", "REST_AREA_CHARGING", "강릉휴게소충전소 있어?",
|
|
{"restAreaName": "강릉휴게소충전소", "chargingType": "all"})
|
|
|
|
for q, ic in [
|
|
("판교 근처 휴게소 어디 있어?", "판교"),
|
|
("서울 톨게이트 지나서 바로 나오는 휴게소 어디야?", "서울"),
|
|
("기흥IC 근처에 쉴 데 있어?", "기흥IC"),
|
|
]:
|
|
add("가까운휴게소", "REST_AREA_NEAREST", q, {"icName": ic})
|
|
|
|
for q, route in [
|
|
("경부선 지금 막히는 구간 있어?", "경부선"),
|
|
("영동선 어디가 막혀?", "영동선"),
|
|
("서해안선 정체 심한 데 알려줘", "서해안선"),
|
|
]:
|
|
add("노선혼잡", "TROAD_CONGESTION", q, {"routeName": route})
|
|
|
|
return rows
|
|
|
|
|
|
def ask_cases() -> List[Dict[str, Any]]:
|
|
"""자연어 /ask — 일반 이용자 구어체 질문."""
|
|
pairs = [
|
|
("미납조회", "FARE_UNPAID", [
|
|
"야 내 차 12가3456인데 미납 있어?",
|
|
"하이패스 미납금 나왔나 봐줘 서울12가3456",
|
|
"톨비 밀린 거 있어요? 45다6789요",
|
|
]),
|
|
("환불조회", "FARE_REFUND", [
|
|
"잘못 낸 톨비 환불받을 수 있어? 12가3456",
|
|
"하이패스 환불금 들어왔나? 서울33하1122",
|
|
]),
|
|
("구간통행요금", "FARE_SEARCH", [
|
|
"서울에서 부산까지 톨비 얼마 나와?",
|
|
"판교에서 신갈 가면 요금이 어떻게 돼?",
|
|
"판교IC에서 신갈IC까지 하이패스 얼마야?",
|
|
"대전에서 대구 가는 데 톨비 얼마쯤 해?",
|
|
]),
|
|
("구간통행시간", "TROAD_TRAVEL_TIME", [
|
|
"서울에서 부산까지 지금 가면 얼마나 걸려?",
|
|
"판교에서 신갈 지금 막혀? 몇 분 걸려?",
|
|
"지금 대전에서 대구까지 가면 몇 시간이야?",
|
|
]),
|
|
("노선별교통속보", "TROAD_BREAKING_NEWS", [
|
|
"고속도로에 사고 났다는데 어디야?",
|
|
"경부선에 사고 났어?",
|
|
"영동선 지금 교통상황 어때?",
|
|
]),
|
|
("차단계획", "CONSTRUCTION_PLAN", [
|
|
"경부선 오늘 공사해서 막히는 데 있어?",
|
|
"영동선 차로 막는 일정 있나?",
|
|
"서해안선 공사구간 좀 알려줘",
|
|
]),
|
|
("영업소정보", "IC_TEL", [
|
|
"판교 톨게이트 전화번호 좀",
|
|
"기흥 톨게이트 주소랑 전화 좀",
|
|
"판교 영업소 연락처 알려줘",
|
|
]),
|
|
("긴급구조기관", "EMERGENCY_HELP_ORG", [
|
|
"경부선에서 사고 나면 어디로 연락해?",
|
|
"고속도로에서 사고 났는데 가까운 구조대 어디야?",
|
|
]),
|
|
("휴게소음식", "REST_AREA_FOOD_LIST", [
|
|
"죽전휴게소에서 뭐 먹을 거 있어?",
|
|
"안성휴게소 밥 뭐 파나?",
|
|
]),
|
|
("휴게소브랜드매장", "BRAND_SHOP", [
|
|
"죽전에 편의점이나 카페 있어?",
|
|
"김천휴게소에 브랜드 매장 뭐 있어?",
|
|
]),
|
|
("휴게소시설", "REST_AREA_FACILITY", [
|
|
"죽전휴게소에 수유실 있나?",
|
|
"안성휴게소 샤워실 쓸 수 있어?",
|
|
]),
|
|
("휴게소주유", "GASSTATION", [
|
|
"죽전휴게소 기름값 얼마야?",
|
|
"경부선에서 기름 제일 싼 데 어디야?",
|
|
"고속도로에서 휘발유 제일 싼 곳 알려줘",
|
|
]),
|
|
("휴게소충전소", "REST_AREA_CHARGING", [
|
|
"죽전에서 전기차 충전돼?",
|
|
"강릉휴게소에 충전소 있어?",
|
|
"김천휴게소에서 수소차 충전해?",
|
|
]),
|
|
("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", [
|
|
"죽전휴게소에 청년들 장사하는 가게 있어?",
|
|
"고속도로에 청년창업 매장 있는 휴게소 어디야?",
|
|
]),
|
|
("가까운휴게소", "REST_AREA_NEAREST", [
|
|
"판교 근처 휴게소 어디 있어?",
|
|
"서울 톨게이트 지나서 바로 나오는 휴게소 어디야?",
|
|
]),
|
|
("노선혼잡", "TROAD_CONGESTION", [
|
|
"경부선 지금 막히는 구간 있어?",
|
|
"영동선 어디가 막혀?",
|
|
]),
|
|
]
|
|
rows: List[Dict[str, Any]] = []
|
|
for feature, intent, qs in pairs:
|
|
for q in qs:
|
|
rows.append(
|
|
{
|
|
"mode": "ask",
|
|
"feature": feature,
|
|
"expected_intent": intent,
|
|
"question": q,
|
|
"params": {},
|
|
"note": "구어체 자연어 채팅",
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
FIELDNAMES = [
|
|
"tested_at",
|
|
"mode",
|
|
"feature",
|
|
"expected_intent",
|
|
"question",
|
|
"params",
|
|
"http_status",
|
|
"elapsed_sec",
|
|
"actual_intent",
|
|
"api_status",
|
|
"answer",
|
|
"result",
|
|
"reason",
|
|
"route_type",
|
|
"note",
|
|
]
|
|
|
|
|
|
def append_row(path: str, row: Dict[str, str], write_header: bool) -> None:
|
|
with open(path, "a", newline="", encoding="utf-8-sig") as f:
|
|
w = csv.DictWriter(f, fieldnames=FIELDNAMES)
|
|
if write_header:
|
|
w.writeheader()
|
|
w.writerow(row)
|
|
f.flush()
|
|
|
|
|
|
def main() -> int:
|
|
global _SSL_CTX
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--base", default="http://127.0.0.1:8083/web")
|
|
ap.add_argument("--out", default="test.csv")
|
|
ap.add_argument("--timeout-domain", type=float, default=45.0)
|
|
ap.add_argument("--timeout-ask", type=float, default=90.0)
|
|
ap.add_argument("--sleep", type=float, default=0.15)
|
|
ap.add_argument("--sleep-ask", type=float, default=0.0, help="ask 전용 sleep(0이면 --sleep)")
|
|
ap.add_argument("--skip-ask", action="store_true")
|
|
ap.add_argument("--skip-domain", action="store_true")
|
|
ap.add_argument("--limit", type=int, default=0)
|
|
ap.add_argument(
|
|
"--profile",
|
|
choices=["full", "prod"],
|
|
default="full",
|
|
help="prod=기능당 소수 + 구어체(운영 부하 최소화)",
|
|
)
|
|
ap.add_argument("--insecure", action="store_true", help="SSL 검증 비활성화")
|
|
args = ap.parse_args()
|
|
|
|
if args.insecure:
|
|
_SSL_CTX = ssl._create_unverified_context()
|
|
|
|
base = args.base.rstrip("/")
|
|
domain_url = base + "/domain-query"
|
|
ask_url = base + "/ask"
|
|
traffic_url = base + "/traffic-news"
|
|
started = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
sleep_ask = args.sleep_ask if args.sleep_ask > 0 else max(args.sleep, 1.5)
|
|
|
|
# reset out
|
|
open(args.out, "w", encoding="utf-8-sig").close()
|
|
header_needed = True
|
|
results: List[str] = []
|
|
|
|
def emit(row: Dict[str, str]) -> None:
|
|
nonlocal header_needed
|
|
append_row(args.out, row, header_needed)
|
|
header_needed = False
|
|
results.append(row["result"])
|
|
print(
|
|
f"{row['result']:7} [{row['mode']:12}] {row['feature']:14} "
|
|
f"{row['expected_intent']:22} → {row['actual_intent'] or '-':22} | {row['question'][:42]}",
|
|
flush=True,
|
|
)
|
|
|
|
# traffic-news GET
|
|
http_st, body, elapsed = get_json(traffic_url, args.timeout_domain)
|
|
ans = extract_answer(body) if isinstance(body, (dict, list)) else one_line(str(body))
|
|
if isinstance(body, list):
|
|
ans = one_line(f"list={len(body)}")
|
|
ok = http_st == 200
|
|
else:
|
|
ok = http_st == 200 and (
|
|
(
|
|
isinstance(body, dict)
|
|
and (
|
|
body.get("status") is True
|
|
or "list" in body
|
|
or "items" in body
|
|
or "routes" in body
|
|
)
|
|
)
|
|
or isinstance(body, list)
|
|
)
|
|
emit(
|
|
{
|
|
"tested_at": started,
|
|
"mode": "domain-api",
|
|
"feature": "노선별교통속보",
|
|
"expected_intent": "TRAFFIC_NEWS",
|
|
"question": "오늘 고속도로에 사고나 통제 소식 있어?",
|
|
"params": "",
|
|
"http_status": str(http_st or ""),
|
|
"elapsed_sec": f"{elapsed:.2f}",
|
|
"actual_intent": "TRAFFIC_NEWS",
|
|
"api_status": "True" if ok else "False",
|
|
"answer": ans,
|
|
"result": "PASS" if ok else "FAIL",
|
|
"reason": "GET /traffic-news" if ok else "GET 실패",
|
|
"route_type": "GET",
|
|
"note": "간편조회 UI 전용",
|
|
}
|
|
)
|
|
|
|
suite: List[Dict[str, Any]] = []
|
|
if not args.skip_domain:
|
|
suite.extend(domain_cases_prod_light() if args.profile == "prod" else domain_cases())
|
|
if not args.skip_ask:
|
|
# prod: 기능당 질문 최대 2개로 축소
|
|
ask = ask_cases()
|
|
if args.profile == "prod":
|
|
by_feat: Dict[str, int] = {}
|
|
trimmed: List[Dict[str, Any]] = []
|
|
for c in ask:
|
|
n = by_feat.get(c["feature"], 0)
|
|
if n < 2:
|
|
trimmed.append(c)
|
|
by_feat[c["feature"]] = n + 1
|
|
ask = trimmed
|
|
suite.extend(ask)
|
|
if args.limit > 0:
|
|
suite = suite[: args.limit]
|
|
|
|
total = len(suite)
|
|
for i, case in enumerate(suite, 1):
|
|
mode = case["mode"]
|
|
print(f"({i}/{total}) …", flush=True)
|
|
if mode == "domain-query":
|
|
http_st, body, elapsed = post_json(
|
|
domain_url,
|
|
{
|
|
"intentType": case["expected_intent"],
|
|
"params": case["params"],
|
|
"botId": f"qi-dom-{i}",
|
|
},
|
|
args.timeout_domain,
|
|
)
|
|
actual = intent_of(body, case["expected_intent"])
|
|
# domain-query often echoes intent in request; keep expected if missing
|
|
if not actual:
|
|
actual = case["expected_intent"]
|
|
else:
|
|
http_st, body, elapsed = post_json(
|
|
ask_url,
|
|
{"query": case["question"], "botId": f"qi-ask-{i}"},
|
|
args.timeout_ask,
|
|
)
|
|
actual = intent_of(body)
|
|
|
|
status = status_of(body)
|
|
answer = extract_answer(body)
|
|
route_type = ""
|
|
if isinstance(body, dict):
|
|
route_type = str(body.get("routeType") or mode)
|
|
result, reason = judge(case["expected_intent"], actual, status, answer, http_st)
|
|
emit(
|
|
{
|
|
"tested_at": started,
|
|
"mode": mode,
|
|
"feature": case["feature"],
|
|
"expected_intent": case["expected_intent"],
|
|
"question": case["question"],
|
|
"params": json.dumps(case.get("params") or {}, ensure_ascii=False),
|
|
"http_status": str(http_st or ""),
|
|
"elapsed_sec": f"{elapsed:.2f}",
|
|
"actual_intent": actual,
|
|
"api_status": status,
|
|
"answer": answer,
|
|
"result": result,
|
|
"reason": reason,
|
|
"route_type": route_type,
|
|
"note": case.get("note") or "",
|
|
}
|
|
)
|
|
time.sleep(sleep_ask if mode == "ask" else args.sleep)
|
|
|
|
c = Counter(results)
|
|
print("---", flush=True)
|
|
print(f"saved={args.out} total={len(results)} {dict(c)}", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|