From d11f69fad1e9e583d410fb4565d5c6e17f242e85 Mon Sep 17 00:00:00 2001 From: digitie <964189+digitie@users.noreply.github.com> Date: Sat, 29 Aug 2026 16:19:18 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20Streamlit=20=EB=94=94=EB=B2=84=EA=B7=B8?= =?UTF-8?q?=20UI=EB=A5=BC=20examples/streamlit=5Fdebug=5Fui.py=EB=A1=9C=20?= =?UTF-8?q?=ED=86=B5=ED=95=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit khoa의 streamlit_debug_ui.py 템플릿에 맞춰 tools/debug_streamlit.py(780줄, operation별 하드코딩 분기, Fixture 탭 미구현, APIHub 470개 endpoint 전부 실행 불가)를 새로 짰다. - src/kma/debug.py 신설: DebugRun/jsonable/redact_sensitive/debug_error/ save_fixture — khoa의 debug.py 구조를 그대로 따른다. - ApiCatalogEntry에 required_params/optional_params/param_defaults/ response_kind/endpoint_path 메타데이터 필드를 추가하고, catalog.py에 apihub_endpoint_catalog()를 새로 만들어 APIHub 470개 endpoint를 실제 호출 가능한 카탈로그 row로 노출했다(기존 api_catalog()는 하위 호환 유지, 테스트 불변 확인됨). - DataGoKrClient.debug_fetch()/ApiHubClient.debug_fetch_endpoint() 제네릭 메서드 추가 — endpoint별 분기 없이 카탈로그가 넘겨주는 service/operation 또는 ApiHubEndpointSpec만으로 요청을 라우팅하고 구조화된 DebugRun을 반환한다(예외도 traceback 포함 구조화 dict로, redact_sensitive 통과). - examples/streamlit_debug_ui.py: Data source -> Category -> API 3단 계단식 사이드바, st.form() 기반 파라미터 폼을 카탈로그 메타데이터에서 자동 생성(하드코딩 if function_name== 분기 없음), 고정 6탭(Raw Response/ Pydantic Model/Processed Result/Validation Errors/Debug Trace/ Fixture-Testcase), 세션 상태는 gateway:dataset_id:service:operation으로 스코프해 데이터소스 전환 시 이전 결과가 섞이지 않게 했다. Fixture 탭은 save_fixture()를 실제로 호출해 파일을 저장한다. - pyproject.toml debug-ui extra에 pandas>=2 추가. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XF9V2q4mAmhmXn6t5G9Hfe --- README.md | 17 +- examples/streamlit_debug_ui.py | 665 ++++++++++++++++++++++++++++ pyproject.toml | 1 + src/kma/__init__.py | 21 +- src/kma/apihub.py | 125 ++++++ src/kma/catalog.py | 236 +++++++++- src/kma/datagokr.py | 126 ++++++ src/kma/debug.py | 168 +++++++ tools/debug_streamlit.py | 780 --------------------------------- 9 files changed, 1353 insertions(+), 786 deletions(-) create mode 100644 examples/streamlit_debug_ui.py create mode 100644 src/kma/debug.py delete mode 100644 tools/debug_streamlit.py diff --git a/README.md b/README.md index 2fe03f0..36cef25 100644 --- a/README.md +++ b/README.md @@ -661,10 +661,18 @@ Streamlit 디버그 화면은 선택 의존성으로 실행합니다. ```bash pip install -e ".[debug-ui]" -streamlit run tools/debug_streamlit.py +streamlit run examples/streamlit_debug_ui.py ``` -Raw Response 탭에는 선택한 API의 필수/선택 파라미터 입력 폼과 인증키를 제외한 request params preview가 표시됩니다. 좌측 메뉴에서는 API 풀네임/설명, 서비스키 링크, 환경변수 키 선택, 요청 timeout, fixture 기본 디렉터리를 조정할 수 있고, 폼에 없는 provider별 파라미터는 `Extra params JSON`으로 추가할 수 있습니다. 실행 후 Pydantic Model 탭에는 row 모델 변환 결과가, Processed Result 탭에는 표 형태 row preview가 표시됩니다. Debug Trace 탭에는 현재 카탈로그 항목, 선택한 데이터셋명, gateway, operation, 인증 파라미터, 키 발급/확인 링크가 표시됩니다. +좌측 메뉴는 Data source(`datagokr`/`apihub`) → Category → API 3단 계단식으로 구성됩니다 — `datagokr`는 `api_catalog(gateway="datagokr")`의 160개 operation을, `apihub`는 `apihub_endpoint_catalog()`의 470개 실제 호출 가능 endpoint를 다룹니다. 선택한 API의 설명 2줄, Environment(env var 사용 여부)와 Auth(실제 쿼리 파라미터명인 `serviceKey`/`authKey` 입력), 서비스키 발급 링크, timeout, fixture 저장 기본 디렉터리를 조정할 수 있습니다. + +메인 영역의 요청 파라미터 입력은 `st.form()`으로 감싸여 있고, 카탈로그의 `required_params`/`optional_params`/`param_defaults` 메타데이터에서 위젯을 자동 생성합니다(endpoint별 `if function_name == ...` 분기 없음). `dataType`/`type`처럼 고정 선택지가 있는 파라미터는 selectbox로, 나머지는 text input으로 렌더링되며, 폼에 없는 provider별 파라미터는 `Extra params JSON`으로 추가할 수 있습니다. + +고정 6개 탭(Raw Response / Pydantic Model / Processed Result / Validation Errors / Debug Trace / Fixture · Testcase)을 제공합니다. Raw Response에는 인증키를 제외한 request params preview와 raw 응답이, Pydantic Model에는 `DataGoKrItem`으로 검증한 row(또는 APIHub `response_kind`별 정리 결과)가, Processed Result에는 list 응답일 때만 표 형태 row preview가 표시됩니다. Validation Errors는 예외/검증 오류가 있을 때만 표시되고, Debug Trace에는 현재 카탈로그, 선택한 API 메타데이터, 요청 URL/파라미터(마스킹됨)/소요시간 trace가 표시됩니다. Fixture / Testcase 탭은 `save_fixture()`를 실제로 호출해 `tests/fixtures//.json`에 저장합니다. + +`src/kma/debug.py`는 이 UI와 fixture 저장에 공통으로 쓰는 `DebugRun`/`jsonable`/`redact_sensitive`/`debug_error`/`save_fixture`를 제공합니다. `DataGoKrClient.debug_fetch()`와 `ApiHubClient.debug_fetch_endpoint()`는 endpoint별 분기 없이 카탈로그가 넘겨주는 service/operation 또는 `ApiHubEndpointSpec`만으로 요청을 라우팅하는 제네릭 메서드입니다. + +APIHub는 470개 endpoint 전부가 이 UI에서 실행 가능하지만, `response_kind`가 `text`/`image`/`file`인 legacy endpoint는 `parse_apihub_text_table()`이 관대하게 만든 근사 표/metadata를 보여줄 뿐 endpoint별 정확한 파싱을 보장하지 않습니다. data.go.kr의 160개 operation 중 로컬로 정리된 필수/선택 파라미터 명세가 있는 것은 10개 service(약 30개 operation)뿐이며, 나머지는 `Extra params JSON`으로 직접 파라미터를 채워야 합니다. 기본 테스트는 실제 API를 호출하지 않아야 합니다. 실제 KMA 호출 테스트를 추가할 경우 `DATA_GO_KR_SERVICE_KEY`가 있을 때만 실행되도록 별도 marker를 사용하세요. @@ -677,6 +685,8 @@ Raw Response 탭에는 선택한 API의 필수/선택 파라미터 입력 폼과 이 문서와 프로젝트 문서의 파일 위치는 모두 프로젝트 루트 기준 상대 경로로 적습니다. 예를 들어 `src/kma/client.py`, `docs/testing.md`처럼 쓰고, 작업자 로컬 절대 경로는 문서에 남기지 않습니다. Python docstring과 내부 설명 문구는 한글로 작성하되, 코드 식별자와 API 파라미터 이름은 원문을 유지합니다. ```text +examples/ +└── streamlit_debug_ui.py src/kma/ ├── __init__.py ├── _credentials.py @@ -689,6 +699,7 @@ src/kma/ ├── codes.py ├── datagokr.py ├── datagokr_catalog.py +├── debug.py ├── enums.py ├── exceptions.py ├── grid.py @@ -716,7 +727,7 @@ tests/ ├── test_time_utils.py └── test_timeline.py tools/ -└── debug_streamlit.py +└── update_apihub_endpoints.py ``` 문서 지도는 상단의 [먼저 읽을 문서](#먼저-읽을-문서) 표를 참고하세요. diff --git a/examples/streamlit_debug_ui.py b/examples/streamlit_debug_ui.py new file mode 100644 index 0000000..b2d1ac5 --- /dev/null +++ b/examples/streamlit_debug_ui.py @@ -0,0 +1,665 @@ +"""Streamlit 기반 기상청 API 디버그 카탈로그 뷰어.""" +# ruff: noqa: E402,I001 + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) +for module_name, module in list(sys.modules.items()): + if module_name != "kma" and not module_name.startswith("kma."): + continue + module_file = getattr(module, "__file__", None) + if module_file is not None and not Path(module_file).resolve().is_relative_to(SRC): + del sys.modules[module_name] + +try: + import pandas as pd + import streamlit as st +except ModuleNotFoundError as exc: # pragma: no cover - 선택 실행 도구 + raise SystemExit('Streamlit UI를 쓰려면 `pip install -e ".[debug-ui]"`를 실행하세요.') from exc + +from kma import ( + ApiCatalogEntry, + ApiHubGeneratedClient, + DataGoKrClient, + DebugRun, + api_catalog, + api_key_for_gateway, + apihub_endpoint_catalog, + debug_error, + env_names_for_gateway, + jsonable, + load_local_env, + redact_sensitive, + save_fixture, +) + +# 요청 파라미터 중 고정된 선택지가 있는 것으로 알려진 이름 -> selectbox choices. +# `dataType`/`type`은 data.go.kr/APIHub 양쪽에서 흔히 쓰는 응답 형식 파라미터다. +# kma의 `enums.py`(WeatherCategory, SkyCode 등)는 응답 값 분류용이라 요청 +# 파라미터로는 재사용하지 않는다 — 실제로 요청 파라미터로 쓰이는 enum이 없다. +_ENUM_CHOICES: dict[str, tuple[str, ...]] = { + "dataType": ("JSON", "XML"), + "type": ("json", "xml"), +} + +_RESPONSE_KIND_DESCRIPTIONS: dict[str, str] = { + "structured": "JSON/XML 구조화 응답을 반환합니다.", + "text": "공백/CSV로 구분된 legacy text 응답을 표 형태로 파싱해 반환합니다.", + "image": "이미지(binary) 응답입니다 — 크기/포맷 metadata만 표시합니다.", + "file": "파일(binary) 응답입니다 — 크기/타입 metadata만 표시합니다.", +} + + +@dataclass(frozen=True) +class ParameterSpec: + """디버그 UI에서 요청 파라미터 입력 폼을 만들기 위한 최소 명세.""" + + name: str + required: bool + label: str + help: str = "" + default: str = "" + choices: tuple[str, ...] | None = None + + +def main() -> None: + st.set_page_config(page_title="KMA API Debug", layout="wide") + st.title("KMA API Debug") + + source = st.sidebar.selectbox("Data source", ["datagokr", "apihub"], key="source") + rows = _catalog_rows(source) + selected = _select_api(source, rows) + + line1, line2 = _api_summary_lines(selected) + st.sidebar.caption(line1) + st.sidebar.caption(line2) + + env_names = env_names_for_gateway(selected.gateway) + env_sources = _env_key_sources(env_names) + + environment = "manual" + if env_sources: + st.sidebar.subheader("Environment") + environment = st.sidebar.selectbox( + "Environment", ["env", "manual"], key=f"env-mode:{source}" + ) + if environment == "env": + source_info = env_sources[0] + st.sidebar.caption( + f"{source_info['name']} 값을 사용합니다. Source: {source_info['source']}" + ) + + st.sidebar.subheader("Auth") + if environment == "manual": + api_key = st.sidebar.text_input( + selected.credential_param, + value="", + type="password", + placeholder="직접 입력", + help=f"사용 가능한 env 이름: {', '.join(env_names)}", + key=f"auth:{source}", + ) + effective_api_key = api_key + else: + effective_api_key = _default_key(selected.gateway) + _service_key_links(selected) + + timeout = st.sidebar.number_input( + "Timeout", + min_value=1.0, + max_value=60.0, + value=10.0, + step=1.0, + help="API 요청 timeout seconds입니다.", + ) + fixture_base_dir = _fixture_base_dir_sidebar() + + tabs = st.tabs( + [ + "Raw Response", + "Pydantic Model", + "Processed Result", + "Validation Errors", + "Debug Trace", + "Fixture / Testcase", + ] + ) + + with tabs[0]: + _raw_response_tab(selected, effective_api_key, timeout=float(timeout)) + with tabs[1]: + _pydantic_model_tab(selected) + with tabs[2]: + _processed_result_tab(selected) + with tabs[3]: + _validation_errors_tab(selected) + with tabs[4]: + _debug_trace_tab(rows, selected, env_names) + with tabs[5]: + _fixture_tab(fixture_base_dir, selected) + + +def _catalog_rows(source: str) -> tuple[ApiCatalogEntry, ...]: + """선택한 gateway의 카탈로그 row를 반환합니다. + + `datagokr`는 `api_catalog(gateway="datagokr")`(160개 operation), + `apihub`는 `apihub_endpoint_catalog()`(470개 실제 호출 가능 endpoint)를 + 씁니다 — data.go.kr에 등록만 되어 있고 실행 불가능한 apihub LINK placeholder + dataset(`api_catalog(gateway="apihub")`)는 이 디버그 UI에서 쓰지 않습니다. + """ + + if source == "apihub": + return apihub_endpoint_catalog() + return api_catalog(gateway="datagokr") + + +def _select_api(source: str, rows: tuple[ApiCatalogEntry, ...]) -> ApiCatalogEntry: + """Category -> API 2단 계단식 selectbox로 카탈로그 row 하나를 고릅니다. + + `datagokr`는 Category=dataset명(38개), API=operation(160개 중 일부)이고, + `apihub`는 Category=관측/예특보 등 category명(11개), API=service/endpoint + 제목(470개 중 일부)입니다. 두 gateway 모두 Data source까지 합쳐 3단 + 계단식이 됩니다. + """ + + categories = sorted({row.dataset_name for row in rows}) + category = st.sidebar.selectbox("Category", categories, key=f"category:{source}") + category_rows = [row for row in rows if row.dataset_name == category] + api_labels = [_api_option_label(source, row) for row in category_rows] + api_label = st.sidebar.selectbox("API", api_labels, key=f"api:{source}:{category}") + return category_rows[api_labels.index(api_label)] + + +def _api_option_label(source: str, row: ApiCatalogEntry) -> str: + if source == "datagokr": + return row.operation or row.label + return row.label + + +def _api_summary_lines(selected: ApiCatalogEntry) -> tuple[str, str]: + """사이드바에 표시할 2줄 설명(무엇을 하는 API + 어떤 데이터를 반환하는지).""" + + if selected.gateway == "datagokr": + line1 = ( + f"{selected.dataset_name} — data.go.kr {selected.service}/{selected.operation} " + "operation을 호출합니다." + ) + required = ", ".join(selected.required_params) or "로컬에 정리된 필수 파라미터 없음" + line2 = f"반환: JSON/XML 응답의 items.item 목록입니다. 필수 파라미터: {required}." + return line1, line2 + + line1 = f"{selected.dataset_name} — APIHub {selected.service}({selected.endpoint_path}) 호출." + line2 = _RESPONSE_KIND_DESCRIPTIONS.get( + selected.response_kind, "알 수 없는 형식의 응답입니다." + ) + return line1, line2 + + +def _raw_response_tab(selected: ApiCatalogEntry, api_key: str, *, timeout: float) -> None: + st.subheader(selected.dataset_name) + st.caption(f"{selected.gateway} / {selected.label}") + + try: + submitted, params, request_options, missing = _request_form(selected) + except ValueError as exc: + st.error(str(exc)) + return + + preview: dict[str, Any] = dict(params) + if selected.gateway == "datagokr": + preview.update( + { + "pageNo": request_options["page_no"], + "numOfRows": request_options["num_of_rows"], + "dataType": request_options["data_type"], + } + ) + st.subheader("Request params preview") + st.json(preview) + + if not submitted: + return + if missing: + st.error("필수 파라미터를 입력하세요: " + ", ".join(missing)) + return + + run = _run_selected_api(selected, api_key, params, request_options, timeout=timeout) + _store_run(selected, run) + if run.error: + st.error(run.error["message"]) + st.json(jsonable(run.response)) + + +def _run_selected_api( + selected: ApiCatalogEntry, + api_key: str, + params: dict[str, Any], + request_options: dict[str, Any], + *, + timeout: float, +) -> DebugRun: + """카탈로그 entry를 gateway에 맞는 클라이언트로 라우팅해 실행합니다. + + endpoint별로 분기하는 코드는 없습니다 — `gateway`(datagokr/apihub) 두 + 값으로만 client 클래스를 고르고, 그 다음은 각 client의 제네릭 + `debug_fetch`/`debug_fetch_endpoint`가 `selected.service`/`.operation` + 이름으로 카탈로그 라우팅을 계속합니다. client 생성 자체가 실패해도(예: + 빈 인증값) 구조화된 `DebugRun.error`로 반환합니다. + """ + + try: + if selected.gateway == "datagokr": + client = DataGoKrClient(api_key, timeout=timeout, retries=0) + return client.debug_fetch( + selected.service or "", + selected.operation or "", + params, + page_no=request_options.get("page_no", 1), + num_of_rows=request_options.get("num_of_rows", 10), + data_type=request_options.get("data_type", "JSON"), + ) + hub_client = ApiHubGeneratedClient(api_key, timeout=timeout, retries=0) + spec = hub_client.endpoint(selected.service or "") + return hub_client.debug_fetch_endpoint(spec, params) + except Exception as exc: # pragma: no cover - UI 표시 + return DebugRun( + function=selected.service or selected.label, + input=redact_sensitive( + { + "gateway": selected.gateway, + "service": selected.service, + "operation": selected.operation, + "params": params, + } + ), + request={}, + response={}, + parsed=None, + processed=None, + trace=[f"{selected.gateway} 클라이언트 준비 실패: {exc.__class__.__name__}"], + error=debug_error(exc), + ) + + +def _request_form( + selected: ApiCatalogEntry, +) -> tuple[bool, dict[str, Any], dict[str, Any], list[str]]: + specs = _parameter_specs(selected) + required_specs = [spec for spec in specs if spec.required] + optional_specs = [spec for spec in specs if not spec.required] + key_prefix = f"{selected.gateway}:{selected.dataset_id}:{selected.service}:{selected.operation}" + + with st.form(f"request-form:{key_prefix}"): + st.subheader("Required parameters") + if required_specs: + required_values = _render_param_grid(required_specs, key_prefix=key_prefix) + else: + st.caption( + "이 API에 대해 로컬에 정리된 필수 파라미터 명세가 없습니다. " + "Extra params JSON으로 파라미터를 직접 추가하세요." + ) + required_values = {} + + st.subheader("Optional parameters") + if optional_specs: + optional_values = _render_param_grid(optional_specs, key_prefix=key_prefix) + else: + st.caption("정리된 선택 파라미터가 없습니다.") + optional_values = {} + + request_options: dict[str, Any] = {} + if selected.gateway == "datagokr": + page_no, num_of_rows, data_type = _render_common_options(key_prefix) + request_options = { + "page_no": page_no, + "num_of_rows": num_of_rows, + "data_type": data_type, + } + + extra_text = st.text_area( + "Extra params JSON", + value="{}", + height=110, + help="폼에 없는 provider 파라미터를 JSON object로 추가합니다.", + key=f"{key_prefix}:extra", + ) + submitted = st.form_submit_button("Run selected API") + + params = {**required_values, **optional_values, **_parse_extra_params(extra_text)} + missing = [spec.name for spec in required_specs if not str(params.get(spec.name, "")).strip()] + clean_params = {key: value for key, value in params.items() if str(value).strip()} + return submitted, clean_params, request_options, missing + + +def _parameter_specs(selected: ApiCatalogEntry) -> tuple[ParameterSpec, ...]: + """카탈로그의 `required_params`/`optional_params`에서 위젯 명세를 만듭니다. + + `if function_name == ...` 같은 endpoint별 분기는 없습니다 — 파라미터 + 이름과 `param_defaults`/enum choices만으로 위젯을 결정합니다. + """ + + defaults = selected.param_defaults + required = tuple( + _param(name, required=True, default=defaults.get(name, "")) + for name in selected.required_params + ) + optional = tuple( + _param(name, required=False, default=defaults.get(name, "")) + for name in selected.optional_params + ) + return required + optional + + +def _param(name: str, *, required: bool, default: str) -> ParameterSpec: + help_text = ( + "이 API의 필수 요청 파라미터입니다." if required else "이 API의 선택 요청 파라미터입니다." + ) + return ParameterSpec( + name=name, + required=required, + label=name, + help=help_text, + default=default, + choices=_ENUM_CHOICES.get(name), + ) + + +def _render_param_grid(specs: list[ParameterSpec], *, key_prefix: str) -> dict[str, str]: + values: dict[str, str] = {} + for index in range(0, len(specs), 2): + columns = st.columns(2) + for column, spec in zip(columns, specs[index : index + 2], strict=False): + with column: + if spec.choices: + default_index = ( + spec.choices.index(spec.default) if spec.default in spec.choices else 0 + ) + values[spec.name] = st.selectbox( + spec.label, + spec.choices, + index=default_index, + help=spec.help or None, + key=f"{key_prefix}:param:{spec.name}", + ) + else: + values[spec.name] = st.text_input( + spec.label, + value=spec.default, + help=spec.help or None, + key=f"{key_prefix}:param:{spec.name}", + ) + return values + + +def _render_common_options(key_prefix: str) -> tuple[int, int, str]: + col1, col2, col3 = st.columns(3) + with col1: + page_no = st.number_input( + "pageNo", + min_value=1, + value=1, + step=1, + help="공공데이터포털 paging 파라미터입니다.", + key=f"{key_prefix}:pageNo", + ) + with col2: + num_of_rows = st.number_input( + "numOfRows", + min_value=1, + value=10, + step=1, + help="한 페이지에 받을 row 수입니다.", + key=f"{key_prefix}:numOfRows", + ) + with col3: + data_type = st.selectbox( + "dataType", + ["JSON", "XML"], + index=0, + help="기본값은 JSON입니다.", + key=f"{key_prefix}:dataType", + ) + return int(page_no), int(num_of_rows), str(data_type) + + +def _parse_extra_params(text: str) -> dict[str, Any]: + try: + payload = json.loads(text or "{}") + except json.JSONDecodeError as exc: + raise ValueError(f"Extra params JSON is invalid: {exc}") from exc + if not isinstance(payload, dict): + raise ValueError("Extra params JSON must be an object") + reserved = { + "serviceKey", + "ServiceKey", + "authKey", + "AuthKey", + "pageNo", + "numOfRows", + "dataType", + } + return {key: value for key, value in payload.items() if key not in reserved} + + +def _pydantic_model_tab(selected: ApiCatalogEntry) -> None: + run = _current_run(selected) + if run is None: + st.info("Raw Response 탭에서 선택한 API를 실행하면 여기에서 Pydantic 모델을 확인합니다.") + return + if run.error: + st.warning("실행 중 오류가 있습니다. Validation Errors 탭을 확인하세요.") + if selected.gateway == "apihub": + st.caption( + "APIHub는 endpoint마다 응답 형식(text/structured/image/file)이 달라 전용 " + "Pydantic row 모델이 없습니다. response_kind에 맞춰 정리한 구조를 표시합니다." + ) + else: + st.caption("각 row를 `DataGoKrItem` Pydantic 모델로 검증한 결과입니다.") + st.json(jsonable(run.parsed)) + + +def _processed_result_tab(selected: ApiCatalogEntry) -> None: + run = _current_run(selected) + if run is None: + st.info("Raw Response 탭에서 API를 실행하면 처리된 row preview를 표시합니다.") + return + data = jsonable(run.processed) + if isinstance(data, list) and data: + st.dataframe(pd.json_normalize(data, sep="."), width="stretch", hide_index=True) + else: + st.json(data) + + +def _validation_errors_tab(selected: ApiCatalogEntry) -> None: + run = _current_run(selected) + if run is None: + st.info("아직 실행된 API가 없습니다.") + return + if not run.error: + st.success("현재 실행 결과에서 validation error 또는 exception이 없습니다.") + return + st.error(run.error["message"]) + st.json(run.error) + + +def _debug_trace_tab( + rows: tuple[ApiCatalogEntry, ...], + selected: ApiCatalogEntry, + env_names: tuple[str, ...], +) -> None: + run = _current_run(selected) + + st.subheader("Catalog") + st.caption(f"현재 Data source 카탈로그: {len(rows)}개 API") + st.dataframe([row.asdict() for row in rows], width="stretch", hide_index=True) + + st.subheader("Selected API") + st.json(selected.asdict()) + st.link_button(f"{selected.credential_param} 발급/확인", selected.service_key_url) + st.caption(f"credential env: {', '.join(env_names)}") + + if run is not None: + st.subheader("Trace") + st.write(run.trace) + st.subheader("Request (redacted)") + st.json(jsonable(run.request)) + + +def _fixture_tab(fixture_base_dir: str, selected: ApiCatalogEntry) -> None: + run = _current_run(selected) + if run is None: + st.info("Raw Response 탭에서 API를 실행한 뒤 fixture를 저장할 수 있습니다.") + st.caption("Fixture base dir") + st.code(fixture_base_dir, language=None) + return + + with st.expander("Save as fixture", expanded=True): + case_name = st.text_input("Case name", value=f"{run.function}_normal") + description = st.text_area("Description", value=f"{selected.label} 정상 케이스") + assertion_mode = st.selectbox( + "Assertion mode", + ["snapshot", "schema_only", "required_fields", "count"], + ) + exclude_fields_raw = st.text_input( + "Exclude fields", + value="fetched_at, collected_at, request_id, updated_at", + ) + required_fields_raw = st.text_input("Required fields", value="") + overwrite = st.checkbox("Overwrite existing fixture", value=False) + + assertion = { + "mode": assertion_mode, + "exclude_fields": [ + value.strip() for value in exclude_fields_raw.split(",") if value.strip() + ], + "required_fields": [ + value.strip() for value in required_fields_raw.split(",") if value.strip() + ], + } + + st.subheader("Fixture preview") + st.json( + { + "function": run.function, + "input": jsonable(run.input), + "request": jsonable(run.request), + "response": jsonable(run.response), + "processed": jsonable(run.processed), + "assertion": assertion, + } + ) + + if st.button("Save as fixture"): + try: + path = save_fixture( + base_dir=fixture_base_dir, + function_name=run.function, + case_name=case_name, + description=description, + input_data=run.input, + request_data=run.request, + response_data=run.response, + parsed_result=run.parsed, + processed_result=run.processed, + assertion=assertion, + overwrite=overwrite, + ) + except Exception as exc: # pragma: no cover - UI 표시 + st.error(str(exc)) + else: + st.success(f"Saved: {path}") + + +def _service_key_links(selected: ApiCatalogEntry) -> None: + st.sidebar.caption("Service key links") + st.sidebar.link_button(f"{selected.credential_param} 발급/확인", selected.service_key_url) + if selected.portal_url != selected.service_key_url: + st.sidebar.link_button("data.go.kr 카탈로그", selected.portal_url) + + +def _env_key_sources(env_names: tuple[str, ...]) -> list[dict[str, str]]: + sources: list[dict[str, str]] = [] + for name in env_names: + value = os.getenv(name) + if value is not None and value.strip(): + sources.append({"name": name, "source": "process env"}) + return sources + + local_env = load_local_env() + for name in env_names: + value = local_env.get(name) + if value is not None and value.strip(): + sources.append({"name": name, "source": ".env 또는 .env.local"}) + return sources + return sources + + +def _default_key(gateway: str) -> str: + try: + return api_key_for_gateway(gateway) + except ValueError: + return "" + + +def _fixture_base_dir_sidebar() -> str: + st.sidebar.subheader("Fixtures") + candidates = _fixture_dir_candidates() + options = [str(path) for path in candidates] + custom_label = "Custom..." + selected = st.sidebar.selectbox("Fixture base dir", [*options, custom_label]) + if selected == custom_label: + selected = st.sidebar.text_input( + "Custom fixture base dir", + value=str((ROOT / "tests" / "fixtures").resolve()), + ) + st.sidebar.caption(selected) + return selected + + +def _fixture_dir_candidates() -> list[Path]: + preferred = [ + ROOT / "tests" / "fixtures", + ROOT / "tests", + ROOT / "examples", + ROOT, + ] + candidates: list[Path] = [] + for path in preferred: + resolved = path.resolve() + if resolved not in candidates: + candidates.append(resolved) + return candidates + + +def _store_run(selected: ApiCatalogEntry, run: DebugRun) -> None: + st.session_state["last_run"] = { + "selection_key": _selection_key(selected), + "run": run, + } + + +def _current_run(selected: ApiCatalogEntry) -> DebugRun | None: + stored = st.session_state.get("last_run") + if not isinstance(stored, dict): + return None + if stored.get("selection_key") != _selection_key(selected): + return None + return stored.get("run") + + +def _selection_key(selected: ApiCatalogEntry) -> str: + return f"{selected.gateway}:{selected.dataset_id}:{selected.service}:{selected.operation}" + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 33065c0..4c91cb9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ dev = [ "mypy>=1.10", ] debug-ui = [ + "pandas>=2", "streamlit>=1.36", ] diff --git a/src/kma/__init__.py b/src/kma/__init__.py index c60c60a..8047afd 100644 --- a/src/kma/__init__.py +++ b/src/kma/__init__.py @@ -10,11 +10,21 @@ AsyncApiHubClient, ) from .apihub_endpoints import APIHUB_ATTACHMENTS, APIHUB_ENDPOINTS, ApiHubGeneratedClient -from .catalog import ApiCatalogEntry, api_catalog +from .catalog import ApiCatalogEntry, api_catalog, apihub_endpoint_catalog from .client import AsyncForecastService, AsyncKmaClient, ForecastService, KmaClient from .codes import label_for, parse_amount, unit_for from .datagokr import AsyncDataGoKrClient, DataGoKrClient from .datagokr_catalog import KMA_DATA_GOKR_DATASETS, DataGoKrDatasetSpec +from .debug import ( + DEFAULT_ASSERTION, + SENSITIVE_KEYS, + DebugRun, + debug_error, + jsonable, + redact_sensitive, + save_fixture, + slugify_case_name, +) from .enums import ( ForecastPrecipitationType, KmaEndpoint, @@ -77,9 +87,11 @@ "BeachTideItem", "BeachWaterTemperature", "BeachWaveHeight", + "DEFAULT_ASSERTION", "DataGoKrClient", "DataGoKrDatasetSpec", "DataGoKrItem", + "DebugRun", "ForecastItem", "ForecastService", "ForecastTimepoint", @@ -97,6 +109,7 @@ "MidForecastItem", "ObservedPrecipitationType", "ResponseMetadata", + "SENSITIVE_KEYS", "SkyCode", "WeatherCategory", "WeatherSnapshot", @@ -104,7 +117,10 @@ "api_key_for_gateway", "has_next_page", "api_catalog", + "apihub_endpoint_catalog", + "debug_error", "iter_pages", + "jsonable", "kma_grid_to_wgs84", "label_for", "base_available_at", @@ -118,7 +134,10 @@ "env_names_for_gateway", "parse_amount", "pivot_forecast_items", + "redact_sensitive", "sanitize_request_params", + "save_fixture", + "slugify_case_name", "to_grid", "to_latlon", "unit_for", diff --git a/src/kma/apihub.py b/src/kma/apihub.py index 072e661..18557a8 100644 --- a/src/kma/apihub.py +++ b/src/kma/apihub.py @@ -7,6 +7,7 @@ import io import json import re +import time from collections.abc import AsyncIterator, Iterable, Iterator, Mapping from dataclasses import dataclass from functools import cached_property @@ -27,6 +28,7 @@ raise_for_kma_result_code, raise_for_kma_xml_error_body, ) +from .debug import DebugRun, debug_error, redact_sensitive from .exceptions import KmaParseError from .metadata import ( ResponseMetadata, @@ -412,6 +414,92 @@ async def adiscover_endpoints( ) return extract_apihub_endpoints(response.text) + def debug_fetch_endpoint( + self, + spec: ApiHubEndpointSpec, + params: Mapping[str, Any] | None = None, + *, + use_sample: bool = False, + ) -> DebugRun: + """디버그 UI/fixture 생성을 위해 APIHub endpoint 하나를 호출합니다. + + `spec`은 카탈로그(`apihub_endpoint_catalog()`)나 + `ApiHubGeneratedClient.endpoint(name)`에서 얻은 `ApiHubEndpointSpec` + 입니다. 여기서는 어떤 endpoint인지에 따라 분기하지 않고, `spec`이 담은 + `path`/`query_parts`/`response_kind` 메타데이터로만 요청을 만들고 결과를 + 해석합니다 — 470개 endpoint 전부가 같은 경로를 지납니다. + + 일부 legacy endpoint는 이름 없는 query 조각(``query_parts``에 + ``"bare"`` 항목)을 쓰므로, 그 경우에만 `request_query_parts`로, + 나머지는 `request_path`로 호출을 위임합니다(``ApiHubGeneratedClient + .call_endpoint``와 같은 판정 규칙). + """ + + request_params: dict[str, Any] = {} + if use_sample: + request_params.update(spec.sample_params) + if params: + request_params.update(params) + + input_data = redact_sensitive( + { + "endpoint": spec.name, + "params": request_params, + "use_sample": use_sample, + } + ) + trace = [ + f"APIHub {spec.name} ({spec.path}) 호출 준비", + f"response_kind={spec.response_kind}", + ] + request_info = redact_sensitive( + { + "method": "GET", + "url": f"{self.base_url}{spec.path}", + "query": request_params, + } + ) + + started_at = time.monotonic() + try: + if any(kind == "bare" for kind, _name in spec.query_parts): + response = self.request_query_parts(spec.path, spec.query_parts, request_params) + else: + response = self.request_path(spec.path, request_params) + except Exception as exc: + elapsed_ms = (time.monotonic() - started_at) * 1000 + trace.append(f"요청 실패: {exc.__class__.__name__} ({elapsed_ms:.0f}ms)") + return DebugRun( + function=spec.name, + input=input_data, + request=request_info, + response={}, + parsed=None, + processed=None, + trace=trace, + error=debug_error(exc), + ) + + elapsed_ms = (time.monotonic() - started_at) * 1000 + trace.append( + f"응답 수신: HTTP {response.status_code}, {len(response.content)} bytes " + f"({elapsed_ms:.0f}ms)" + ) + parsed, processed = _debug_parse_apihub_response(response, spec.response_kind) + return DebugRun( + function=spec.name, + input=input_data, + request=request_info, + response={ + "status_code": response.status_code, + "content_type": response.content_type, + "body": parsed, + }, + parsed=parsed, + processed=processed, + trace=trace, + ) + def iter_pages( self, service: str, @@ -913,6 +1001,43 @@ def _check_apihub_result_code(response: ApiHubResponse, *, endpoint: str) -> Non ) +def _debug_parse_apihub_response(response: ApiHubResponse, response_kind: str) -> tuple[Any, Any]: + """`debug_fetch_endpoint`용으로 `response_kind`에 맞춰 raw/processed 값을 만듭니다. + + 반환값은 `(parsed, processed)`입니다. `processed`는 list 모양이면 Streamlit + 쪽에서 dataframe으로, 아니면 단일 object로 표시됩니다. + """ + + if response_kind == "structured": + try: + parsed = response.json() + except KmaParseError: + parsed = {"text_preview": response.text[:2000]} + return parsed, parsed + if response_kind == "text": + table = response.text_table() + rows = [dict(row) for row in table.rows] + parsed = { + "headers": list(table.headers), + "rows": rows, + "comments": list(table.comments), + } + return parsed, (rows if rows else parsed) + if response_kind == "image": + image = response.image() + parsed = { + "content_type": image.content_type, + "format": image.format, + "width": image.width, + "height": image.height, + "bytes": len(image.content), + } + return parsed, parsed + # "file" 또는 알 수 없는 response_kind — 내용을 해석하지 않고 크기/타입만 보여준다. + parsed = {"content_type": response.content_type, "bytes": len(response.content)} + return parsed, parsed + + def _apihub_open_api_body(response: ApiHubResponse, *, endpoint: str) -> Mapping[str, Any]: payload = response.json() try: diff --git a/src/kma/catalog.py b/src/kma/catalog.py index 598b25a..793eacc 100644 --- a/src/kma/catalog.py +++ b/src/kma/catalog.py @@ -2,7 +2,8 @@ from __future__ import annotations -from dataclasses import dataclass +from collections.abc import Mapping +from dataclasses import dataclass, field from functools import lru_cache from typing import Any @@ -36,7 +37,13 @@ def _apihub_equivalent_path(gateway: str, service: str | None, operation: str | @dataclass(frozen=True) class ApiCatalogEntry: - """data.go.kr dataset 카탈로그를 operation 단위로 펼친 항목.""" + """data.go.kr dataset 카탈로그를 operation 단위로 펼친 항목. + + `required_params`/`optional_params`는 디버그 UI가 `st.form()` 위젯을 + 자동으로 만드는 데 쓰는 파라미터 이름 메타데이터입니다. 로컬에 정리된 + 명세가 없는 operation은 두 값 모두 빈 tuple이며, 호출자는 자유 형식 + "Extra params JSON" 입력으로 파라미터를 보충해야 합니다. + """ dataset_id: str dataset_name: str @@ -50,6 +57,11 @@ class ApiCatalogEntry: label: str has_apihub_equivalent: bool = False apihub_equivalent_path: str | None = None + required_params: tuple[str, ...] = () + optional_params: tuple[str, ...] = () + param_defaults: Mapping[str, str] = field(default_factory=dict) + response_kind: str = "structured" + endpoint_path: str | None = None def asdict(self) -> dict[str, Any]: """Streamlit, JSON, 표 렌더링에서 쓰기 쉬운 dict로 변환합니다.""" @@ -67,6 +79,11 @@ def asdict(self) -> dict[str, Any]: "label": self.label, "has_apihub_equivalent": self.has_apihub_equivalent, "apihub_equivalent_path": self.apihub_equivalent_path, + "required_params": list(self.required_params), + "optional_params": list(self.optional_params), + "param_defaults": dict(self.param_defaults), + "response_kind": self.response_kind, + "endpoint_path": self.endpoint_path, } @@ -83,6 +100,12 @@ def api_catalog( `service_key_url`은 data.go.kr `serviceKey` 또는 APIHub `authKey`를 발급, 확인할 수 있는 포털 링크입니다. + + 이 함수는 data.go.kr 카탈로그(operation 단위)만 다룹니다. APIHub의 470개 + 실제 호출 가능 endpoint는 :func:`apihub_endpoint_catalog`가 별도로 + 반환합니다 — 서로 다른 두 소스(data.go.kr 카탈로그 vs + `apiList.do`/`generateAPIUrl.do` 스크래핑 결과)를 섞으면 기존 dataset + 단위 카운트가 흔들리기 때문입니다. """ clean_gateway = gateway.strip().lower() if gateway is not None else None @@ -107,6 +130,14 @@ def _catalog_entry(dataset: Any, *, operation: str | None) -> ApiCatalogEntry: credential_param = "serviceKey" if dataset.gateway == "datagokr" else "authKey" service_key_url = dataset.portal_url if dataset.gateway == "datagokr" else APIHUB_AUTH_KEY_URL apihub_path = _apihub_equivalent_path(dataset.gateway, dataset.service, operation) + required_params, optional_params, param_defaults = _datagokr_param_spec( + dataset.service, operation + ) + endpoint_path = ( + f"/{dataset.service}/{operation}" + if dataset.gateway == "datagokr" and dataset.service and operation + else None + ) return ApiCatalogEntry( dataset_id=dataset.dataset_id, dataset_name=dataset.title, @@ -120,4 +151,205 @@ def _catalog_entry(dataset: Any, *, operation: str | None) -> ApiCatalogEntry: label=label, has_apihub_equivalent=apihub_path is not None, apihub_equivalent_path=apihub_path, + required_params=required_params, + optional_params=optional_params, + param_defaults=param_defaults, + response_kind="structured", + endpoint_path=endpoint_path, ) + + +def apihub_endpoint_catalog() -> tuple[ApiCatalogEntry, ...]: + """APIHub `apiList.do`/`generateAPIUrl.do`에서 수집한 470개 실제 호출 + 가능 endpoint를 `ApiCatalogEntry` row로 반환합니다. + + 각 row의 `service`는 `ApiHubGeneratedClient.call_endpoint(name, ...)`가 + 받는 endpoint 식별자(`ApiHubEndpointSpec.name`)입니다. `optional_params`는 + `authKey`를 제외한 그 endpoint의 모든 알려진 query parameter 이름이고, + `param_defaults`는 `apiList.do`에서 실제로 관찰된 sample 값입니다 — 이 + 카탈로그는 어떤 parameter가 진짜 필수인지 표기하지 않으므로(원본 문서에 + 없음) 전부 optional로 두고 sample 값으로 미리 채워 실행 가능하게 합니다. + """ + + from .apihub_endpoints import APIHUB_ENDPOINTS + + rows: list[ApiCatalogEntry] = [] + for spec in APIHUB_ENDPOINTS: + optional_params = tuple(name for name in spec.parameters if name != "authKey") + rows.append( + ApiCatalogEntry( + dataset_id=spec.name, + dataset_name=spec.category_name, + gateway="apihub", + service=spec.name, + operation=spec.title, + portal_url=APIHUB_AUTH_KEY_URL, + service_key_url=APIHUB_AUTH_KEY_URL, + credential_param="authKey", + page=spec.category_id, + label=f"{spec.service_name} / {spec.title}", + has_apihub_equivalent=False, + apihub_equivalent_path=None, + required_params=(), + optional_params=optional_params, + param_defaults=dict(spec.sample_params), + response_kind=spec.response_kind, + endpoint_path=spec.path, + ) + ) + return tuple(rows) + + +def _datagokr_param_spec( + service: str | None, operation: str | None +) -> tuple[tuple[str, ...], tuple[str, ...], dict[str, str]]: + """`(service, operation)`에 로컬로 정리된 파라미터 명세가 있으면 반환합니다. + + 명세가 없으면 `((), (), {})`를 반환합니다 — 디버그 UI는 이 경우 "Extra + params JSON" 자유 입력으로 파라미터를 보충하도록 안내합니다. 시간에 따라 + 달라지는 값(base_date/tmFc 등)은 기본값을 채우지 않고, 안정적인 코드성 + 값(dataCd, regId 등)만 `param_defaults`로 제공합니다. + """ + + spec = _DATAGOKR_PARAM_SPECS.get((service, operation)) + if spec is None: + return (), (), {} + required, optional, defaults = spec + return required, optional, dict(defaults) + + +# service/operation 조합별 파라미터 명세. 값은 (required, optional, static_defaults)다. +# `static_defaults`는 시간에 의존하지 않는 안정적인 코드 값만 담는다 — base_date, +# base_time, tmFc, currentDate 같은 시각 기반 값은 사용자가 직접 입력한다. +_ParamSpec = tuple[tuple[str, ...], tuple[str, ...], dict[str, str]] +_DATAGOKR_PARAM_SPECS: dict[tuple[str | None, str | None], _ParamSpec] = { + ("VilageFcstInfoService_2.0", "getUltraSrtNcst"): ( + ("base_date", "base_time", "nx", "ny"), + (), + {}, + ), + ("VilageFcstInfoService_2.0", "getUltraSrtFcst"): ( + ("base_date", "base_time", "nx", "ny"), + (), + {}, + ), + ("VilageFcstInfoService_2.0", "getVilageFcst"): ( + ("base_date", "base_time", "nx", "ny"), + (), + {}, + ), + ("VilageFcstInfoService_2.0", "getFcstVersion"): ( + ("ftype", "basedatetime"), + (), + {"ftype": "ODAM"}, + ), + ("MidFcstInfoService", "getMidFcst"): ( + ("stnId", "tmFc"), + (), + {"stnId": "108"}, + ), + ("MidFcstInfoService", "getMidLandFcst"): ( + ("regId", "tmFc"), + (), + {"regId": "11B00000"}, + ), + ("MidFcstInfoService", "getMidTa"): ( + ("regId", "tmFc"), + (), + {"regId": "11B10101"}, + ), + ("MidFcstInfoService", "getMidSeaFcst"): ( + ("regId", "tmFc"), + (), + {"regId": "11B00000"}, + ), + ("AsosDalyInfoService", "getWthrDataList"): ( + ("startDt", "endDt", "dataCd", "dateCd"), + ("stnIds",), + {"dataCd": "ASOS", "dateCd": "DAY"}, + ), + ("AsosHourlyInfoService", "getWthrDataList"): ( + ("startDt", "startHh", "endDt", "endHh", "dataCd", "dateCd"), + ("stnIds",), + {"dataCd": "ASOS", "dateCd": "HR"}, + ), + ("WthrWrnInfoService", "getWthrWrnList"): ( + ("stnId", "fromTmFc", "toTmFc"), + (), + {"stnId": "108"}, + ), + ("TourStnInfoService1", "getTourStnVilageFcst1"): ( + ("courseId", "currentDate", "hour"), + (), + {"courseId": "1"}, + ), + ("TourStnInfoService1", "getCityTourClmIdx1"): ( + ("cityAreaId", "currentDate", "day"), + (), + {"cityAreaId": "1", "day": "0"}, + ), + ("VilageFcstMsgService", "getWthrSituation"): ( + ("stnId",), + (), + {"stnId": "108"}, + ), + ("VilageFcstMsgService", "getLandFcst"): ( + ("regId",), + (), + {"regId": "11B00000"}, + ), + ("VilageFcstMsgService", "getSeaFcst"): ( + ("regId",), + (), + {"regId": "11B00000"}, + ), + ("BeachInfoservice", "getUltraSrtFcstBeach"): ( + ("beach_num", "base_date", "base_time"), + (), + {"beach_num": "1"}, + ), + ("BeachInfoservice", "getVilageFcstBeach"): ( + ("beach_num", "base_date", "base_time"), + (), + {"beach_num": "1"}, + ), + ("BeachInfoservice", "getWhBuoyBeach"): ( + ("beach_num", "searchTime"), + (), + {"beach_num": "1"}, + ), + ("BeachInfoservice", "getTwBuoyBeach"): ( + ("beach_num", "searchTime"), + (), + {"beach_num": "1"}, + ), + ("BeachInfoservice", "getSunInfoBeach"): ( + ("beach_num", "Base_date"), + (), + {"beach_num": "1"}, + ), + ("BeachInfoservice", "getTideInfoBeach"): ( + ("beach_num", "base_date"), + (), + {"beach_num": "1"}, + ), + ("LivingWthrIdxServiceV4", "getSenTaIdxV4"): ( + ("areaNo", "time", "requestCode"), + (), + {"areaNo": "1100000000", "requestCode": "A01"}, + ), + ("LivingWthrIdxServiceV4", "getUVIdxV4"): ( + ("areaNo", "time"), + (), + {"areaNo": "1100000000"}, + ), + ("LivingWthrIdxServiceV4", "getAirDiffusionIdxV4"): ( + ("areaNo", "time"), + (), + {"areaNo": "1100000000"}, + ), + ("EqkInfoService", "getEqkMsg"): (("fromTmFc", "toTmFc"), (), {}), + ("EqkInfoService", "getEqkMsgList"): (("fromTmFc", "toTmFc"), (), {}), + ("EqkInfoService", "getTsunamiMsg"): (("fromTmFc", "toTmFc"), (), {}), + ("EqkInfoService", "getTsunamiMsgList"): (("fromTmFc", "toTmFc"), (), {}), +} diff --git a/src/kma/datagokr.py b/src/kma/datagokr.py index 45e0aae..9cddf6c 100644 --- a/src/kma/datagokr.py +++ b/src/kma/datagokr.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from collections.abc import AsyncIterator, Callable, Iterator, Mapping from dataclasses import dataclass from datetime import date, datetime @@ -32,6 +33,7 @@ KMA_DATA_GOKR_DATASETS_BY_ID, DataGoKrDatasetSpec, ) +from .debug import DebugRun, debug_error, jsonable, redact_sensitive from .enums import coerce_category from .exceptions import KmaParseError from .metadata import ResponseMetadata, make_response_metadata @@ -257,6 +259,111 @@ async def arequest_with_metadata( ) return response.body, response.metadata + def debug_fetch( + self, + service: str, + operation: str, + params: Mapping[str, Any] | None = None, + *, + page_no: int = 1, + num_of_rows: int = 10, + data_type: str = "JSON", + ) -> DebugRun: + """디버그 UI/fixture 생성을 위해 service operation을 호출하고 실행 정보를 반환합니다. + + `service`/`operation`은 `request()`와 같은 인자이므로, 카탈로그 + (`ApiCatalogEntry.service`/`.operation`)에서 곧바로 전달할 수 있습니다. + endpoint별로 분기하는 코드는 없습니다 — 모든 data.go.kr operation이 + 같은 경로를 지납니다. + """ + + clean_service = service.strip("/") + clean_operation = operation.strip("/") + endpoint = f"{clean_service}/{clean_operation}" + clean_params = dict(params or {}) + input_data = redact_sensitive( + { + "service": clean_service, + "operation": clean_operation, + "params": clean_params, + "page_no": page_no, + "num_of_rows": num_of_rows, + "data_type": data_type, + } + ) + trace = [ + f"data.go.kr {endpoint} 호출 준비", + f"pageNo={page_no} numOfRows={num_of_rows} dataType={data_type}", + ] + request_info = redact_sensitive( + { + "method": "GET", + "url": f"{self.base_url}/{endpoint}", + "query": { + **clean_params, + "pageNo": page_no, + "numOfRows": num_of_rows, + "dataType": data_type, + }, + } + ) + + started_at = time.monotonic() + try: + body, metadata = self.request_with_metadata( + clean_service, + clean_operation, + clean_params, + data_type=data_type, + page_no=page_no, + num_of_rows=num_of_rows, + ) + processed = _debug_processed_rows(body) + # `items.item`이 list/object로 나오면 실제 `DataGoKrItem` pydantic 모델로 + # 감싸본다 — Pydantic Model 탭이 raw body를 그대로 되풀이하지 않고, 이 + # 시점에 진짜 검증 오류가 나면 Validation Errors 탭에 그대로 반영된다. + row_source = processed if isinstance(processed, list) else [processed] + models = [ + DataGoKrItem( + service=clean_service, operation=clean_operation, raw=dict(row) + ).model_dump(mode="json") + for row in row_source + if isinstance(row, Mapping) + ] + except Exception as exc: + elapsed_ms = (time.monotonic() - started_at) * 1000 + trace.append(f"요청 실패: {exc.__class__.__name__} ({elapsed_ms:.0f}ms)") + return DebugRun( + function=endpoint, + input=input_data, + request=request_info, + response={}, + parsed=None, + processed=None, + trace=trace, + error=debug_error(exc), + ) + + elapsed_ms = (time.monotonic() - started_at) * 1000 + trace.append(f"응답 수신 ({elapsed_ms:.0f}ms), resultCode 정상") + trace.append( + f"row {len(processed)}건 추출, DataGoKrItem {len(models)}건 생성" + if isinstance(processed, list) + else "단일 object 응답 — row 추출 없음" + ) + parsed_model = ( + models if isinstance(processed, list) else (models[0] if models else jsonable(body)) + ) + return DebugRun( + function=endpoint, + input=input_data, + request={**request_info, "query": redact_sensitive(dict(metadata.request_params))}, + response={"status_code": 200, "body": jsonable(body)}, + parsed=parsed_model, + processed=processed, + trace=trace, + ) + def _request_with_metadata( self, service: str, @@ -1788,6 +1895,25 @@ def _items_from_body(body: Mapping[str, Any], *, endpoint: str) -> list[Mapping[ ) +def _debug_processed_rows(body: Mapping[str, Any]) -> Any: + """디버그 UI Processed Result 탭용으로 `body.items.item`을 관대하게 추출합니다. + + `_items_from_body`와 달리 shape이 예상과 다르면 예외 대신 원본 body를 + 그대로 반환합니다 — `getFcstVersion` 같은 단일 object 응답도 debug_fetch가 + 실패하지 않아야 하기 때문입니다. + """ + + try: + raw_items = body["items"]["item"] + except (KeyError, TypeError): + return jsonable(body) + if isinstance(raw_items, Mapping): + return [jsonable(raw_items)] + if isinstance(raw_items, list): + return [jsonable(item) for item in raw_items] + return jsonable(body) + + def _mid_forecast_item( row: Mapping[str, Any], operation: str, diff --git a/src/kma/debug.py b/src/kma/debug.py new file mode 100644 index 0000000..322af94 --- /dev/null +++ b/src/kma/debug.py @@ -0,0 +1,168 @@ +"""디버그 UI와 fixture 저장에 공통으로 쓰는 보조 기능.""" + +from __future__ import annotations + +import json +import re +import traceback as _traceback +from collections.abc import Mapping +from dataclasses import asdict, dataclass, is_dataclass +from datetime import date, datetime +from os import PathLike +from pathlib import Path +from typing import Any, cast +from zoneinfo import ZoneInfo + +from pydantic import BaseModel + +from .exceptions import KmaError +from .metadata import redact_credentials_in_text + +SENSITIVE_KEYS = { + "authorization", + "x-api-key", + "api_key", + "apikey", + "api-key", + "auth_key", + "authkey", + "auth-key", + "service_key", + "servicekey", + "service-key", + "access_token", + "refresh_token", +} +DEFAULT_ASSERTION = { + "mode": "snapshot", + "exclude_fields": ["fetched_at", "collected_at", "request_id", "updated_at"], + "required_fields": [], +} + + +@dataclass(frozen=True) +class DebugRun: + """API 디버깅 한 번의 입력, 요청, 응답, 파싱, 가공 결과 묶음.""" + + function: str + input: dict[str, Any] + request: dict[str, Any] + response: dict[str, Any] + parsed: Any + processed: Any + trace: list[str] + error: dict[str, Any] | None = None + catalog: dict[str, Any] | None = None + + +def jsonable(obj: Any) -> Any: + """Pydantic v2 모델과 날짜/Path 값을 JSON으로 저장 가능한 값으로 변환합니다.""" + + if isinstance(obj, BaseModel): + return obj.model_dump(mode="json") + if is_dataclass(obj) and not isinstance(obj, type): + return jsonable(asdict(obj)) + if isinstance(obj, Mapping): + return {str(key): jsonable(value) for key, value in obj.items()} + if isinstance(obj, (list, tuple, set, frozenset)): + return [jsonable(item) for item in obj] + if isinstance(obj, datetime | date): + return obj.isoformat() + if isinstance(obj, Path): + return str(obj) + if isinstance(obj, bytes): + return f"<{len(obj)} bytes>" + return obj + + +def redact_sensitive(obj: Any) -> Any: + """dict/list 구조와 문자열 값에서 API key/token 성격의 값을 마스킹합니다.""" + + if isinstance(obj, Mapping): + redacted: dict[str, Any] = {} + for key, value in obj.items(): + text_key = str(key) + if text_key.replace("-", "_").lower() in SENSITIVE_KEYS: + redacted[text_key] = "" + else: + redacted[text_key] = redact_sensitive(value) + return redacted + if isinstance(obj, list | tuple): + return [redact_sensitive(item) for item in obj] + if isinstance(obj, str): + return redact_credentials_in_text(obj) + return obj + + +def debug_error(exc: Exception) -> dict[str, Any]: + """예외를 디버그 UI/fixture에 넣기 쉬운 dict로 변환합니다. + + `{type, message, traceback}`은 항상 채워지고, `kma` 자체 예외 + (`KmaError` 서브클래스)이면 `provider`/`endpoint`/`status_code`/ + `result_code`/`failure_kind`/`retryable` 필드도 추가됩니다. 반환값은 + `redact_sensitive()`를 거쳐 인증값이 남지 않습니다. + """ + + payload: dict[str, Any] = { + "type": exc.__class__.__name__, + "message": str(exc), + "traceback": "".join(_traceback.format_exception(type(exc), exc, exc.__traceback__)), + } + if isinstance(exc, KmaError): + payload.update(exc.metadata) + return cast(dict[str, Any], redact_sensitive(payload)) + + +def save_fixture( + *, + base_dir: str | PathLike[str], + function_name: str, + case_name: str, + description: str, + input_data: Any, + request_data: Any, + response_data: Any, + parsed_result: Any, + processed_result: Any, + assertion: Mapping[str, Any] | None = None, + library_version: str | None = None, + overwrite: bool = False, +) -> Path: + """디버그 실행 결과를 pytest replay용 fixture JSON 파일로 저장합니다.""" + + safe_case_name = slugify_case_name(case_name) + fixture_dir = Path(base_dir) / function_name + fixture_dir.mkdir(parents=True, exist_ok=True) + fixture_path = fixture_dir / f"{safe_case_name}.json" + if fixture_path.exists() and not overwrite: + raise FileExistsError(f"Fixture already exists: {fixture_path}") + + fixture = { + "name": safe_case_name, + "function": function_name, + "description": description, + "input": redact_sensitive(jsonable(input_data)), + "request": redact_sensitive(jsonable(request_data)), + "response": redact_sensitive(jsonable(response_data)), + "parsed": jsonable(parsed_result), + "processed": jsonable(processed_result), + "assertion": dict(assertion or DEFAULT_ASSERTION), + "meta": { + "created_at": datetime.now(ZoneInfo("Asia/Seoul")).isoformat(), + "library_version": library_version, + "source": "debug_ui", + }, + } + with fixture_path.open("w", encoding="utf-8") as handle: + json.dump(fixture, handle, ensure_ascii=False, indent=2) + handle.write("\n") + return fixture_path + + +def slugify_case_name(value: str) -> str: + """fixture 파일명에 쓸 수 있도록 case 이름을 느슨하게 정규화합니다.""" + + cleaned = value.strip().lower() + slug = re.sub(r"[^\w.-]+", "-", cleaned, flags=re.UNICODE) + slug = re.sub(r"-{2,}", "-", slug).strip("-._") + return slug or "case" diff --git a/tools/debug_streamlit.py b/tools/debug_streamlit.py deleted file mode 100644 index 922b17b..0000000 --- a/tools/debug_streamlit.py +++ /dev/null @@ -1,780 +0,0 @@ -"""Streamlit 기반 KMA API 디버그 카탈로그 뷰어.""" -# ruff: noqa: E402,I001 - -from __future__ import annotations - -from dataclasses import dataclass -from datetime import datetime -import json -import os -import sys -from pathlib import Path -from typing import Any - -ROOT = Path(__file__).resolve().parents[1] -SRC = ROOT / "src" -if str(SRC) not in sys.path: - sys.path.insert(0, str(SRC)) -for module_name, module in list(sys.modules.items()): - if module_name != "kma" and not module_name.startswith("kma."): - continue - module_file = getattr(module, "__file__", None) - if module_file is not None and not Path(module_file).resolve().is_relative_to(SRC): - del sys.modules[module_name] - -try: - import streamlit as st -except ModuleNotFoundError as exc: # pragma: no cover - 선택 실행 도구 - raise SystemExit('Streamlit UI를 쓰려면 `pip install -e ".[debug-ui]"`를 실행하세요.') from exc - -from kma import ( # noqa: E402 - DataGoKrClient, - DataGoKrItem, - MidForecastItem, - api_catalog, - api_key_for_gateway, - env_names_for_gateway, - load_local_env, -) -from kma.time_utils import ( # noqa: E402 - KST, - latest_mid_fcst_time, - latest_ultra_srt_fcst_base, - latest_ultra_srt_ncst_base, - latest_vilage_base, -) - - -@dataclass(frozen=True) -class ParameterSpec: - """디버그 UI에서 요청 파라미터 입력 폼을 만들기 위한 최소 명세.""" - - name: str - required: bool - label: str - placeholder: str = "" - help: str = "" - default: str = "" - - -def _param( - name: str, - *, - required: bool = True, - label: str | None = None, - placeholder: str = "", - help: str = "", - default: str = "", -) -> ParameterSpec: - return ParameterSpec( - name=name, - required=required, - label=label or name, - placeholder=placeholder, - help=help, - default=default, - ) - - -_DEFAULT_STATION = "108" -_DEFAULT_GRID_X = "60" -_DEFAULT_GRID_Y = "127" -_DEFAULT_AREA_NO = "1100000000" - - -def main() -> None: - st.set_page_config(page_title="KMA API Debug", layout="wide") - st.title("KMA API Debug") - - source = st.sidebar.selectbox("Data source", ["datagokr", "apihub"]) - rows = list(api_catalog(gateway=source)) - selected_label = st.sidebar.selectbox("API", [row.label for row in rows]) - selected = rows[[row.label for row in rows].index(selected_label)] - st.sidebar.caption("API full name") - st.sidebar.write(_api_full_name(selected)) - st.sidebar.caption(_api_description(selected)) - - env_names = env_names_for_gateway(selected.gateway) - default_key = _default_key(selected.gateway) - env_sources = _env_key_sources(selected.gateway) - - environment = "manual" - if env_sources: - st.sidebar.subheader("Environment") - environment = st.sidebar.selectbox("Environment", ["env", "manual"]) - if environment == "env": - source_info = env_sources[0] - st.sidebar.caption( - f"{source_info['name']} 값을 사용합니다. Source: {source_info['source']}" - ) - - st.sidebar.subheader("Auth") - if environment == "manual": - api_key = st.sidebar.text_input( - f"{selected.credential_param}", - value="", - type="password", - placeholder="직접 입력", - help=f"사용 가능한 env 이름: {', '.join(env_names)}", - ) - effective_api_key = api_key - else: - effective_api_key = default_key - _service_key_links(selected) - - timeout = st.sidebar.number_input( - "Timeout", - min_value=1.0, - max_value=60.0, - value=10.0, - step=1.0, - help="API 요청 timeout seconds입니다.", - ) - fixture_base_dir = _fixture_base_dir_sidebar() - - tabs = st.tabs( - [ - "Raw Response", - "Pydantic Model", - "Processed Result", - "Validation Errors", - "Debug Trace", - "Fixture / Testcase", - ] - ) - - with tabs[0]: - _raw_response_tab(selected, effective_api_key, timeout=float(timeout)) - with tabs[1]: - _pydantic_model_tab(selected) - with tabs[2]: - _processed_result_tab(selected) - with tabs[3]: - _validation_errors_tab(selected) - with tabs[4]: - _debug_trace_tab(rows, selected, env_names) - with tabs[5]: - _fixture_tab(fixture_base_dir) - - -def _raw_response_tab(selected: Any, api_key: str, *, timeout: float) -> None: - st.subheader(selected.dataset_name) - st.caption(f"{selected.gateway} / {selected.service or '-'} / {selected.operation or '-'}") - if selected.gateway != "datagokr" or not selected.service or not selected.operation: - st.info("APIHub 연계 항목은 APIHub 함수형 wrapper에서 endpoint를 선택해 호출합니다.") - return - - try: - submitted, params, extra_params, request_options, missing = _request_form(selected) - except ValueError as exc: - st.error(str(exc)) - return - preview = { - **params, - **extra_params, - "pageNo": request_options["page_no"], - "numOfRows": request_options["num_of_rows"], - "dataType": request_options["data_type"], - } - st.subheader("Request params preview") - st.json(preview) - - if not submitted: - return - if missing: - st.error("필수 파라미터를 입력하세요: " + ", ".join(missing)) - return - - try: - params.update(extra_params) - client = DataGoKrClient(api_key, timeout=timeout) - body = client.request( - selected.service, - selected.operation, - params, - data_type=request_options["data_type"], - page_no=request_options["page_no"], - num_of_rows=request_options["num_of_rows"], - ) - except Exception as exc: # pragma: no cover - UI 표시 - _store_run( - selected, - body=None, - request_params=preview, - model_name=None, - models=[], - validation_errors=[str(exc)], - ) - st.error(str(exc)) - return - model_name, models, validation_errors = _parse_models(selected, body) - _store_run( - selected, - body=body, - request_params=preview, - model_name=model_name, - models=models, - validation_errors=validation_errors, - ) - st.json(body) - - -def _request_form( - selected: Any, -) -> tuple[bool, dict[str, Any], dict[str, Any], dict[str, Any], list[str]]: - specs = _parameter_specs(selected.service, selected.operation) - required_specs = [spec for spec in specs if spec.required] - optional_specs = [spec for spec in specs if not spec.required] - key_prefix = f"{selected.dataset_id}:{selected.service}:{selected.operation}" - - with st.form(f"request-form:{key_prefix}"): - st.subheader("Required parameters") - if required_specs: - required_values = _render_param_grid(required_specs, key_prefix=key_prefix) - else: - st.caption("이 API에 대해 로컬에 정리된 필수 파라미터 명세가 없습니다.") - required_values = {} - - st.subheader("Optional parameters") - optional_values = _render_param_grid(optional_specs, key_prefix=key_prefix) - page_no, num_of_rows, data_type = _render_common_options(key_prefix) - - extra_text = st.text_area( - "Extra params JSON", - value="{}", - height=110, - help="폼에 없는 provider 파라미터를 JSON object로 추가합니다.", - key=f"{key_prefix}:extra", - ) - submitted = st.form_submit_button("Run selected API") - - params = {**required_values, **optional_values} - missing = [spec.name for spec in required_specs if not str(params.get(spec.name, "")).strip()] - extra_params = _parse_extra_params(extra_text) - return ( - submitted, - {key: value for key, value in params.items() if str(value).strip()}, - extra_params, - {"page_no": page_no, "num_of_rows": num_of_rows, "data_type": data_type}, - missing, - ) - - -def _render_param_grid(specs: list[ParameterSpec], *, key_prefix: str) -> dict[str, str]: - values: dict[str, str] = {} - for index in range(0, len(specs), 2): - columns = st.columns(2) - for column, spec in zip(columns, specs[index : index + 2]): - with column: - values[spec.name] = st.text_input( - spec.label, - value=spec.default, - placeholder=spec.placeholder, - help=spec.help or None, - key=f"{key_prefix}:param:{spec.name}", - ) - return values - - -def _render_common_options(key_prefix: str) -> tuple[int, int, str]: - col1, col2, col3 = st.columns(3) - with col1: - page_no = st.number_input( - "pageNo", - min_value=1, - value=1, - step=1, - help="공공데이터포털 paging 파라미터입니다.", - key=f"{key_prefix}:pageNo", - ) - with col2: - num_of_rows = st.number_input( - "numOfRows", - min_value=1, - value=10, - step=1, - help="한 페이지에 받을 row 수입니다.", - key=f"{key_prefix}:numOfRows", - ) - with col3: - data_type = st.selectbox( - "dataType", - ["JSON", "XML"], - index=0, - help="기본값은 JSON입니다.", - key=f"{key_prefix}:dataType", - ) - return int(page_no), int(num_of_rows), str(data_type) - - -def _parse_extra_params(text: str) -> dict[str, Any]: - try: - payload = json.loads(text or "{}") - except json.JSONDecodeError as exc: - raise ValueError(f"Extra params JSON is invalid: {exc}") from exc - if not isinstance(payload, dict): - raise ValueError("Extra params JSON must be an object") - return { - key: value - for key, value in payload.items() - if key not in {"serviceKey", "ServiceKey", "authKey", "pageNo", "numOfRows", "dataType"} - } - - -def _parameter_specs(service: str | None, operation: str | None) -> tuple[ParameterSpec, ...]: - if not service or not operation: - return () - endpoint = (service, operation) - if endpoint in _PARAMETER_SPECS: - return _PARAMETER_SPECS[endpoint]() - if service == "MidFcstInfoService": - return _mid_forecast_specs(operation) - if service == "BeachInfoservice": - return _beach_specs(operation) - if service == "VilageFcstMsgService": - return _forecast_message_specs(operation) - if service == "LivingWthrIdxServiceV4": - return _living_weather_specs(operation) - if service == "EqkInfoService": - return _date_range_specs() - return () - - -def _short_forecast_specs() -> tuple[ParameterSpec, ...]: - base_date, base_time = latest_ultra_srt_fcst_base() - return _base_grid_specs(base_date, base_time) - - -def _now_specs() -> tuple[ParameterSpec, ...]: - base_date, base_time = latest_ultra_srt_ncst_base() - return _base_grid_specs(base_date, base_time) - - -def _vilage_specs() -> tuple[ParameterSpec, ...]: - base_date, base_time = latest_vilage_base() - return _base_grid_specs(base_date, base_time) - - -def _base_grid_specs(base_date: str, base_time: str) -> tuple[ParameterSpec, ...]: - return ( - _param("base_date", label="base_date (YYYYMMDD)", default=base_date), - _param("base_time", label="base_time (HHMM)", default=base_time), - _param("nx", label="nx", default=_DEFAULT_GRID_X, help="KMA DFS 격자 x입니다."), - _param("ny", label="ny", default=_DEFAULT_GRID_Y, help="KMA DFS 격자 y입니다."), - ) - - -def _version_specs() -> tuple[ParameterSpec, ...]: - return ( - _param("ftype", default="ODAM", help="예보 버전 조회 타입입니다."), - _param("basedatetime", label="basedatetime (YYYYMMDDHHMM)", default=_now("%Y%m%d%H%M")), - ) - - -def _mid_forecast_specs(operation: str) -> tuple[ParameterSpec, ...]: - if operation == "getMidFcst": - return ( - _param("stnId", default=_DEFAULT_STATION, help="전국 예보 통보문 지점 코드입니다."), - _param("tmFc", label="tmFc (YYYYMMDDHHMM)", default=latest_mid_fcst_time()), - ) - default_reg_id = "11B10101" if operation == "getMidTa" else "11B00000" - return ( - _param("regId", default=default_reg_id, help="중기예보 권역 코드입니다."), - _param("tmFc", label="tmFc (YYYYMMDDHHMM)", default=latest_mid_fcst_time()), - ) - - -def _asos_daily_specs() -> tuple[ParameterSpec, ...]: - today = _now("%Y%m%d") - return ( - _param("startDt", label="startDt (YYYYMMDD)", default=today), - _param("endDt", label="endDt (YYYYMMDD)", default=today), - _param("dataCd", default="ASOS"), - _param("dateCd", default="DAY"), - _param("stnIds", required=False, default=_DEFAULT_STATION, help="비우면 전체 지점입니다."), - ) - - -def _asos_hourly_specs() -> tuple[ParameterSpec, ...]: - today = _now("%Y%m%d") - return ( - _param("startDt", label="startDt (YYYYMMDD)", default=today), - _param("startHh", label="startHh (HH)", default="00"), - _param("endDt", label="endDt (YYYYMMDD)", default=today), - _param("endHh", label="endHh (HH)", default="23"), - _param("dataCd", default="ASOS"), - _param("dateCd", default="HR"), - _param("stnIds", required=False, default=_DEFAULT_STATION, help="비우면 전체 지점입니다."), - ) - - -def _weather_warning_specs() -> tuple[ParameterSpec, ...]: - today = _now("%Y%m%d") - return ( - _param("stnId", default=_DEFAULT_STATION), - _param("fromTmFc", label="fromTmFc (YYYYMMDD)", default=today), - _param("toTmFc", label="toTmFc (YYYYMMDD)", default=today), - ) - - -def _forecast_message_specs(operation: str) -> tuple[ParameterSpec, ...]: - if operation == "getWthrSituation": - return (_param("stnId", default=_DEFAULT_STATION),) - if operation in {"getLandFcst", "getSeaFcst"}: - return (_param("regId", default="11B00000"),) - return () - - -def _beach_specs(operation: str) -> tuple[ParameterSpec, ...]: - if operation == "getUltraSrtFcstBeach": - base_date, base_time = latest_ultra_srt_fcst_base() - return ( - _param("beach_num", default="1", help="해수욕장 코드입니다."), - _param("base_date", label="base_date (YYYYMMDD)", default=base_date), - _param("base_time", label="base_time (HHMM)", default=base_time), - ) - if operation == "getVilageFcstBeach": - base_date, base_time = latest_vilage_base() - return ( - _param("beach_num", default="1", help="해수욕장 코드입니다."), - _param("base_date", label="base_date (YYYYMMDD)", default=base_date), - _param("base_time", label="base_time (HHMM)", default=base_time), - ) - if operation in {"getWhBuoyBeach", "getTwBuoyBeach"}: - return ( - _param("beach_num", default="1", help="해수욕장 코드입니다."), - _param("searchTime", label="searchTime (YYYYMMDDHHMM)", default=_now("%Y%m%d%H%M")), - ) - if operation == "getSunInfoBeach": - return ( - _param("beach_num", default="1", help="해수욕장 코드입니다."), - _param("Base_date", label="Base_date (YYYYMMDD)", default=_now("%Y%m%d")), - ) - return ( - _param("beach_num", default="1", help="해수욕장 코드입니다."), - _param("base_date", label="base_date (YYYYMMDD)", default=_now("%Y%m%d")), - ) - - -def _tour_village_specs() -> tuple[ParameterSpec, ...]: - return ( - _param("courseId", default="1"), - _param("currentDate", label="currentDate (YYYYMMDD)", default=_now("%Y%m%d")), - _param("hour", label="hour (HH)", default=_now("%H")), - ) - - -def _city_tour_specs() -> tuple[ParameterSpec, ...]: - return ( - _param("cityAreaId", default="1"), - _param("currentDate", label="currentDate (YYYYMMDD)", default=_now("%Y%m%d")), - _param("day", default="0"), - ) - - -def _living_weather_specs(operation: str) -> tuple[ParameterSpec, ...]: - specs = [ - _param("areaNo", default=_DEFAULT_AREA_NO), - _param("time", label="time (YYYYMMDDHH)", default=_now("%Y%m%d%H")), - ] - if operation == "getSenTaIdxV4": - specs.append(_param("requestCode", default="A01")) - return tuple(specs) - - -def _date_range_specs() -> tuple[ParameterSpec, ...]: - today = _now("%Y%m%d") - return ( - _param("fromTmFc", label="fromTmFc (YYYYMMDD)", default=today), - _param("toTmFc", label="toTmFc (YYYYMMDD)", default=today), - ) - - -def _now(fmt: str) -> str: - return datetime.now(tz=KST).strftime(fmt) - - -_PARAMETER_SPECS: dict[tuple[str, str], Any] = { - ("VilageFcstInfoService_2.0", "getUltraSrtNcst"): _now_specs, - ("VilageFcstInfoService_2.0", "getUltraSrtFcst"): _short_forecast_specs, - ("VilageFcstInfoService_2.0", "getVilageFcst"): _vilage_specs, - ("VilageFcstInfoService_2.0", "getFcstVersion"): _version_specs, - ("AsosDalyInfoService", "getWthrDataList"): _asos_daily_specs, - ("AsosHourlyInfoService", "getWthrDataList"): _asos_hourly_specs, - ("WthrWrnInfoService", "getWthrWrnList"): _weather_warning_specs, - ("TourStnInfoService1", "getTourStnVilageFcst1"): _tour_village_specs, - ("TourStnInfoService1", "getCityTourClmIdx1"): _city_tour_specs, -} - - -def _api_full_name(selected: Any) -> str: - if selected.service and selected.operation: - return f"{selected.dataset_name} / {selected.service} / {selected.operation}" - return f"{selected.dataset_name} / {selected.gateway}" - - -def _api_description(selected: Any) -> str: - key = (selected.service, selected.operation) - if key in _API_DESCRIPTIONS: - return _API_DESCRIPTIONS[key] - if selected.gateway == "apihub": - return "data.go.kr 카탈로그에서 APIHub로 연결되는 기상청 API입니다." - if selected.operation: - return f"{selected.dataset_name}의 {selected.operation} operation입니다." - return f"{selected.dataset_name} API입니다." - - -_API_DESCRIPTIONS: dict[tuple[str | None, str | None], str] = { - ("VilageFcstInfoService_2.0", "getUltraSrtNcst"): ( - "초단기실황 관측값을 KMA DFS 격자 기준으로 조회합니다." - ), - ("VilageFcstInfoService_2.0", "getUltraSrtFcst"): ( - "초단기예보를 발표시각과 격자 좌표 기준으로 조회합니다." - ), - ("VilageFcstInfoService_2.0", "getVilageFcst"): ( - "단기예보를 발표시각과 격자 좌표 기준으로 조회합니다." - ), - ("VilageFcstInfoService_2.0", "getFcstVersion"): ( - "예보 데이터의 버전 metadata를 조회합니다." - ), - ("MidFcstInfoService", "getMidFcst"): "전국 중기예보 통보문을 조회합니다.", - ("MidFcstInfoService", "getMidLandFcst"): ( - "중기 육상예보를 권역 코드와 발표시각 기준으로 조회합니다." - ), - ("MidFcstInfoService", "getMidTa"): ( - "중기 기온예보를 권역 코드와 발표시각 기준으로 조회합니다." - ), - ("MidFcstInfoService", "getMidSeaFcst"): ( - "중기 해상예보를 권역 코드와 발표시각 기준으로 조회합니다." - ), - ("AsosDalyInfoService", "getWthrDataList"): ( - "ASOS 지상 관측 일자료를 기간과 지점 기준으로 조회합니다." - ), - ("AsosHourlyInfoService", "getWthrDataList"): ( - "ASOS 지상 관측 시간자료를 기간과 지점 기준으로 조회합니다." - ), - ("WthrWrnInfoService", "getWthrWrnList"): "기상특보 목록을 지점과 발표일 범위로 조회합니다.", - ("BeachInfoservice", "getUltraSrtFcstBeach"): ( - "해수욕장 초단기예보를 해변 코드와 발표시각 기준으로 조회합니다." - ), - ("BeachInfoservice", "getVilageFcstBeach"): ( - "해수욕장 단기예보를 해변 코드와 발표시각 기준으로 조회합니다." - ), - ("BeachInfoservice", "getWhBuoyBeach"): "해수욕장 주변 파고 관측값을 조회합니다.", - ("BeachInfoservice", "getTideInfoBeach"): "해수욕장 조석 정보를 조회합니다.", - ("BeachInfoservice", "getSunInfoBeach"): "해수욕장 일출/일몰 정보를 조회합니다.", - ("BeachInfoservice", "getTwBuoyBeach"): "해수욕장 주변 수온 관측값을 조회합니다.", -} - - -def _service_key_links(selected: Any) -> None: - st.sidebar.caption("Service key links") - st.sidebar.link_button( - f"{selected.credential_param} 발급/확인", - selected.service_key_url, - ) - if selected.portal_url != selected.service_key_url: - st.sidebar.link_button("data.go.kr 카탈로그", selected.portal_url) - - -def _env_key_sources(gateway: str) -> list[dict[str, str]]: - names = env_names_for_gateway(gateway) - sources: list[dict[str, str]] = [] - for name in names: - value = os.getenv(name) - if value is not None and value.strip(): - sources.append({"name": name, "source": "process env"}) - return sources - - local_env = load_local_env() - for name in names: - value = local_env.get(name) - if value is not None and value.strip(): - sources.append({"name": name, "source": ".env 또는 .env.local"}) - return sources - return sources - - -def _fixture_base_dir_sidebar() -> str: - st.sidebar.subheader("Fixtures") - candidates = _fixture_dir_candidates() - options = [str(path) for path in candidates] - custom_label = "Custom..." - selected = st.sidebar.selectbox("Fixture base dir", [*options, custom_label]) - if selected == custom_label: - selected = st.sidebar.text_input( - "Custom fixture base dir", - value=str((ROOT / "tests" / "fixtures").resolve()), - ) - st.sidebar.caption(selected) - return selected - - -def _fixture_dir_candidates() -> list[Path]: - preferred = [ - ROOT / "tests" / "fixtures", - ROOT / "tests", - ROOT / "tools", - ROOT, - ] - candidates: list[Path] = [] - for path in preferred: - resolved = path.resolve() - if resolved not in candidates: - candidates.append(resolved) - return candidates - - -def _pydantic_model_tab(selected: Any) -> None: - run = _current_run(selected) - if run is None: - st.info("Raw Response 탭에서 선택한 API를 실행하면 여기에서 Pydantic 모델을 확인합니다.") - return - if run["validation_errors"]: - st.warning("모델 파싱 중 확인할 내용이 있습니다. Validation Errors 탭을 확인하세요.") - if not run["models"]: - st.info("응답에서 `body.items.item` row를 찾지 못해 Pydantic row 모델을 만들지 않았습니다.") - if run["body"] is not None: - st.json(run["body"]) - return - - st.caption(f"{run['model_name']} · {len(run['models'])} rows") - st.json(run["models"]) - - -def _processed_result_tab(selected: Any) -> None: - run = _current_run(selected) - if run is None: - st.info("Raw Response 탭에서 API를 실행하면 처리된 row preview를 표시합니다.") - return - if not run["models"]: - st.info("표시할 처리 결과가 없습니다.") - return - - rows = [model.get("raw", model) for model in run["models"]] - st.dataframe(rows, width="stretch", hide_index=True) - - -def _validation_errors_tab(selected: Any) -> None: - run = _current_run(selected) - if run is None: - st.info("아직 실행된 API가 없습니다.") - return - if not run["validation_errors"]: - st.success("현재 실행 결과에서 validation error가 없습니다.") - return - for error in run["validation_errors"]: - st.error(error) - - -def _fixture_tab(fixture_base_dir: str) -> None: - st.info("Fixture 저장 기능은 replay runner와 함께 별도 단계에서 연결합니다.") - st.caption("Fixture base dir") - st.code(fixture_base_dir, language=None) - - -def _parse_models(selected: Any, body: Any) -> tuple[str | None, list[dict[str, Any]], list[str]]: - try: - rows = _items_from_body(body) - except ValueError as exc: - return None, [], [str(exc)] - - models: list[dict[str, Any]] = [] - errors: list[str] = [] - model_name = "MidForecastItem" if selected.service == "MidFcstInfoService" else "DataGoKrItem" - for index, row in enumerate(rows): - try: - if selected.service == "MidFcstInfoService": - model = MidForecastItem( - operation=str(selected.operation), - tm_fc=_str_or_none(row.get("tmFc")), - reg_id=_str_or_none(row.get("regId")), - stn_id=_str_or_none(row.get("stnId")), - raw=dict(row), - ) - else: - model = DataGoKrItem( - service=str(selected.service), - operation=str(selected.operation), - raw=dict(row), - ) - models.append(model.model_dump(mode="json")) - except Exception as exc: - errors.append(f"row {index}: {exc}") - return model_name, models, errors - - -def _items_from_body(body: Any) -> list[dict[str, Any]]: - try: - raw_items = body["items"]["item"] - except (KeyError, TypeError) as exc: - raise ValueError("response body does not contain `items.item`") from exc - if isinstance(raw_items, dict): - return [raw_items] - if isinstance(raw_items, list): - return [dict(item) for item in raw_items if isinstance(item, dict)] - raise ValueError("response `items.item` is not an object or list") - - -def _store_run( - selected: Any, - *, - body: Any, - request_params: dict[str, Any], - model_name: str | None, - models: list[dict[str, Any]], - validation_errors: list[str], -) -> None: - st.session_state["last_run"] = { - "selection_key": _selection_key(selected), - "body": body, - "request_params": request_params, - "model_name": model_name, - "models": models, - "validation_errors": validation_errors, - } - - -def _current_run(selected: Any) -> dict[str, Any] | None: - run = st.session_state.get("last_run") - if not isinstance(run, dict): - return None - if run.get("selection_key") != _selection_key(selected): - return None - return run - - -def _selection_key(selected: Any) -> str: - return f"{selected.gateway}:{selected.dataset_id}:{selected.service}:{selected.operation}" - - -def _str_or_none(value: Any) -> str | None: - if value is None: - return None - text = str(value).strip() - return text or None - - -def _debug_trace_tab(rows: list[Any], selected: Any, env_names: tuple[str, ...]) -> None: - st.subheader("Catalog") - st.dataframe( - [row.asdict() for row in rows], - width="stretch", - hide_index=True, - ) - - st.subheader("Selected API") - st.json(selected.asdict()) - st.link_button(f"{selected.credential_param} 발급/확인", selected.service_key_url) - st.caption(f"credential env: {', '.join(env_names)}") - - -def _default_key(gateway: str) -> str: - try: - return api_key_for_gateway(gateway) - except ValueError: - return "" - - -if __name__ == "__main__": - main()