Kosis Stats

#!/usr/bin/env python3
"""Read-only KOSIS (kosis.kr) Open API helper.

The script wraps four KOSIS endpoints needed to answer everyday Korean
official-statistics questions:

    - statisticsSearch.do        : keyword search of statistical tables
    - statisticsData.do?getMeta  : table metadata (dimensions, units)
    - statisticsParameterData.do : actual data cells filtered by classifier
    - statisticsBigData.do       : large datasets (requires userStatsId)

It only reads. It never registers user statistics, edits anything, or
performs aggressive polling.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any

SEARCH_URL = "https://kosis.kr/openapi/statisticsSearch.do"
META_URL = "https://kosis.kr/openapi/statisticsData.do"
DATA_URL = "https://kosis.kr/openapi/Param/statisticsParameterData.do"
BIGDATA_URL = "https://kosis.kr/openapi/statisticsBigData.do"
PROXY_BASE_URL_ENV_VAR = "KSKILL_PROXY_BASE_URL"
DEFAULT_PROXY_BASE_URL = "https://k-skill-proxy.nomadamas.org"

DEFAULT_TIMEOUT = 30
PRD_SE_VALUES = {"M", "Q", "S", "Y", "F", "IR"}
# `xls` is intentionally omitted: KOSIS returns it as a binary Excel payload,
# but the helper streams text-only output. Use bigdata json/sdmx/csv (text)
# for now; download xls files manually from the KOSIS web UI if you need them.
BIGDATA_FORMATS = {"json", "sdmx", "csv"}

ERROR_CODE_HINTS: dict[str, str] = {
    "10": "인증키가 누락되었습니다. KSKILL_KOSIS_API_KEY 환경변수를 확인하세요.",
    "11": "인증키가 만료되었거나 해당 endpoint에서 무효입니다. https://kosis.kr/openapi/ 에서 갱신하거나, bigdata는 본인이 등록한 userStatsId 인지 확인하세요.",
    "20": "필수 요청 변수가 누락되었습니다. `meta --table-id <ID> --meta-type ITM --json` 으로 ITM 안에 들어 있는 OBJ_ID(분류 차원)와 코드를 확인하세요(많은 표가 OBJ 메타는 비어 있고 분류가 ITM 안에 들어 있음). 그 뒤 `--obj-l 1=<코드> --obj-l 2=<코드>` 형태로 필요한 차원을 모두 지정해 재호출하세요. 별도 OBJ 메타가 있는 표는 `--meta-type OBJ` 로도 확인 가능합니다.",
    "21": "잘못된 요청 변수입니다. orgId/tblId/기간 형식을 재확인하세요. tblId가 의심되면 `search --query <키워드>` 로 정확한 ID를 다시 찾으세요.",
    "30": "조회 결과가 없습니다. 키워드를 더 짧게(예: '1인 가구' → '가구') 또는 다른 표현으로 재검색하거나, 기간/분류 필터를 완화하세요. meta 호출에서 이 에러가 나면 해당 메타 타입을 표가 지원하지 않는 경우이므로 다른 `--meta-type` 을 시도하세요.",
    "31": "조회 결과가 한도(40,000셀)를 초과했습니다. 기간을 좁히거나(예: 5년→1년) 분류 필터의 ALL 을 특정 코드로 바꾸세요(예: `--obj-l 1=ALL` → `--obj-l 1=11` 서울만). 그래도 부족하면 `bigdata` 서브커맨드 + 사전 등록한 userStatsId 를 사용하세요.",
    "40": "분당 호출 한도(1,000건)를 초과했습니다. 잠시 대기 후 재시도하거나 호출 간 sleep 을 두세요.",
    "41": "1회 호출 ROW 한도를 초과했습니다. 기간이나 분류를 좁혀 쿼리를 분할하세요.",
    "42": "사용자별 이용이 제한되었습니다. KOSIS 운영팀에 문의하세요.",
    "50": "KOSIS 서버 오류입니다. 1~2초 대기 후 재시도하세요.",
}


@dataclass
class KosisConfig:
    api_key: str
    timeout: int = DEFAULT_TIMEOUT


class KosisError(RuntimeError):
    def __init__(self, code: str | None, message: str) -> None:
        self.code = code or ""
        super().__init__(message)


def parse_resultcount(value: str) -> int:
    try:
        result = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be an integer") from exc
    if not 1 <= result <= 5000:
        raise argparse.ArgumentTypeError("must be between 1 and 5000")
    return result


def parse_prd_se(value: str) -> str:
    upper = value.strip().upper()
    if upper not in PRD_SE_VALUES:
        raise argparse.ArgumentTypeError(
            "must be one of: " + ", ".join(sorted(PRD_SE_VALUES))
        )
    return upper


def parse_bigdata_format(value: str) -> str:
    lower = value.strip().lower()
    if lower not in BIGDATA_FORMATS:
        raise argparse.ArgumentTypeError(
            "must be one of: " + ", ".join(sorted(BIGDATA_FORMATS))
        )
    return lower


def _add_common_flags(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--timeout",
        type=int,
        default=DEFAULT_TIMEOUT,
        help=f"HTTP timeout in seconds (default {DEFAULT_TIMEOUT}).",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print the request URL and parameters without calling KOSIS.",
    )
    parser.add_argument(
        "--proxy-base-url",
        help=(
            "k-skill-proxy base URL for search/meta/data "
            f"(default {DEFAULT_PROXY_BASE_URL}; override with {PROXY_BASE_URL_ENV_VAR})."
        ),
    )
    parser.add_argument(
        "--direct",
        action="store_true",
        help="Call KOSIS directly with KSKILL_KOSIS_API_KEY instead of k-skill-proxy.",
    )
    output = parser.add_mutually_exclusive_group()
    output.add_argument("--json", action="store_true", help="Print JSON output.")
    output.add_argument("--text", action="store_true", help="Print human-readable output.")


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read-only KOSIS Open API helper. "
        "Place output flags (--json/--text/--dry-run/--timeout) AFTER the subcommand.",
    )

    sub = parser.add_subparsers(dest="command", required=True)

    search = sub.add_parser("search", help="Search statistical tables by keyword.")
    _add_common_flags(search)
    search.add_argument("--query", required=True, help="Korean keyword (e.g. '1인 가구').")
    search.add_argument(
        "--result-count",
        type=parse_resultcount,
        default=20,
        help="Result count, 1-5000 (default 20).",
    )
    search.add_argument(
        "--start-count",
        type=int,
        default=1,
        help="Result offset for pagination (default 1).",
    )

    meta = sub.add_parser("meta", help="Fetch table metadata (dimensions, units).")
    _add_common_flags(meta)
    meta.add_argument("--org-id", default="101", help="Organization ID (default 101).")
    meta.add_argument("--table-id", required=True, help="KOSIS table ID, e.g. DT_1IN0001.")
    meta.add_argument(
        "--meta-type",
        default="TBL",
        choices=["TBL", "ITM", "OBJ"],
        help="Meta type (default TBL).",
    )

    data = sub.add_parser("data", help="Fetch table data filtered by classifiers.")
    _add_common_flags(data)
    data.add_argument("--org-id", default="101", help="Organization ID (default 101).")
    data.add_argument("--table-id", required=True, help="KOSIS table ID.")
    data.add_argument(
        "--prd-se",
        type=parse_prd_se,
        required=True,
        help="Period frequency: M Q S Y F IR.",
    )
    data.add_argument("--start", required=True, help="Start period (format depends on --prd-se).")
    data.add_argument("--end", required=True, help="End period.")
    data.add_argument("--itm-id", default="ALL", help="Item ID filter (default ALL).")
    data.add_argument(
        "--obj-l",
        action="append",
        default=[],
        metavar="N=VALUE",
        help="Classifier filter, e.g. --obj-l 1=ALL --obj-l 2=00. Repeatable. "
        "If omitted, --obj-l 1=ALL is used.",
    )

    bigdata = sub.add_parser(
        "bigdata",
        help="Fetch large datasets via statisticsBigData (requires userStatsId).",
    )
    _add_common_flags(bigdata)
    bigdata.add_argument(
        "--user-stats-id",
        required=True,
        help="userStatsId pre-registered on KOSIS (개발가이드 > 대용량 통계자료 > URL생성).",
    )
    bigdata.add_argument(
        "--format",
        dest="bigdata_format",
        type=parse_bigdata_format,
        default="json",
        help="Output format: json sdmx csv xls (default json).",
    )
    bigdata.add_argument(
        "--prd-se",
        type=parse_prd_se,
        help="Period frequency override.",
    )
    bigdata.add_argument(
        "--new-est-prd-cnt",
        type=int,
        help="Count of latest periods to fetch (alternative to start/end).",
    )

    return parser.parse_args(argv)


def load_secrets_env(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}
    secrets: dict[str, str] = {}
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            continue
        key, _, value = line.partition("=")
        secrets[key.strip()] = value.strip().strip('"').strip("'")
    return secrets


def resolve_api_key(
    *,
    env: dict[str, str] | None = None,
    secrets_path: Path | None = None,
) -> str:
    env_map = env if env is not None else os.environ
    direct = env_map.get("KSKILL_KOSIS_API_KEY")
    if direct:
        return direct.strip()

    candidate = secrets_path or Path("~/.config/k-skill/secrets.env").expanduser()
    secrets = load_secrets_env(candidate)
    fallback = secrets.get("KSKILL_KOSIS_API_KEY")
    if fallback:
        return fallback.strip()

    raise SystemExit(
        "missing required environment variable: KSKILL_KOSIS_API_KEY\n"
        "발급: https://kosis.kr/openapi/ (무료, KOSIS 회원가입 후 활용신청)\n"
        "참조: kosis-stats/references/kosis-openapi-guide.md"
    )


def resolve_proxy_base_url(
    explicit_base_url: str | None = None,
    env: dict[str, str] | None = None,
) -> str:
    env_map = env if env is not None else os.environ
    candidate = (explicit_base_url or env_map.get(PROXY_BASE_URL_ENV_VAR) or "").strip()
    if candidate.casefold() in {"off", "false", "0", "disable", "disabled", "none"}:
        raise SystemExit(f"{PROXY_BASE_URL_ENV_VAR} is disabled; pass --direct to use KSKILL_KOSIS_API_KEY.")
    if candidate and candidate != "replace-me":
        return candidate.rstrip("/")
    return DEFAULT_PROXY_BASE_URL


def parse_obj_l(values: list[str]) -> dict[str, str]:
    objs: dict[str, str] = {}
    for raw in values:
        if "=" not in raw:
            raise SystemExit(f"--obj-l must be N=VALUE, got: {raw}")
        key, _, value = raw.partition("=")
        key = key.strip()
        if not key.isdigit() or not 1 <= int(key) <= 8:
            raise SystemExit(f"--obj-l index must be 1..8, got: {key}")
        objs[f"objL{key}"] = value.strip() or "ALL"
    if not objs:
        objs["objL1"] = "ALL"
    return objs


def build_search_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    return {
        "method": "getList",
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "searchNm": args.query,
        "resultCount": str(args.result_count),
        "startCount": str(args.start_count),
    }


def build_meta_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    return {
        "method": "getMeta",
        "type": args.meta_type,
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "orgId": args.org_id,
        "tblId": args.table_id,
    }


def build_data_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    params: dict[str, str] = {
        "method": "getList",
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "orgId": args.org_id,
        "tblId": args.table_id,
        "itmId": args.itm_id,
        "prdSe": args.prd_se,
        "startPrdDe": args.start,
        "endPrdDe": args.end,
    }
    params.update(parse_obj_l(args.obj_l))
    return params


def build_bigdata_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    params: dict[str, str] = {
        "method": "getList",
        "apiKey": api_key,
        "format": args.bigdata_format,
        "jsonVD": "Y",
        "userStatsId": args.user_stats_id,
    }
    if args.prd_se:
        params["prdSe"] = args.prd_se
    if args.new_est_prd_cnt is not None:
        params["newEstPrdCnt"] = str(args.new_est_prd_cnt)
    return params


def build_url(base: str, params: dict[str, str]) -> str:
    query = urllib.parse.urlencode(params, quote_via=urllib.parse.quote)
    return f"{base}?{query}"


def fetch_text(url: str, timeout: int) -> str:
    request = urllib.request.Request(url, headers={"User-Agent": "k-skill/kosis-stats"})
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            charset = response.headers.get_content_charset() or "utf-8"
            return response.read().decode(charset, errors="replace")
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise KosisError(str(exc.code), f"HTTP {exc.code}: {body[:200]}") from exc
    except urllib.error.URLError as exc:
        raise KosisError(None, f"network error: {exc.reason}") from exc


def fix_unquoted_keys(text: str) -> str:
    """KOSIS sometimes returns JSON with unquoted keys."""
    return re.sub(r'([{,])\s*([A-Za-z_][A-Za-z0-9_]*)\s*:', r'\1"\2":', text)


def parse_kosis_json(text: str) -> Any:
    body = text.strip()
    try:
        return json.loads(body)
    except json.JSONDecodeError:
        return json.loads(fix_unquoted_keys(body))


_XML_ERROR_RE = re.compile(
    r"<error>\s*<err>([^<]*)</err>\s*<errMsg>([^<]*)</errMsg>", re.IGNORECASE
)


def detect_xml_error(text: str) -> KosisError | None:
    match = _XML_ERROR_RE.search(text)
    if not match:
        return None
    code = match.group(1).strip()
    message = match.group(2).strip()
    hint = ERROR_CODE_HINTS.get(code, "")
    full = f"KOSIS error {code or '?'}: {message}"
    if hint:
        full += f" ({hint})"
    return KosisError(code, full)


def detect_kosis_error(payload: Any) -> KosisError | None:
    if isinstance(payload, dict):
        message = payload.get("errMsg")
        if message:
            code = str(payload.get("err") or payload.get("errCode") or "").strip()
            hint = ERROR_CODE_HINTS.get(code, "")
            full = f"KOSIS error {code or '?'}: {message}"
            if hint:
                full += f" ({hint})"
            return KosisError(code, full)
    return None


def call_kosis(url: str, timeout: int, *, format_hint: str = "json") -> Any:
    text = fetch_text(url, timeout)
    xml_err = detect_xml_error(text)
    if xml_err is not None:
        raise xml_err
    if format_hint != "json":
        stripped = text.lstrip()
        if stripped.startswith("{") or stripped.startswith("["):
            try:
                payload = parse_kosis_json(text)
            except json.JSONDecodeError:
                return text
            err = detect_kosis_error(payload)
            if err is not None:
                raise err
        return text
    payload = parse_kosis_json(text)
    err = detect_kosis_error(payload)
    if err is not None:
        raise err
    return payload


def render_search_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "조회 결과가 없습니다. 키워드를 더 짧게(예: '1인 가구' → '가구') "
            "또는 다른 표현으로 재검색해 보세요. "
            "`--result-count` 와 `--start-count` 로 더 많은 후보를 페이징할 수도 있습니다."
        )
    lines = []
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        org = entry.get("ORG_NM", "?")
        tbl = entry.get("TBL_NM", "?")
        org_id = entry.get("ORG_ID", "?")
        tbl_id = entry.get("TBL_ID", "?")
        prd = f"{entry.get('STRT_PRD_DE', '?')}~{entry.get('END_PRD_DE', '?')}"
        lines.append(f"- [{org_id}/{tbl_id}] {org} / {tbl} ({prd})")
    lines.append(
        "\nNext: `meta --table-id <ID>` 로 분류·항목·단위 확인 → "
        "`data --table-id <ID> --prd-se <Y|M|Q|...> --start ... --end ...` 로 작은 슬라이스부터 받기."
    )
    return "\n".join(lines)


def render_meta_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "메타 정보가 없습니다. 표가 해당 메타 타입을 지원하지 않을 수 있습니다. "
            "다른 `--meta-type` (TBL/ITM/OBJ) 을 시도해 보세요."
        )
    lines = []
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        kr = entry.get("TBL_NM") or entry.get("ITM_NM") or entry.get("C_NM") or "?"
        en = entry.get("TBL_NM_ENG") or entry.get("ITM_NM_ENG") or entry.get("C_NM_ENG") or ""
        suffix = f" / {en}" if en else ""
        lines.append(f"- {kr}{suffix}")
    return "\n".join(lines)


def render_data_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "데이터가 없습니다. 기간(`--start`/`--end`), 항목(`--itm-id`), "
            "분류(`--obj-l`) 필터를 완화하거나 `meta` 로 표 구조를 다시 확인하세요."
        )
    lines = []
    units: set[str] = set()
    periods: set[str] = set()
    for entry in payload[:50]:
        if not isinstance(entry, dict):
            continue
        prd = str(entry.get("PRD_DE", "?"))
        unit = str(entry.get("UNIT_NM", "")).strip()
        item = entry.get("ITM_NM", "?")
        c1 = entry.get("C1_NM", "")
        value = entry.get("DT", "?")
        suffix = f" ({c1})" if c1 else ""
        unit_suffix = f" {unit}" if unit else ""
        lines.append(f"- {prd} | {item}{suffix} = {value}{unit_suffix}".rstrip())
        if unit:
            units.add(unit)
        periods.add(prd)
    if len(payload) > 50:
        lines.append(f"... ({len(payload) - 50} rows omitted; --json 으로 전체 받기)")
    summary_parts = [f"rows={len(payload)}"]
    if periods:
        period_list = sorted(p for p in periods if p and p != "?")
        if period_list:
            summary_parts.append(
                f"period={period_list[0]}~{period_list[-1]}"
                if len(period_list) > 1 else f"period={period_list[0]}"
            )
    if units:
        summary_parts.append("unit=" + ",".join(sorted(units)))
    else:
        summary_parts.append("unit=(KOSIS 응답에 UNIT_NM 미포함)")
    lines.append("\n[summary] " + ", ".join(summary_parts))
    return "\n".join(lines)


def render_text(command: str, payload: Any) -> str:
    if command == "search":
        return render_search_text(payload)
    if command == "meta":
        return render_meta_text(payload)
    if command == "data":
        return render_data_text(payload)
    return json.dumps(payload, ensure_ascii=False, indent=2)


def cite_endpoint(command: str) -> str:
    return {
        "search": SEARCH_URL,
        "meta": META_URL,
        "data": DATA_URL,
        "bigdata": BIGDATA_URL,
    }[command]


def should_use_proxy(args: argparse.Namespace) -> bool:
    return args.command in {"search", "meta", "data"} and not args.direct


def proxy_endpoint(command: str, base_url: str) -> str:
    path = {
        "search": "/v1/kosis/search",
        "meta": "/v1/kosis/meta",
        "data": "/v1/kosis/data",
    }[command]
    return f"{base_url.rstrip('/')}{path}"


def params_without_api_key(params: dict[str, str]) -> dict[str, str]:
    return {key: value for key, value in params.items() if key != "apiKey"}


def run(args: argparse.Namespace) -> int:
    use_json = args.json or not args.text

    use_proxy = should_use_proxy(args)

    if use_proxy or args.dry_run:
        api_key = "<PROXY>" if use_proxy else "<DRY-RUN>"
    else:
        api_key = resolve_api_key()

    builder = {
        "search": build_search_params,
        "meta": build_meta_params,
        "data": build_data_params,
        "bigdata": build_bigdata_params,
    }[args.command]
    base = cite_endpoint(args.command)
    params = builder(api_key, args)
    if use_proxy:
        call_base = proxy_endpoint(args.command, resolve_proxy_base_url(args.proxy_base_url))
        call_params = params_without_api_key(params)
    else:
        call_base = base
        call_params = params
    url = build_url(call_base, call_params)

    if args.dry_run:
        redacted = dict(call_params)
        if not use_proxy and "apiKey" in redacted:
            redacted["apiKey"] = "<DRY-RUN>"
        if use_json:
            print(json.dumps({
                "endpoint": call_base,
                "upstream_endpoint": base,
                "via_proxy": use_proxy,
                "params": redacted,
                "url": build_url(call_base, redacted)
            }, ensure_ascii=False, indent=2))
        else:
            print(f"endpoint: {call_base}")
            print(f"upstream_endpoint: {base}")
            print(f"via_proxy: {str(use_proxy).lower()}")
            print(f"url: {build_url(call_base, redacted)}")
            for key, value in redacted.items():
                print(f"  {key}={value}")
        return 0

    format_hint = params.get("format", "json")
    try:
        payload = call_kosis(url, args.timeout, format_hint=format_hint)
    except KosisError as exc:
        sys.stderr.write(f"{exc}\n")
        return 2

    if use_json:
        if isinstance(payload, str):
            sys.stdout.write(payload)
            if not payload.endswith("\n"):
                sys.stdout.write("\n")
        else:
            print(json.dumps(payload, ensure_ascii=False, indent=2))
    else:
        print(render_text(args.command, payload))
        print(f"\nsource: {base}")
        if use_proxy:
            print(f"via: {call_base}")

    return 0


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    return run(args)


if __name__ == "__main__":
    sys.exit(main())

[
  {
    "TBL_ID": "DT_1JC1501",
    "ORG_ID": "101",
    "PRD_SE": "Y",
    "PRD_DE": "2023",
    "ITM_ID": "T001",
    "ITM_NM": "1인 가구 비율",
    "UNIT_NM": "%",
    "C1": "00",
    "C1_NM": "전국",
    "DT": "35.5",
    "TBL_NM": "1인 가구 비율"
  },
  {
    "TBL_ID": "DT_1JC1501",
    "ORG_ID": "101",
    "PRD_SE": "Y",
    "PRD_DE": "2024",
    "ITM_ID": "T001",
    "ITM_NM": "1인 가구 비율",
    "UNIT_NM": "%",
    "C1": "00",
    "C1_NM": "전국",
    "DT": "36.2",
    "TBL_NM": "1인 가구 비율"
  }
]

[
  {
    "TBL_NM": "1인 가구 비율",
    "TBL_NM_ENG": "Single-person Household Ratio"
  }
]

[
  {
    "ORG_ID": "101",
    "ORG_NM": "통계청",
    "TBL_ID": "DT_1JC1501",
    "TBL_NM": "1인 가구 비율",
    "STAT_ID": "A11320180501112233",
    "STAT_NM": "인구주택총조사",
    "VW_CD": "MT_ZTITLE",
    "MT_ATITLE": "주제별 통계 > 인구·가구 > 가구",
    "STRT_PRD_DE": "2015",
    "END_PRD_DE": "2024",
    "TBL_VIEW_URL": "https://kosis.kr/statisticsList/example",
    "LINK_URL": "https://kosis.kr/statHtml/statHtml.do?orgId=101&tblId=DT_1JC1501",
    "QUERY": "1인 가구"
  }
]

import importlib.util
import io
import json
import os
import sys
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from unittest import mock


SCRIPT_DIR = Path(__file__).resolve().parent
HELPER_PATH = SCRIPT_DIR.parent / "scripts" / "run_kosis_stats.py"
FIXTURES_DIR = SCRIPT_DIR / "fixtures"


def load_helper():
    spec = importlib.util.spec_from_file_location("run_kosis_stats", HELPER_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"cannot load helper from {HELPER_PATH}")
    module = importlib.util.module_from_spec(spec)
    sys.modules["run_kosis_stats"] = module
    spec.loader.exec_module(module)
    return module


helper = load_helper()


def read_fixture(name: str) -> str:
    return (FIXTURES_DIR / name).read_text(encoding="utf-8")


class ParseArgsTest(unittest.TestCase):
    def test_search_subcommand_parses_query(self):
        args = helper.parse_args(["search", "--query", "인구"])
        self.assertEqual(args.command, "search")
        self.assertEqual(args.query, "인구")
        self.assertEqual(args.result_count, 20)

    def test_data_subcommand_requires_period_fields(self):
        args = helper.parse_args([
            "data", "--table-id", "DT_1JC1501",
            "--prd-se", "Y", "--start", "2020", "--end", "2023",
        ])
        self.assertEqual(args.prd_se, "Y")
        self.assertEqual(args.itm_id, "ALL")

    def test_meta_subcommand_defaults_org_id_to_101(self):
        args = helper.parse_args(["meta", "--table-id", "DT_1IN0001"])
        self.assertEqual(args.org_id, "101")
        self.assertEqual(args.meta_type, "TBL")

    def test_bigdata_subcommand_requires_user_stats_id(self):
        args = helper.parse_args([
            "bigdata", "--user-stats-id", "openapisample/101/DT_1IN1502/2/1/abc",
        ])
        self.assertEqual(args.user_stats_id, "openapisample/101/DT_1IN1502/2/1/abc")
        self.assertEqual(args.bigdata_format, "json")

    def test_invalid_prd_se_rejected(self):
        with self.assertRaises(SystemExit):
            helper.parse_args([
                "data", "--table-id", "X", "--prd-se", "X",
                "--start", "2020", "--end", "2021",
            ])

    def test_result_count_out_of_range_rejected(self):
        with self.assertRaises(SystemExit):
            helper.parse_args([
                "search", "--query", "x", "--result-count", "9999",
            ])


class CredentialResolutionTest(unittest.TestCase):
    def test_env_var_takes_precedence(self):
        env = {"KSKILL_KOSIS_API_KEY": "env-key"}
        secrets = SCRIPT_DIR / "fixtures" / "missing.env"
        self.assertEqual(helper.resolve_api_key(env=env, secrets_path=secrets), "env-key")

    def test_secrets_env_used_when_env_var_missing(self):
        secrets = SCRIPT_DIR / "fixtures" / "tmp_secrets.env"
        secrets.write_text("KSKILL_KOSIS_API_KEY=file-key\n", encoding="utf-8")
        try:
            self.assertEqual(
                helper.resolve_api_key(env={}, secrets_path=secrets), "file-key"
            )
        finally:
            secrets.unlink()

    def test_missing_credentials_exits_with_helpful_message(self):
        secrets = SCRIPT_DIR / "fixtures" / "missing.env"
        with self.assertRaises(SystemExit) as ctx:
            helper.resolve_api_key(env={}, secrets_path=secrets)
        message = str(ctx.exception)
        self.assertIn("KSKILL_KOSIS_API_KEY", message)
        self.assertIn("kosis.kr/openapi", message)

    def test_proxy_base_url_defaults_to_hosted_proxy(self):
        self.assertEqual(
            helper.resolve_proxy_base_url(env={}),
            "https://k-skill-proxy.nomadamas.org"
        )

    def test_proxy_base_url_env_override_is_trimmed(self):
        self.assertEqual(
            helper.resolve_proxy_base_url(env={"KSKILL_PROXY_BASE_URL": "https://proxy.example/"}),
            "https://proxy.example"
        )

    def test_proxy_base_url_can_be_disabled_for_direct_mode(self):
        with self.assertRaises(SystemExit):
            helper.resolve_proxy_base_url(env={"KSKILL_PROXY_BASE_URL": "off"})


class UrlBuilderTest(unittest.TestCase):
    def test_search_params_include_required_fields(self):
        args = helper.parse_args(["search", "--query", "인구"])
        params = helper.build_search_params("KEY", args)
        self.assertEqual(params["method"], "getList")
        self.assertEqual(params["searchNm"], "인구")
        self.assertEqual(params["apiKey"], "KEY")
        self.assertEqual(params["format"], "json")

    def test_data_params_include_obj_l_default(self):
        args = helper.parse_args([
            "data", "--table-id", "DT_1JC1501",
            "--prd-se", "Y", "--start", "2020", "--end", "2023",
        ])
        params = helper.build_data_params("KEY", args)
        self.assertEqual(params["objL1"], "ALL")
        self.assertEqual(params["prdSe"], "Y")
        self.assertEqual(params["startPrdDe"], "2020")
        self.assertEqual(params["endPrdDe"], "2023")

    def test_data_params_apply_obj_l_overrides(self):
        args = helper.parse_args([
            "data", "--table-id", "DT_X", "--prd-se", "Y",
            "--start", "2020", "--end", "2020",
            "--obj-l", "1=ALL", "--obj-l", "2=00",
        ])
        params = helper.build_data_params("KEY", args)
        self.assertEqual(params["objL1"], "ALL")
        self.assertEqual(params["objL2"], "00")

    def test_obj_l_rejects_bad_index(self):
        args = helper.parse_args([
            "data", "--table-id", "DT_X", "--prd-se", "Y",
            "--start", "2020", "--end", "2020", "--obj-l", "9=ALL",
        ])
        with self.assertRaises(SystemExit):
            helper.build_data_params("KEY", args)

    def test_meta_params_include_type_and_table(self):
        args = helper.parse_args(["meta", "--table-id", "DT_1IN0001"])
        params = helper.build_meta_params("KEY", args)
        self.assertEqual(params["method"], "getMeta")
        self.assertEqual(params["type"], "TBL")
        self.assertEqual(params["tblId"], "DT_1IN0001")

    def test_bigdata_params_include_format_and_optional_fields(self):
        args = helper.parse_args([
            "bigdata", "--user-stats-id", "abc/def",
            "--format", "sdmx", "--prd-se", "Y", "--new-est-prd-cnt", "5",
        ])
        params = helper.build_bigdata_params("KEY", args)
        self.assertEqual(params["format"], "sdmx")
        self.assertEqual(params["userStatsId"], "abc/def")
        self.assertEqual(params["prdSe"], "Y")
        self.assertEqual(params["newEstPrdCnt"], "5")

    def test_build_url_round_trip(self):
        url = helper.build_url("https://example.com/x", {"a": "1", "b": "한글"})
        self.assertTrue(url.startswith("https://example.com/x?"))
        self.assertIn("a=1", url)
        self.assertIn("b=", url)


class JsonHandlingTest(unittest.TestCase):
    def test_unquoted_keys_are_fixed(self):
        text = '{ORG_ID:"101",TBL_ID:"DT_X"}'
        fixed = helper.fix_unquoted_keys(text)
        self.assertEqual(json.loads(fixed), {"ORG_ID": "101", "TBL_ID": "DT_X"})

    def test_parse_kosis_json_handles_clean_json(self):
        payload = helper.parse_kosis_json('{"a":1,"b":"x"}')
        self.assertEqual(payload, {"a": 1, "b": "x"})

    def test_parse_kosis_json_handles_unquoted_keys(self):
        payload = helper.parse_kosis_json('{a:1,b:"x"}')
        self.assertEqual(payload, {"a": 1, "b": "x"})


class ErrorDetectionTest(unittest.TestCase):
    def test_err_field_detected(self):
        err = helper.detect_kosis_error({"err": "31", "errMsg": "조회결과 초과"})
        self.assertIsNotNone(err)
        self.assertEqual(err.code, "31")
        self.assertIn("31", str(err))
        # 31 hint 가 분할 + bigdata + 구체 예시 모두 안내
        self.assertIn("좁히", str(err))
        self.assertIn("obj-l", str(err))
        self.assertIn("bigdata", str(err))

    def test_errcode_field_detected(self):
        err = helper.detect_kosis_error({"errCode": "10", "errMsg": "인증키 누락"})
        self.assertIsNotNone(err)
        self.assertEqual(err.code, "10")
        self.assertIn("KSKILL_KOSIS_API_KEY", str(err))

    def test_code_20_hint_directs_to_meta(self):
        err = helper.detect_kosis_error({"err": "20", "errMsg": "필수요청변수값이 누락"})
        self.assertIsNotNone(err)
        self.assertIn("meta", str(err))
        self.assertIn("--obj-l", str(err))

    def test_code_21_hint_suggests_search(self):
        err = helper.detect_kosis_error({"err": "21", "errMsg": "잘못된 요청"})
        self.assertIsNotNone(err)
        self.assertIn("search", str(err))

    def test_code_30_hint_includes_keyword_relaxation(self):
        err = helper.detect_kosis_error({"err": "30", "errMsg": "결과 없음"})
        self.assertIsNotNone(err)
        self.assertIn("키워드", str(err))
        self.assertIn("meta-type", str(err))

    def test_unknown_code_still_reported(self):
        err = helper.detect_kosis_error({"err": "99", "errMsg": "알 수 없음"})
        self.assertIsNotNone(err)
        self.assertEqual(err.code, "99")

    def test_normal_payload_returns_none(self):
        self.assertIsNone(helper.detect_kosis_error([{"DT": "1"}]))
        self.assertIsNone(helper.detect_kosis_error({"OK": True}))

    def test_xml_error_detected(self):
        body = '<?xml version="1.0"?><error><err>11</err><errMsg>유효하지않은 인증KEY입니다.</errMsg></error>'
        err = helper.detect_xml_error(body)
        self.assertIsNotNone(err)
        self.assertEqual(err.code, "11")
        self.assertIn("11", str(err))

    def test_xml_error_returns_none_for_normal_text(self):
        self.assertIsNone(helper.detect_xml_error("<sdmx:GenericData/>"))


class CallKosisTest(unittest.TestCase):
    def test_call_kosis_returns_payload_on_success(self):
        with mock.patch.object(helper, "fetch_text", return_value=read_fixture("search_response.json")):
            payload = helper.call_kosis("https://example", 5)
        self.assertIsInstance(payload, list)
        self.assertEqual(payload[0]["TBL_ID"], "DT_1JC1501")

    def test_call_kosis_raises_on_kosis_error_payload(self):
        body = json.dumps({"err": "31", "errMsg": "조회결과 초과"})
        with mock.patch.object(helper, "fetch_text", return_value=body):
            with self.assertRaises(helper.KosisError) as ctx:
                helper.call_kosis("https://example", 5)
        self.assertEqual(ctx.exception.code, "31")

    def test_call_kosis_returns_text_for_non_json_format(self):
        with mock.patch.object(helper, "fetch_text", return_value="<sdmx/>"):
            payload = helper.call_kosis("https://example", 5, format_hint="sdmx")
        self.assertEqual(payload, "<sdmx/>")

    def test_call_kosis_detects_json_error_envelope_in_non_json_format(self):
        body = json.dumps({"err": "11", "errMsg": "유효하지 않은 인증KEY입니다."})
        with mock.patch.object(helper, "fetch_text", return_value=body):
            with self.assertRaises(helper.KosisError) as ctx:
                helper.call_kosis("https://example", 5, format_hint="csv")
        self.assertEqual(ctx.exception.code, "11")

    def test_call_kosis_returns_csv_text_when_no_error_envelope(self):
        body = "PRD_DE,DT\n2024,1\n"
        with mock.patch.object(helper, "fetch_text", return_value=body):
            payload = helper.call_kosis("https://example", 5, format_hint="csv")
        self.assertEqual(payload, body)

    def test_bigdata_format_xls_is_rejected(self):
        with self.assertRaises(SystemExit):
            helper.parse_args([
                "bigdata", "--user-stats-id", "abc/def", "--format", "xls",
            ])


class RenderTextTest(unittest.TestCase):
    def test_search_text_lists_each_table(self):
        payload = json.loads(read_fixture("search_response.json"))
        text = helper.render_search_text(payload)
        self.assertIn("DT_1JC1501", text)
        self.assertIn("1인 가구 비율", text)

    def test_search_empty_renders_friendly_message(self):
        text = helper.render_search_text([])
        self.assertIn("결과가 없습니다", text)
        self.assertIn("키워드", text)
        self.assertIn("--start-count", text)

    def test_search_text_includes_next_step_hint(self):
        payload = json.loads(read_fixture("search_response.json"))
        text = helper.render_search_text(payload)
        self.assertIn("Next", text)
        self.assertIn("meta", text)
        self.assertIn("data", text)

    def test_meta_empty_suggests_other_meta_type(self):
        text = helper.render_meta_text([])
        self.assertIn("--meta-type", text)
        self.assertIn("TBL", text)

    def test_data_empty_suggests_filter_relaxation(self):
        text = helper.render_data_text([])
        self.assertIn("--obj-l", text)
        self.assertIn("meta", text)

    def test_data_text_includes_summary_with_period_and_unit(self):
        payload = json.loads(read_fixture("data_response.json"))
        text = helper.render_data_text(payload)
        self.assertIn("[summary]", text)
        self.assertIn("rows=2", text)
        self.assertIn("period=2023~2024", text)
        self.assertIn("unit=%", text)

    def test_data_text_summary_marks_missing_unit(self):
        text = helper.render_data_text([{"PRD_DE": "2024", "ITM_NM": "x", "DT": "1"}])
        self.assertIn("UNIT_NM 미포함", text)

    def test_meta_text_includes_korean_and_english(self):
        payload = json.loads(read_fixture("meta_response.json"))
        text = helper.render_meta_text(payload)
        self.assertIn("1인 가구 비율", text)
        self.assertIn("Single-person", text)

    def test_data_text_lists_period_and_value(self):
        payload = json.loads(read_fixture("data_response.json"))
        text = helper.render_data_text(payload)
        self.assertIn("2023", text)
        self.assertIn("35.5", text)
        self.assertIn("%", text)


class DryRunTest(unittest.TestCase):
    def test_dry_run_redacts_api_key_and_does_not_call_network(self):
        args = helper.parse_args(["search", "--query", "인구", "--dry-run", "--json"])
        with mock.patch.object(helper, "fetch_text") as fetch_mock:
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = helper.run(args)
        self.assertEqual(rc, 0)
        fetch_mock.assert_not_called()
        out = buf.getvalue()
        self.assertIn('"via_proxy": true', out)
        self.assertNotIn("apiKey", json.dumps(json.loads(out)["params"]))
        self.assertIn("/v1/kosis/search", out)
        self.assertIn("statisticsSearch.do", out)

    def test_direct_dry_run_redacts_api_key(self):
        args = helper.parse_args(["search", "--query", "인구", "--dry-run", "--direct", "--json"])
        with mock.patch.object(helper, "fetch_text") as fetch_mock:
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = helper.run(args)
        self.assertEqual(rc, 0)
        fetch_mock.assert_not_called()
        out = buf.getvalue()
        self.assertIn("<DRY-RUN>", out)
        self.assertIn('"via_proxy": false', out)
        self.assertIn("apiKey", out)


class RunIntegrationTest(unittest.TestCase):
    def test_run_search_text_renders_fixture_payload(self):
        args = helper.parse_args(["search", "--query", "1인 가구", "--text"])
        with mock.patch.object(helper, "fetch_text", return_value=read_fixture("search_response.json")) as fetch_mock:
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = helper.run(args)
        self.assertEqual(rc, 0)
        out = buf.getvalue()
        self.assertIn("DT_1JC1501", out)
        self.assertIn("statisticsSearch.do", out)
        self.assertIn("/v1/kosis/search", fetch_mock.call_args.args[0])
        self.assertNotIn("apiKey=", fetch_mock.call_args.args[0])

    def test_run_returns_2_on_kosis_error(self):
        args = helper.parse_args(["data", "--table-id", "DT_X",
                                  "--prd-se", "Y", "--start", "2020", "--end", "2024", "--json"])
        body = json.dumps({"err": "31", "errMsg": "조회결과 초과"})
        with mock.patch.object(helper, "resolve_api_key", return_value="KEY"), \
             mock.patch.object(helper, "fetch_text", return_value=body):
            rc = helper.run(args)
        self.assertEqual(rc, 2)


@unittest.skipUnless(os.getenv("KSKILL_KOSIS_API_KEY"), "live KOSIS test skipped without KSKILL_KOSIS_API_KEY")
class LiveKosisSmokeTest(unittest.TestCase):
    def test_live_search_returns_list(self):
        args = helper.parse_args(["search", "--query", "인구", "--result-count", "1", "--json", "--direct"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            rc = helper.run(args)
        self.assertEqual(rc, 0)
        payload = json.loads(buf.getvalue())
        self.assertIsInstance(payload, list)
        self.assertGreaterEqual(len(payload), 1)


if __name__ == "__main__":
    unittest.main()