Deploy: exdev sync 2026-08-18 09:26:35

This commit is contained in:
Macbook
2026-08-18 09:26:35 +09:00
parent 88221b828e
commit 8d0a8bdb5c
8 changed files with 1310 additions and 158 deletions
+526
View File
@@ -0,0 +1,526 @@
#!/usr/bin/env python3
"""간편조회 파라미터 분기(폼/부분선택/상세/위치/별칭) 구어체 테스트 → test.csv
운영 예:
PYTHONUNBUFFERED=1 python3 scripts/quick_inquiry_branch_test.py \\
--base https://chatbot.ex.co.kr/chatbot --insecure \\
--out test.csv --sleep-ask 2.0 --sleep-domain 1.2
"""
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
_SSL_CTX = ssl.create_default_context()
FIELDNAMES = [
"tested_at",
"feature",
"param_tier", # none | partial | full | alias | location
"expected_mode", # form | detail | clarify | list | location
"expected_intent",
"mode", # ask | domain-query
"question",
"params",
"http_status",
"elapsed_sec",
"actual_intent",
"actual_mode",
"api_status",
"route_type",
"form_flags",
"answer",
"result",
"reason",
"note",
]
def one_line(text: str, limit: int = 260) -> 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 extract_answer(body: Any) -> str:
if not isinstance(body, dict):
return one_line(str(body))
for key in ("answer", "message", "text", "statusMsg", "llmSummary"):
val = body.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
dd = body.get("domainData")
if isinstance(dd, dict):
for key in ("llmSummary", "statusMsg", "answer"):
val = dd.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
return one_line(json.dumps(body, ensure_ascii=False))
def form_flags(body: Any) -> str:
flags = []
if not isinstance(body, dict):
return ""
dd = body.get("domainData") if isinstance(body.get("domainData"), dict) else {}
raw = body.get("rawData") if isinstance(body.get("rawData"), dict) else {}
raw_dd = raw.get("domainData") if isinstance(raw.get("domainData"), dict) else {}
for src in (body, dd, raw, raw_dd):
for k, v in src.items():
if str(k).startswith("needs") and v:
flags.append(f"{k}={v}")
if src.get("needsClarification"):
flags.append("needsClarification=True")
if src.get("listMode"):
flags.append(f"listMode={src.get('listMode')}")
# dedupe
seen = set()
out = []
for f in flags:
if f not in seen:
seen.add(f)
out.append(f)
return ";".join(out)
def detect_mode(body: Any, answer: str) -> str:
"""응답을 form/detail/clarify/list/location 으로 분류."""
if not isinstance(body, dict):
return "unknown"
flags = form_flags(body)
route = str(body.get("routeType") or "")
dd = body.get("domainData") if isinstance(body.get("domainData"), dict) else {}
ans = answer or ""
# 구체 데이터(전화·요금 등)가 있으면 form 플래그보다 detail 우선
if re.search(r"0\d{1,2}-\d{3,4}-\d{4}", ans) or "연락처입니다" in ans:
return "detail"
if route == "clarify" or "needsClarification=True" in flags:
return "clarify"
if "위치정보" in ans or ("동의" in ans and "위치" in ans):
return "location"
if any(p in ans for p in ("선택하면", "선택해 주세요", "알려주세요", "어느 노선", "어느 방향", "차량번호가 필요")):
if any(x in flags for x in ("needsRestAreaForm=True", "needsIcTelForm=True")):
return "form"
return "clarify"
if any(x in flags for x in ("needsRestAreaForm=True", "needsIcTelForm=True")):
return "form"
lst = dd.get("list") or dd.get("items")
if isinstance(lst, list) and len(lst) > 0:
return "list"
if any(
k in ans
for k in ("전화", "원", "km/h", "메뉴", "유가", "충전", "구조", "공사", "정체", "서행", "주소")
):
return "detail"
if body.get("status") is True and len(ans) > 15:
return "detail"
return "unknown"
def scenarios() -> List[Dict[str, Any]]:
"""기능 × 파라미터 단계 × 구어체/별칭/위치."""
rows: List[Dict[str, Any]] = []
def ask(
feature: str,
tier: str,
expected_mode: str,
intent: str,
question: str,
note: str = "",
) -> None:
rows.append(
{
"mode": "ask",
"feature": feature,
"param_tier": tier,
"expected_mode": expected_mode,
"expected_intent": intent,
"question": question,
"params": {},
"note": note,
}
)
def domain(
feature: str,
tier: str,
expected_mode: str,
intent: str,
question: str,
params: Dict[str, str],
note: str = "",
) -> None:
rows.append(
{
"mode": "domain-query",
"feature": feature,
"param_tier": tier,
"expected_mode": expected_mode,
"expected_intent": intent,
"question": question,
"params": params,
"note": note,
}
)
# ---- 정체/혼잡 ----
ask("노선혼잡", "none", "clarify", "TROAD_CONGESTION", "고속도로 정체구간 알려줘")
ask("노선혼잡", "none", "clarify", "TROAD_CONGESTION", "지금 어디가 막혀?")
ask("노선혼잡", "partial", "detail", "TROAD_CONGESTION", "경부선에 정체구간 있어?")
ask("노선혼잡", "partial", "detail", "TROAD_CONGESTION", "영동선 지금 막혀?")
ask("노선혼잡", "alias", "detail", "TROAD_CONGESTION", "경부고속도로 정체 어때?")
ask(
"노선혼잡",
"location",
"location",
"TROAD_CONGESTION",
"이 근처에 정체되는 구간 있어?",
"GPS 동의/근처 영업소 파생 가능",
)
domain(
"노선혼잡",
"full",
"detail",
"TROAD_CONGESTION",
"경부선 정체 상세",
{"routeName": "경부선"},
)
# ---- 영업소 정보 ----
ask("영업소정보", "none", "form", "IC_TEL", "영업소 전화번호 좀 알려줘")
ask("영업소정보", "full", "detail", "IC_TEL", "판교 톨게이트 전화번호 좀")
ask("영업소정보", "alias", "detail", "IC_TEL", "동김천 톨게이트 번호 뭐야?")
ask("영업소정보", "alias", "detail", "IC_TEL", "동김천ic 연락처 알려줘")
ask("영업소정보", "alias", "detail", "IC_TEL", "동감천 톨게이트 전화 있어?") # 오표기
ask("영업소정보", "alias", "detail", "IC_TEL", "판교영업소 대표번호")
ask(
"영업소정보",
"location",
"location",
"IC_TEL",
"여기 근처 영업소 전화번호 알려줘",
"위치기반 폼/상세",
)
domain("영업소정보", "none", "form", "IC_TEL", "영업소 조회 폼", {})
domain("영업소정보", "full", "detail", "IC_TEL", "동김천 상세", {"icName": "동김천"})
# ---- 통행요금 ----
ask("구간통행요금", "none", "clarify", "FARE_SEARCH", "톨비 얼마야?")
ask("구간통행요금", "partial", "clarify", "FARE_SEARCH", "서울에서 출발하면 요금 얼마야?")
ask("구간통행요금", "full", "detail", "FARE_SEARCH", "서울에서 부산까지 톨비 얼마 나와?")
ask("구간통행요금", "alias", "detail", "FARE_SEARCH", "판교IC에서 신갈 톨게이트까지 요금 어때?")
ask("구간통행요금", "alias", "detail", "FARE_SEARCH", "판교 영업소에서 수원까지 하이패스 얼마야?")
# ---- 통행시간 ----
ask("구간통행시간", "none", "clarify", "TROAD_TRAVEL_TIME", "지금 고속도로 얼마나 걸려?")
ask("구간통행시간", "full", "detail", "TROAD_TRAVEL_TIME", "서울에서 부산까지 지금 가면 얼마나 걸려?")
ask("구간통행시간", "alias", "detail", "TROAD_TRAVEL_TIME", "판교에서 신갈 지금 막혀? 몇 분 걸려?")
# ---- 교통속보 ----
ask("노선별교통속보", "none", "clarify", "TROAD_BREAKING_NEWS", "고속도로에 사고 났다는데 어디야?")
ask("노선별교통속보", "partial", "detail", "TROAD_BREAKING_NEWS", "경부선에 사고 났어?")
ask("노선별교통속보", "alias", "detail", "TROAD_BREAKING_NEWS", "영동고속도로 교통속보 있어?")
# ---- 차단계획 ----
ask("차단계획", "none", "clarify", "CONSTRUCTION_PLAN", "고속도로 공사하는 데 있어?")
ask("차단계획", "partial", "detail", "CONSTRUCTION_PLAN", "경부선 오늘 공사해서 막히는 데 있어?")
ask("차단계획", "alias", "detail", "CONSTRUCTION_PLAN", "경부고속도로 차로 막는 일정 있나?")
domain(
"차단계획",
"partial",
"detail",
"CONSTRUCTION_PLAN",
"영동선 차단계획",
{"routeName": "영동선"},
)
# ---- 긴급구조 ----
ask("긴급구조기관", "none", "form", "EMERGENCY_HELP_ORG", "고속도로에서 사고 나면 어디로 연락해?")
ask("긴급구조기관", "partial", "form", "EMERGENCY_HELP_ORG", "경부선 긴급구조기관 알려줘")
ask("긴급구조기관", "alias", "form", "EMERGENCY_HELP_ORG", "경부고속도로 긴급구조대 어디야?")
domain("긴급구조기관", "none", "form", "EMERGENCY_HELP_ORG", "긴급구조 폼", {})
domain(
"긴급구조기관",
"partial",
"form",
"EMERGENCY_HELP_ORG",
"경부선 선택 폼",
{"routeName": "경부선"},
)
# ---- 휴게소 음식/브랜드/시설 ----
ask("휴게소음식", "none", "clarify", "REST_AREA_FOOD_LIST", "휴게소에서 뭐 먹지?")
ask("휴게소음식", "full", "detail", "REST_AREA_FOOD_LIST", "죽전휴게소에서 뭐 먹을 거 있어?")
ask("휴게소음식", "alias", "detail", "REST_AREA_FOOD_LIST", "죽전 휴게소 밥 뭐 파나?")
ask("휴게소브랜드매장", "none", "clarify", "BRAND_SHOP", "휴게소에 카페 있어?")
ask("휴게소브랜드매장", "full", "detail", "BRAND_SHOP", "김천휴게소에 브랜드 매장 뭐 있어?")
ask("휴게소브랜드매장", "alias", "detail", "BRAND_SHOP", "안성 휴게소에 편의점 있나?")
ask("휴게소시설", "none", "clarify", "REST_AREA_FACILITY", "휴게소에 수유실 있어?")
ask("휴게소시설", "full", "detail", "REST_AREA_FACILITY", "죽전휴게소에 수유실 있나?")
ask("휴게소시설", "alias", "detail", "REST_AREA_FACILITY", "안성SA 샤워실 쓸 수 있어?")
# ---- 주유 ----
ask("휴게소주유", "none", "form", "GASSTATION", "고속도로 기름값 얼마야?")
ask("휴게소주유", "full", "detail", "GASSTATION", "죽전휴게소 기름값 얼마야?")
ask("휴게소주유", "partial", "detail", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?")
ask("휴게소주유", "alias", "detail", "GASSTATION", "안성 휴게소 휘발유값 어때?")
domain("휴게소주유", "none", "form", "GASSTATION", "주유 폼", {})
# ---- 충전 ----
ask("휴게소충전소", "none", "form", "REST_AREA_CHARGING", "고속도로에 전기차 충전소 있어?")
ask("휴게소충전소", "full", "detail", "REST_AREA_CHARGING", "죽전에서 전기차 충전돼?")
ask("휴게소충전소", "partial", "form", "REST_AREA_CHARGING", "경부선에 충전되는 휴게소 어디 있어?")
ask("휴게소충전소", "alias", "detail", "REST_AREA_CHARGING", "강릉휴게소충전소 있어?")
ask("휴게소충전소", "alias", "detail", "REST_AREA_CHARGING", "김천 휴게소에서 수소차 충전해?")
domain("휴게소충전소", "none", "form", "REST_AREA_CHARGING", "충전 폼", {})
domain(
"휴게소충전소",
"partial",
"form",
"REST_AREA_CHARGING",
"경부선 충전 폼",
{"routeName": "경부선"},
)
# ---- 청년창업 ----
ask("휴게소청년창업매장", "none", "form", "REST_AREA_YOUTH_SHOP", "고속도로에 청년창업 매장 있어?")
ask("휴게소청년창업매장", "full", "detail", "REST_AREA_YOUTH_SHOP", "죽전휴게소에 청년들 장사하는 가게 있어?")
ask(
"휴게소청년창업매장",
"alias",
"detail",
"REST_AREA_YOUTH_SHOP",
"안성 휴게소 청년몰 있나?",
)
# ---- 가까운 휴게소 ----
ask("가까운휴게소", "partial", "detail", "REST_AREA_NEAREST", "판교 근처 휴게소 어디 있어?")
ask(
"가까운휴게소",
"location",
"location",
"REST_AREA_NEAREST",
"여기 제일 가까운 휴게소 어디야?",
"GPS 경로",
)
ask("가까운휴게소", "alias", "detail", "REST_AREA_NEAREST", "기흥IC 근처에 쉴 데 있어?")
# ---- 미납/환불 (파라미터=차량번호) ----
ask("미납조회", "none", "clarify", "FARE_UNPAID", "미납요금 조회해줘")
ask("미납조회", "full", "detail", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?")
ask("환불조회", "none", "clarify", "FARE_REFUND", "톨비 환불받을 거 있어?")
ask("환불조회", "full", "detail", "FARE_REFUND", "하이패스 환불금 들어왔나? 서울33하1122")
return rows
def mode_match(expected: str, actual: str) -> bool:
if expected == actual:
return True
# form↔clarify 는 둘 다 '추가 선택 유도'로 허용
if {expected, actual} <= {"form", "clarify"}:
return True
# location 질문은 form/clarify/detail 도 허용(클라이언트 GPS 없을 때)
if expected == "location" and actual in {"location", "form", "clarify", "detail"}:
return True
# list 는 detail 상위 개념으로 허용
if expected == "detail" and actual in {"detail", "list"}:
return True
if expected == "list" and actual in {"list", "form", "detail"}:
return True
return False
def intent_ok(expected: str, actual: str) -> bool:
if not actual:
return False
if actual == expected:
return True
aliases = {
"TROAD_BREAKING_NEWS": {"TROAD_BREAKING_NEWS", "TRAFFIC_NEWS"},
"TROAD_CONGESTION": {"TROAD_CONGESTION"},
"BRAND_SHOP": {"BRAND_SHOP", "REST_AREA_FACILITY"}, # 편의점 질문은 시설로도 갈 수 있음
}
return actual in aliases.get(expected, {expected})
def judge(
expected_intent: str,
actual_intent: str,
expected_mode: str,
actual_mode: str,
http_st: Optional[int],
answer: str,
) -> Tuple[str, str]:
if http_st != 200:
return "FAIL", f"HTTP {http_st}"
natural = len(answer or "") >= 6
i_ok = intent_ok(expected_intent, actual_intent)
m_ok = mode_match(expected_mode, actual_mode)
if i_ok and m_ok and natural:
return "PASS", "의도·응답모드 적합"
if i_ok and natural:
return "PARTIAL", f"모드상이 expected={expected_mode} actual={actual_mode}"
if m_ok and natural:
return "PARTIAL", f"의도상이 expected={expected_intent} actual={actual_intent or '-'}"
if natural:
return "PARTIAL", "응답만 존재"
return "FAIL", "응답 불량"
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="https://chatbot.ex.co.kr/chatbot")
ap.add_argument("--out", default="test.csv")
ap.add_argument("--timeout-ask", type=float, default=90.0)
ap.add_argument("--timeout-domain", type=float, default=45.0)
ap.add_argument("--sleep-ask", type=float, default=2.0)
ap.add_argument("--sleep-domain", type=float, default=1.2)
ap.add_argument("--insecure", action="store_true")
ap.add_argument("--limit", type=int, default=0)
ap.add_argument("--skip-domain", action="store_true")
ap.add_argument("--skip-ask", action="store_true")
args = ap.parse_args()
if args.insecure:
_SSL_CTX = ssl._create_unverified_context()
base = args.base.rstrip("/")
ask_url = base + "/ask"
domain_url = base + "/domain-query"
started = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
open(args.out, "w", encoding="utf-8-sig").close()
header = True
results: List[str] = []
suite = scenarios()
if args.skip_ask:
suite = [c for c in suite if c["mode"] != "ask"]
if args.skip_domain:
suite = [c for c in suite if c["mode"] != "domain-query"]
if args.limit > 0:
suite = suite[: args.limit]
total = len(suite)
for i, case in enumerate(suite, 1):
print(f"({i}/{total}) {case['mode']} {case['feature']} [{case['param_tier']}] {case['question'][:40]}", flush=True)
if case["mode"] == "ask":
http_st, body, elapsed = post_json(
ask_url,
{"query": case["question"], "botId": f"branch-ask-{i}"},
args.timeout_ask,
)
sleep = args.sleep_ask
else:
http_st, body, elapsed = post_json(
domain_url,
{
"intentType": case["expected_intent"],
"params": case["params"],
"botId": f"branch-dom-{i}",
},
args.timeout_domain,
)
sleep = args.sleep_domain
answer = extract_answer(body)
actual_intent = ""
api_status = ""
route_type = ""
if isinstance(body, dict):
actual_intent = str(body.get("intentType") or "")
api_status = str(body.get("status"))
route_type = str(body.get("routeType") or "")
if case["mode"] == "domain-query" and not actual_intent:
actual_intent = case["expected_intent"]
actual_mode = detect_mode(body, answer)
result, reason = judge(
case["expected_intent"],
actual_intent,
case["expected_mode"],
actual_mode,
http_st,
answer,
)
row = {
"tested_at": started,
"feature": case["feature"],
"param_tier": case["param_tier"],
"expected_mode": case["expected_mode"],
"expected_intent": case["expected_intent"],
"mode": case["mode"],
"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_intent,
"actual_mode": actual_mode,
"api_status": api_status,
"route_type": route_type,
"form_flags": form_flags(body),
"answer": answer,
"result": result,
"reason": reason,
"note": case.get("note") or "",
}
append_row(args.out, row, header)
header = False
results.append(result)
print(f" → {result:7} intent={actual_intent or '-':22} mode={actual_mode:8} | {answer[:70]}", flush=True)
time.sleep(sleep)
print("---", flush=True)
print(f"saved={args.out} total={len(results)} {dict(Counter(results))}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+626
View File
@@ -0,0 +1,626 @@
#!/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())
+154 -154
View File
@@ -1,154 +1,154 @@
# exdev 통합 HTTPS (443) — stock / chatbot(Agent) / bizcare / nationcompany / github / apex
# Agent 배포 시 chatbot-only Caddyfile 로 덮어써 다른 서브도메인이 깨진 것을 복구
{
email {$CADDY_ACME_EMAIL:admin@exdev.co.kr}
auto_https disable_redirects
}
(tls_certs) {
tls /etc/caddy/certs/fullchain.pem /etc/caddy/certs/privkey.pem
}
(proxy_std) {
flush_interval -1
transport http {
read_timeout 120s
write_timeout 120s
versions 1.1 2
}
header_up Host {host}
header_up X-Real-IP {remote_host}
header_up X-Forwarded-For {remote_host}
header_up X-Forwarded-Proto {scheme}
}
# ── GoldenChart ───────────────────────────────────────────────────────────────
https://{$STOCK_DOMAIN:stock.exdev.co.kr} {
import tls_certs
handle {
reverse_proxy host.docker.internal:{$GOLDENCHART_HOST_PORT:80} {
import proxy_std
}
}
}
# ── Chatbot Agent 2.0 ─────────────────────────────────────────────────────────
https://{$CHATBOT_DOMAIN:chatbot.exdev.co.kr} {
import tls_certs
# 업스트림 nginx 가 HTTP 로 absolute redirect 해서 http:// 로 떨어지지 않게 Caddy 에서 처리
redir / /chatbot/ai/ 302
handle /agent/* {
reverse_proxy host.docker.internal:{$RAG_API_PORT:28012} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /chatbot/admin* {
uri strip_prefix /chatbot/admin
reverse_proxy host.docker.internal:{$ADMIN_PORT:28013} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /web/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
# 카카오 오픈빌더 Skill 웹훅 (/kakao/skill, /oakak/skill)
handle /kakao/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /oakak/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
# iOS 홈화면 Web Clip — MIME 를 Caddy에서 강제 (인앱/프록시에서 text 로 깨지는 것 방지)
handle /chatbot/ai/downloads/*.mobileconfig {
header Content-Type application/x-apple-aspen-config
header Content-Disposition inline
header Cache-Control "no-cache, no-store, must-revalidate"
reverse_proxy host.docker.internal:{$CHATBOT_APP_PORT:8092} {
import proxy_std
}
}
handle {
reverse_proxy host.docker.internal:{$CHATBOT_APP_PORT:8092} {
import proxy_std
}
}
}
# ── BizCare ───────────────────────────────────────────────────────────────────
https://{$BIZCARE_DOMAIN:bizcare.exdev.co.kr} {
tls /etc/caddy/certs/bizcare/fullchain.pem /etc/caddy/certs/bizcare/privkey.pem
handle {
reverse_proxy host.docker.internal:{$BIZCARE_HOST_PORT:83} {
import proxy_std
}
}
}
# ── ChannelONE (nationcompany) — TLS-ALPN ACME ────────────────────────────────
https://{$NATIONCOMPANY_DOMAIN:nationcompany.exdev.co.kr} {
handle {
reverse_proxy host.docker.internal:{$NATIONCOMPANY_HOST_PORT:84} {
import proxy_std
}
}
}
# ── Gitea (github.exdev.co.kr) ────────────────────────────────────────────────
https://{$GITHUB_DOMAIN:github.exdev.co.kr} {
tls /etc/caddy/certs/github/fullchain.pem /etc/caddy/certs/github/privkey.pem
handle {
reverse_proxy host.docker.internal:{$GITEA_HOST_PORT:3000} {
flush_interval -1
transport http {
read_timeout 3600s
write_timeout 3600s
versions 1.1 2
}
header_up Host {host}
header_up X-Real-IP {remote_host}
header_up X-Forwarded-For {remote_host}
header_up X-Forwarded-Proto {scheme}
}
}
}
# ── Legacy apex ───────────────────────────────────────────────────────────────
https://{$HTTPS_DOMAIN:exdev.co.kr} {
import tls_certs
@legacy_chatbot path /chatbot /chatbot/*
handle @legacy_chatbot {
redir https://{$CHATBOT_DOMAIN:chatbot.exdev.co.kr}{uri} permanent
}
handle {
redir https://{$STOCK_DOMAIN:stock.exdev.co.kr}{uri} permanent
}
}
# exdev 통합 HTTPS (443) — stock / chatbot(Agent) / bizcare / nationcompany / github / apex
# Agent 배포 시 chatbot-only Caddyfile 로 덮어써 다른 서브도메인이 깨진 것을 복구
{
email {$CADDY_ACME_EMAIL:admin@exdev.co.kr}
auto_https disable_redirects
}
(tls_certs) {
tls /etc/caddy/certs/fullchain.pem /etc/caddy/certs/privkey.pem
}
(proxy_std) {
flush_interval -1
transport http {
read_timeout 120s
write_timeout 120s
versions 1.1 2
}
header_up Host {host}
header_up X-Real-IP {remote_host}
header_up X-Forwarded-For {remote_host}
header_up X-Forwarded-Proto {scheme}
}
# ── GoldenChart ───────────────────────────────────────────────────────────────
https://{$STOCK_DOMAIN:stock.exdev.co.kr} {
import tls_certs
handle {
reverse_proxy host.docker.internal:{$GOLDENCHART_HOST_PORT:80} {
import proxy_std
}
}
}
# ── Chatbot Agent 2.0 ─────────────────────────────────────────────────────────
https://{$CHATBOT_DOMAIN:chatbot.exdev.co.kr} {
import tls_certs
# 업스트림 nginx 가 HTTP 로 absolute redirect 해서 http:// 로 떨어지지 않게 Caddy 에서 처리
redir / /chatbot/ai/ 302
handle /agent/* {
reverse_proxy host.docker.internal:{$RAG_API_PORT:28012} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /chatbot/admin* {
uri strip_prefix /chatbot/admin
reverse_proxy host.docker.internal:{$ADMIN_PORT:28013} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /web/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
# 카카오 오픈빌더 Skill 웹훅 (/kakao/skill, /oakak/skill)
handle /kakao/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
handle /oakak/* {
reverse_proxy host.docker.internal:{$SKILL_PORT:8083} {
transport http {
read_timeout 120s
write_timeout 120s
}
}
}
# iOS 홈화면 Web Clip — MIME 를 Caddy에서 강제 (인앱/프록시에서 text 로 깨지는 것 방지)
handle /chatbot/ai/downloads/*.mobileconfig {
header Content-Type application/x-apple-aspen-config
header Content-Disposition inline
header Cache-Control "no-cache, no-store, must-revalidate"
reverse_proxy host.docker.internal:{$CHATBOT_APP_PORT:8092} {
import proxy_std
}
}
handle {
reverse_proxy host.docker.internal:{$CHATBOT_APP_PORT:8092} {
import proxy_std
}
}
}
# ── BizCare ───────────────────────────────────────────────────────────────────
https://{$BIZCARE_DOMAIN:bizcare.exdev.co.kr} {
tls /etc/caddy/certs/bizcare/fullchain.pem /etc/caddy/certs/bizcare/privkey.pem
handle {
reverse_proxy host.docker.internal:{$BIZCARE_HOST_PORT:83} {
import proxy_std
}
}
}
# ── ChannelONE (nationcompany) — TLS-ALPN ACME ────────────────────────────────
https://{$NATIONCOMPANY_DOMAIN:nationcompany.exdev.co.kr} {
handle {
reverse_proxy host.docker.internal:{$NATIONCOMPANY_HOST_PORT:84} {
import proxy_std
}
}
}
# ── Gitea (github.exdev.co.kr) ────────────────────────────────────────────────
https://{$GITHUB_DOMAIN:github.exdev.co.kr} {
tls /etc/caddy/certs/github/fullchain.pem /etc/caddy/certs/github/privkey.pem
handle {
reverse_proxy host.docker.internal:{$GITEA_HOST_PORT:3000} {
flush_interval -1
transport http {
read_timeout 3600s
write_timeout 3600s
versions 1.1 2
}
header_up Host {host}
header_up X-Real-IP {remote_host}
header_up X-Forwarded-For {remote_host}
header_up X-Forwarded-Proto {scheme}
}
}
}
# ── Legacy apex ───────────────────────────────────────────────────────────────
https://{$HTTPS_DOMAIN:exdev.co.kr} {
import tls_certs
@legacy_chatbot path /chatbot /chatbot/*
handle @legacy_chatbot {
redir https://{$CHATBOT_DOMAIN:chatbot.exdev.co.kr}{uri} permanent
}
handle {
redir https://{$STOCK_DOMAIN:stock.exdev.co.kr}{uri} permanent
}
}