kosis-stats

instruction.md

# KOSIS Stats

## What this skill does

국가데이터처(구 통계청)가 운영하는 KOSIS(국가통계포털) Open API `https://kosis.kr/openapi/` 로 한국 공식 통계 자료를 조회 자동화한다.

이 스킬은 **조회 전용**이다. 통계 작성, 데이터 변경, 대시보드 등록, 사용자별 통계 자료 등록은 범위에 포함하지 않는다.

지원 endpoint:

- `statisticsSearch.do` — 키워드로 통계표 검색
- `statisticsData.do?method=getMeta` — 통계표 메타데이터 (분류·항목·단위)
- `statisticsParameterData.do` — 통계표 데이터 셀 조회 (기간/분류 필터)
- `statisticsBigData.do` — 대용량 자료 (사전 등록한 `userStatsId` 필요)
- `statisticsList.do` — 통계목록 카테고리 트리 탐색 (주제별/기관별/국제통계/북한통계 등)
- `statisticsExplData.do` — 통계설명 (조사목적·주기·대상·법적근거·공표방법 등 27개 항목)
- `pkNumberService.do` — 통계주요지표 (1,473개 핵심 지표의 개념·산정방법·출처)

## When to use

- "1인 가구 비율 통계 찾아줘"
- "KOSIS에서 고령인구 비율 시도별 데이터 가져와"
- "DT_1IN0001 표 메타데이터 보여줘"
- "최근 5년치 소비자물가지수 KOSIS에서 뽑아줘"
- "국내통계 주제별로 어떤 카테고리가 있어?" (`list`)
- "인구총조사 조사목적·조사주기 알려줘" (`explain`)
- "인구밀도 지표 산정방법 보여줘" (`indicator`)

## When not to use

- 실시간 시세나 거래소 데이터를 원하는 경우 (KOSIS는 공식 통계용)
- 데이터 시각화·분석·보고서 작성이 주 목적인 경우 (이 스킬은 raw 데이터 조회만)
- 통계 작성·등록·수정이 필요한 경우
- 대용량 자료를 받기 위해 사용자별 자료(`userStatsId`)를 새로 등록해야 하는 경우 (KOSIS 웹에서 직접 등록)

## Prerequisites

- Python 3.9+ (stdlib only, 외부 패키지 없음)
- 일반 `search`/`meta`/`data`/`list`/`explain`/`indicator`: `k-skill-proxy`의 KOSIS route가 있는 hosted/self-host 프록시에 접근 가능할 것
- `bigdata` 또는 `--direct`: KOSIS Open API 인증키 (무료, https://kosis.kr/openapi/ 에서 회원가입 후 활용신청)

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- --help
```

### Bundled helper and reference

KOSIS helper와 OpenAPI guide reference는 `@nomadamas/k-skill` CLI 패키지에
함께 번들된다. raw `SKILL.md`만 설치된 환경에서도 별도로 다운로드하거나
복구하지 않는다. 조립된 instruction의 `k-skill exec`와 `k-skill read` 명령을
사용한다.

## Required environment variables

- 일반 `search`/`meta`/`data`/`list`/`explain`/`indicator`: 없음. 기본 hosted `https://k-skill-proxy.nomadamas.org` 를 사용한다.
- `KSKILL_PROXY_BASE_URL` — self-host·별도 프록시를 쓸 때만 설정. 비우면 기본 hosted proxy를 사용한다.
- `KSKILL_KOSIS_API_KEY` — `bigdata` 또는 `--direct`로 KOSIS를 직접 호출할 때만 필요하다.

발급 절차와 호출 한도, 에러 코드 등 자세한 내용은 `npx -y @nomadamas/k-skill@0 read kosis-stats references/kosis-openapi-guide.md` 참고.

### Credential resolution order (`bigdata` 또는 `--direct` 전용)

1. 돌쇠 credential mode에서는 provisioned `vault-run` capability를 사용하고, 없으면 `request_vault_credential`로 KOSIS API key 입력 UI를 호출한다.
2. 그 밖의 환경에서는 이미 주입된 환경변수 → host vault → `~/.config/k-skill/secrets.env` (`0600`) 순서로 사용한다.
3. generic fallback에서 값이 없으면 호스트의 가장 안전한 입력 표면으로 받아 vault 또는 dotenv에 저장한다.

일반 조회 helper는 proxy URL만 읽고, KOSIS 인증키는 proxy 서버에서만 주입한다. `bigdata`/`--direct` 호출만 `KSKILL_KOSIS_API_KEY` 환경변수와 위 secrets 파일을 읽는다.
`list`/`explain`/`indicator`는 `search`/`meta`/`data`와 마찬가지로 proxy 경유로 동작한다.

## Inputs

서브커맨드: `search`, `meta`, `data`, `bigdata`, `list`, `explain`, `indicator`.

공통 옵션:

- `--text`: 사람용 요약
- `--json`: 구조화 결과 (기본값)
- `--dry-run`: 인증키 없이 요청 URL/파라미터만 출력
- `--timeout N`: HTTP 타임아웃 초 단위 (기본 30)
- `--proxy-base-url URL`: 기본 hosted proxy 대신 self-host/alternate proxy 사용
- `--direct`: proxy를 우회하고 `KSKILL_KOSIS_API_KEY` 로 KOSIS 직접 호출

서브커맨드별 입력:

- `search`
  - `--query "키워드"`
  - `--result-count N` (1-5000, 기본 20)
  - `--start-count N` (페이징 시작, 기본 1)
- `meta`
  - `--org-id 101` (기본 101=통계청)
  - `--table-id DT_1IN0001`
  - `--meta-type TBL|ITM|OBJ` (기본 TBL)
- `data`
  - `--org-id 101`
  - `--table-id DT_1IN0001`
  - `--prd-se M|Q|S|Y|F|IR` (수록 주기)
  - `--start YYYY[MM|QQ|HH]`, `--end YYYY[MM|QQ|HH]`
  - `--itm-id ALL` (항목 ID, 기본 ALL)
  - `--obj-l 1=ALL --obj-l 2=00` (분류 필터, 반복 가능)
- `bigdata`
  - `--user-stats-id <KOSIS 등록 ID>`
  - `--format json|sdmx|csv` (xls는 바이너리라 helper 미지원 — 필요 시 KOSIS 웹에서 직접 다운로드)
  - `--prd-se`, `--new-est-prd-cnt` (선택)
- `list`
  - `--vw-cd MT_ZTITLE|MT_OTITLE|MT_GTITLE01|MT_GTITLE02|MT_CHOSUN_TITLE|MT_HANKUK_TITLE|MT_STOP_TITLE|MT_RTITLE|MT_BUKHAN|MT_TM1_TITLE|MT_TM2_TITLE|MT_ETITLE` (필수)
  - `--parent-id <LIST_ID>` (하위 카테고리 탐색, 기본 빈 문자열=최상위)
- `explain`
  - `--stat-id <통계조사ID>` (단독) 또는 `--org-id 101 --table-id DT_1IN0001` (조합, 둘 중 하나 필수)
  - `--meta-itm All|statsNm|statsKind|...` (기본 All, 한 번에 단일 필드 또는 All)
- `indicator`
  - `--jipyo-id 160` (필수, 지표ID)
  - `--service 1|2|3` (기본 1: 1=개념, 2=산정방법·출처, 3=전체설명)
  - `--page-no N`, `--num-of-rows N` (페이징)

## Workflow

### 1. Ensure proxy access is available

일반 `search`/`meta`/`data` 는 기본 hosted `k-skill-proxy`를 사용하므로 사용자 KOSIS 키가 필요 없다. self-host를 쓰면 `KSKILL_PROXY_BASE_URL`을 설정한다.

`bigdata` 또는 `--direct`가 필요할 때만 `KSKILL_KOSIS_API_KEY` 를 credential resolution order에 따라 확보한다. 시크릿이 없다는 이유로 다른 통계 사이트나 비공식 경로를 찾지 않는다.

### 2. Search for candidate tables

질문을 먼저 한국어 키워드로 좁히고 `search` 로 후보 통계표를 본다.

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- search --query "1인 가구" --text
```

출력에서 `[ORG_ID/TBL_ID]`를 골라 다음 단계에 사용한다.

### 3. Inspect the table meta before fetching data

데이터를 받기 전에 분류/단위를 확인한다.

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- meta --table-id DT_1JC1501 --text
```

**수록주기는 `meta` 로 확정되지 않는다.** `meta --meta-type TBL` 은 표 명칭(국/영문) 위주라 `--prd-se` 에 넣을 주기를 돌려주지 않고, `search` 응답의 `PRD_SE` 는 `--prd-se` 와 코드 체계가 달라(예: `DT_1IN0001` → `PRD_SE=A` 이지만 실제 조회는 `F`) 그대로 쓰면 안 된다. `STRT_PRD_DE`~`END_PRD_DE` 도 수록 범위일 뿐 간격을 알려주지 않는다. 조사주기는 `explain` 으로 확인하거나 §4 의 프로브로 확정한다.

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- explain \
  --org-id 101 --table-id DT_1IN0001 --meta-itm All --text
```

### 4. Fetch a small bounded slice first

`--prd-se`, `--start`, `--end`, `--itm-id`, `--obj-l` 으로 범위를 좁혀 작은 슬라이스를 먼저 조회한다. `DT_1JC1501` 같은 읍면동 단위 표는 기본값(`--itm-id ALL`, `--obj-l 1=ALL`)으로 조회하면 40,000셀 한도(code `31`)를 넘으므로 아래처럼 분류·항목까지 함께 좁힌다.

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- data \
  --table-id DT_1JC1501 --prd-se Y --start 2020 --end 2022 \
  --itm-id T1100 --obj-l 1=11 --json
```

프록시는 KOSIS 앱 레벨 에러를 HTTP 200 으로 흘려보내지 않고 매핑한다: `20`/`21`/`30`/`31` → `400`, `10`/`11`/`41`/`42` → `502 upstream_forbidden`, `40`(분당 호출 한도) → `503 upstream_quota_exceeded`, `50`/`99`/기타/빈 응답 → `502 upstream_error`.

**`--prd-se` 를 먼저 확정한다.** 주기를 틀리면 다른 파라미터가 전부 맞아도 KOSIS는 코드 `30`(결과 없음)만 돌려주며, 메시지로는 원인을 알 수 없다. `explain`(§3)으로 조사주기를 확인하거나, 좁은 기간·최소 분류로 `Y` → `F` → `IR` → `M`/`Q`/`S` 순으로 프로브해 데이터가 나오는 코드를 찾은 뒤 본 조회를 한다. 총조사·인구주택총조사처럼 5년 간격인 표는 연 단위로 보여도 `F`(다년)다.

```bash
for p in Y F IR; do
  npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- data \
    --org-id 101 --table-id DT_1IN0001 --prd-se "$p" \
    --start 2010 --end 2010 --itm-id ALL --obj-l 1=00 --obj-l 2=00 --text
done
```

표마다 필수 분류 차원 수가 다르다. **default `--obj-l 1=ALL` 만으로는 부족한 표가 많다.** KOSIS가 코드 `20` (필수요청변수값 누락 objL)을 돌려주면, `meta --table-id <ID> --meta-type ITM --json` 으로 ITM 안에 들어 있는 `OBJ_ID`(분류 차원)와 코드를 확인한 뒤 `--obj-l 1=<코드> --obj-l 2=<코드>` 형태로 필요한 차원을 모두 지정한다. (많은 표가 OBJ 메타는 비어 있고 분류가 ITM 안에 들어 있음.)

40,000셀을 초과하면 KOSIS는 에러 코드 `31` 또는 `41` 을 반환한다. 기간을 좁히거나(예: 5년→1년) 분류 필터의 ALL 을 특정 코드로 바꿔(예: `--obj-l 1=11` 서울만) 호출을 분할한다. 그래도 부족하면 사용자별 통계자료(`userStatsId`)를 등록해 `bigdata` 서브커맨드를 사용한다.

행정구역 코드 관례: `C1` 코드는 보통 시도가 2자리(`11` 서울, `26` 부산 등), 시군구가 5자리다. `data --json` 응답의 `C1` 필드를 확인해 원하는 단위만 후속 처리에서 필터한다.

### 5. (Optional) Use bigdata for large datasets

`bigdata` 는 KOSIS 웹에서 미리 등록한 `userStatsId` 가 필요하다. 미등록 상태면 사용자에게 등록 안내만 하고 멈춘다.

```bash
npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- bigdata \
  --user-stats-id "openapisample/101/DT_1IN1502/2/1/20191106094026_1" \
  --format json --new-est-prd-cnt 5
```

### 6. Cite the source

응답을 요약할 때는 `org_id`, `tbl_id`, 기간, 단위(`UNIT_NM`), 그리고 endpoint URL을 함께 적는다.

## Done when

- 사용자 질문에 대응하는 통계표 ID(`org_id`/`tbl_id`)가 명확하다.
- 메타데이터를 1회 이상 조회해 분류·단위를 확인했고, `--prd-se` 는 실제 데이터가 나오는 코드로 확정했다.
- 작은 슬라이스부터 단계적으로 데이터를 받았다.
- 결과에 출처(table id, 기간, 단위, endpoint)를 명시했다.
- 한도 초과 시 분할 또는 `bigdata` 안내로 처리했다.

## Failure modes

- `KSKILL_KOSIS_API_KEY` 누락: `bigdata` 또는 `--direct` 호출에서만 발급 안내 메시지와 함께 종료(exit 1)
- KOSIS 에러 코드 `10`/`11`: 인증키 누락/만료 → 키 점검. `bigdata` 에서 `11` 이 나오면 `userStatsId` 가 본인 KOSIS 계정에 등록된 것이 아닐 가능성이 크다.
- 코드 `20`: 필수 분류 누락 → `meta --meta-type OBJ` (또는 비어 있으면 `ITM`) 으로 필요한 차원 수와 코드를 확인하고 `--obj-l 1=... --obj-l 2=...` 모두 지정 후 재시도
- 코드 `21`: 잘못된 요청 변수 → `org_id`/`tbl_id`/기간 형식 재확인. tblId 의심 시 `search` 로 정확한 ID 다시 찾기
- 코드 `30`: 결과 없음. 어느 서브커맨드에서 났는지로 원인이 갈린다.
  - **`data` 에서 30 → 먼저 `--prd-se` 오지정을 의심한다.** 표의 실제 수록주기와 다른 코드를 주면 `--start`/`--end`/`--obj-l` 이 전부 맞아도 항상 30 이다. 연 단위처럼 보이는 표라도 총조사·센서스류는 다년주기 `F`(2·3·4·5·10년) 또는 부정기 `IR` 인 경우가 많다. `explain` 으로 조사주기를 확인하거나 `Y` → `F` → `IR` → `M`/`Q`/`S` 순으로 프로브해 확정한다 (§4 참고).
  - `search` 에서 30 → 키워드를 더 짧게 또는 다른 표현으로 바꾸거나 기간/분류 완화
  - **`meta` 에서 30** → 표가 해당 메타 타입을 지원하지 않는 경우이므로 다른 `--meta-type` 시도
- 코드 `31`/`41`: 한도 초과 → 기간 좁히기, 분류 ALL 을 특정 코드로 바꾸기, 또는 `bigdata` 사용
- 코드 `40`: 분당 1,000건 호출 한도 → 잠시 대기
- 코드 `50`: KOSIS 서버 오류 → 1~2초 후 재시도
- 비표준 JSON: KOSIS는 따옴표 없는 키를 가끔 반환한다. helper는 자동 보정한다.
- 응답에 `UNIT_NM` 누락: 일부 표는 KOSIS 응답에 단위가 비어 있다. helper text 출력의 `[summary]` 라인에 `unit=(KOSIS 응답에 UNIT_NM 미포함)` 으로 명시되며, 단위는 `meta` 응답이나 KOSIS 웹 화면에서 별도 확인한다.
- HTTPS 전용 (2026-03-05 이후): URL은 항상 `https://`. HTTP 요청은 차단된다.

### 회복 시나리오 예시

- 코드 20 회복: `data --table-id DT_1J22001 --prd-se M --start 202401 --end 202401` → 코드 20 → `meta --table-id DT_1J22001 --meta-type ITM --json` 으로 차원 확인 → `data ... --obj-l 1=T10 --obj-l 2=0` 재호출 → 성공
- 코드 31 회복: `data --table-id DT_1B26001 --prd-se Y --start 2020 --end 2024 --obj-l 1=ALL --obj-l 2=ALL --obj-l 3=ALL` → 코드 31 → `... --start 2024 --end 2024 --obj-l 1=11 --obj-l 2=ALL --obj-l 3=ALL` (서울만) 재호출 → 성공
- 코드 30 회복(수록주기 오지정): `data --org-id 101 --table-id DT_1IN0001 --prd-se Y --start 1990 --end 2010 --itm-id ALL --obj-l 1=00 --obj-l 2=00` → 코드 30 → 연도를 2015·2010·2005·2000·1995·1990 으로 바꿔 재시도해도 전부 코드 30, `--prd-se IR` 도 코드 30 → `--prd-se F --start 1925 --end 2010` 으로 재호출 → 성공(수록 시점 18개, 72셀). `DT_1IN0001`(총조사인구 총괄)은 연 단위 표처럼 보이지만 5년 간격 총조사라 `F` 로만 조회된다. 이 표는 `meta --meta-type TBL` 에도 주기가 없고 `search` 응답 `PRD_SE` 는 `A` 로 나와 둘 다 `--prd-se` 값으로 쓸 수 없었다.

## Maintainer review notes

메인테이너가 이 스킬을 검토하기 위해 KOSIS 인증키를 새로 발급받을 필요는 없다.
일반 조회는 `k-skill-proxy`가 KOSIS 인증키를 서버 쪽에서 주입한다. `bigdata` 와 `--direct`만 개인 KOSIS 키가 필요하다.

키 없이 가능한 검증:

- `./scripts/validate-skills.sh`
- `python3 -m py_compile kosis-stats/scripts/run_kosis_stats.py kosis-stats/tests/test_run_kosis_stats.py`
- `npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- --help`
- `npx -y @nomadamas/k-skill@0 exec kosis-stats scripts/run_kosis_stats.py -- search --query 인구 --dry-run` (URL/파라미터 출력만)
- `PYTHONPATH=kosis-stats/scripts python3 -m unittest discover -s kosis-stats/tests -p 'test_*.py' -v`
- `npm run ci`

실제 direct live smoke는 기여자 또는 이미 KOSIS 키가 있는 사용자가 선택적으로 수행한다. Proxy live smoke는 배포 proxy에 `KOSIS_API_KEY`가 설정된 뒤 수행한다. PR에는 호출 endpoint, 파라미터, 응답 행 수 같은 비민감 요약만 남기고 인증키와 개인 조회 세부 내역은 공유하지 않는다.

## Safety notes

- 조회 전용 스킬이다.
- 사용자별 통계자료(`userStatsId`) 등록, 데이터 수정, KOSIS 웹 자동화는 하지 않는다.
- 일반 조회 인증키는 proxy 서버에서만 다룬다. direct/bigdata 인증키는 환경변수 또는 `~/.config/k-skill/secrets.env` 로만 다룬다.
- 응답 JSON에 인증키가 echo 되지 않도록 helper는 `--dry-run` 시에도 키를 `<DRY-RUN>` 으로 대체한다.

references/kosis-openapi-guide.md

# KOSIS Open API 가이드

이 문서는 **국가데이터처**(구 통계청)가 운영하는 **KOSIS(국가통계포털, https://kosis.kr)** Open API의 인증키 발급 절차, 호출 한도, 주요 endpoint 사용법, 응답 포맷, 에러 코드를 한국어로 정리한 reference이다. 영문 명칭은 기존 KOSIS / Statistics Korea 그대로 사용된다.

출처:

- KOSIS Open API 공식 진입: https://kosis.kr/openapi/
  - 회원가입·활용신청·개발 가이드·내 신청 현황은 사이트 좌측 메뉴에서 진입한다.
  - deep-link(`devGuide/...`, `serviceUse/...`)는 SSO/SPA 라우팅에 따라 직접 접근 시 빈 화면이 뜰 수 있으니, 위 진입 URL에서 메뉴를 따라간다.
- KOSIS 공식 공지(2026-03-05 시행, HTTPS 전용·rate limit) — 활용신청 페이지 공지사항에서 확인.
- 본 가이드는 `kosis-mcp` 프로젝트의 `KOSIS_API_REFERENCE.md` 문서를 참고했다.

---

## 1. 인증키 발급 절차

KOSIS Open API는 무료다. 발급 단계:

1. **KOSIS 회원가입** — https://kosis.kr/ 우측 상단 "회원가입"
2. **활용신청** — https://kosis.kr/openapi/serviceUse/serviceUseUnityReg_01Detail.do
   - 활용 목적, 호출 빈도, 사용 서비스 종류를 입력한다
   - 신청 후 인증키가 즉시 발급된다 (관리자 승인 불필요)
3. **인증키 확인** — https://kosis.kr/openapi/serviceUse/myMain_01List.do
4. **`KSKILL_KOSIS_API_KEY` 환경변수에 저장**
5. (선택) 활용 사례 등록 — 작성한 어플리케이션 정보를 등록할 수 있다

발급된 키는 `apiKey=<KEY>` 형태로 모든 endpoint 호출에 포함한다.

---

## 2. 호출 한도와 프로토콜 (2026-03-05 적용)

| 항목 | 제한 | 비고 |
|------|------|------|
| Rate limit | 분당 1,000건 / 키 | 초과 시 에러 코드 `40` |
| 1회 호출당 최대 결과 | 40,000셀 | 초과 시 에러 코드 `31` 또는 `41` |
| 대용량 SDMX | 40,000셀 초과 시 SDMX 불가 → XLS 사용 | `statisticsBigData.do` |
| 대용량 XLS | 200,000셀 초과 시 XLS 불가 → 쿼리 분할 | |
| 프로토콜 | **HTTPS 전용** (HTTP 차단) | 모든 URL은 `https://` |

40,000셀이란 1회 응답에 포함되는 데이터 셀(수치 값) 수다. 예를 들어 100개 지역 × 10개 항목 × 50년 = 50,000셀 → 1회 호출 불가 → 기간/지역/항목으로 분할.

---

## 3. 주요 Endpoint

### 3.1 통계 검색 (`statisticsSearch.do`)

키워드로 통계표를 검색한다.

```
GET https://kosis.kr/openapi/statisticsSearch.do
  ?method=getList&apiKey={KEY}&format=json&jsonVD=Y
  &searchNm=인구&resultCount=20&startCount=1
```

응답 필드(주요): `ORG_ID`, `ORG_NM`, `TBL_ID`, `TBL_NM`, `STAT_ID`, `STAT_NM`, `VW_CD`, `MT_ATITLE`, `PRD_SE`, `STRT_PRD_DE`, `END_PRD_DE`, `LINK_URL`.

데이터 조회는 `ORG_ID` + `TBL_ID` 조합을 다음 단계에서 사용한다.

⚠️ **`search` 의 `PRD_SE` 를 §3.3 의 `prdSe` 요청 파라미터로 그대로 쓰지 말 것.** 두 필드는 코드 체계가 다르다 — `DT_1IN0001`(총조사인구 총괄)은 `search` 가 `PRD_SE=A` 를 돌려주지만 실제 데이터 조회는 `prdSe=F` 로만 성공하고 `A`·`Y`·`IR` 은 모두 코드 `30`(결과 없음)이다. `STRT_PRD_DE`~`END_PRD_DE` 도 수록 범위일 뿐 간격(주기)을 알려주지 않는다. 요청용 주기는 `statisticsExplData.do`(helper 의 `explain` 서브커맨드) 의 조사주기 항목에서 확인하거나, §3.3 표의 코드를 좁은 슬라이스로 프로브해 확정한다.

### 3.2 통계표 메타데이터 (`statisticsData.do?method=getMeta`)

통계표의 분류·항목·단위·국문/영문명 등을 조회한다.

```
GET https://kosis.kr/openapi/statisticsData.do
  ?method=getMeta&type=TBL&apiKey={KEY}&format=json
  &orgId=101&tblId=DT_1IN0001
```

`type` 값:

- `TBL` — 통계표 명칭(국/영문)
- `ITM` — 항목(item) 메타
- `OBJ` — 분류(classifier) 메타

### 3.3 통계 데이터 조회 (`statisticsParameterData.do`)

실제 통계 데이터 셀을 조회한다. 가장 많이 쓰는 endpoint다.

```
GET https://kosis.kr/openapi/Param/statisticsParameterData.do
  ?method=getList&apiKey={KEY}&format=json&jsonVD=Y
  &orgId=101&tblId=DT_1YL20631
  &objL1=ALL&itmId=ALL
  &prdSe=Y&startPrdDe=2020&endPrdDe=2024
```

수록주기(`prdSe`)와 기간 형식:

| 코드 | 설명 | 기간 형식 |
|------|------|----------|
| `M` | 월간/격월 | `YYYYMM` (202401) |
| `Q` | 분기 | `YYYYQQ` (202401 = 1분기) |
| `S` | 반기 | `YYYYHH` (202401 = 상반기) |
| `Y` | 연간 | `YYYY` (2024) |
| `F` | 다년(2,3,4,5,10년) | `YYYY` |
| `IR` | 부정기 | `YYYY` 또는 `YYYYMMDD` |

`prdSe` 를 틀리면 `startPrdDe`/`endPrdDe`/`objL*` 이 전부 맞아도 응답은 항상 코드 `30`(결과 없음)이고, 메시지만으로는 주기 문제인지 구분되지 않는다. 5년 간격 총조사처럼 연 단위로 보이는 표가 `F`(다년)인 경우가 흔하므로, `statisticsExplData.do` 로 조사주기를 확인하거나 최소 슬라이스로 `Y` → `F` → `IR` → `M`/`Q`/`S` 순으로 프로브해 확정한 뒤 본 조회를 한다.

분류 파라미터는 `objL1` ~ `objL8` (필요한 만큼만), 항목은 `itmId` (`ALL` 또는 특정 ID).

응답 셀 필드: `PRD_DE` (기간), `ITM_NM` (항목), `UNIT_NM` (단위), `DT` (값), `C1_NM`~`C8_NM` (분류명).

### 3.4 대용량 통계자료 (`statisticsBigData.do`)

40,000셀 초과 데이터를 한 번에 받기 위한 endpoint. **`userStatsId` 사전 등록 필요**.

```
GET https://kosis.kr/openapi/statisticsBigData.do
  ?method=getList&apiKey={KEY}&format=sdmx
  &userStatsId=openapisample/101/DT_1IN1502/2/1/20191106094026_1
  &prdSe=Y&newEstPrdCnt=5
```

`userStatsId` 발급:

1. KOSIS 로그인 → "개발 가이드 > 대용량 통계자료 > URL생성"
2. 통계표 ID, 항목, 분류, 기간을 선택해 자료 등록
3. 발급된 `userStatsId` 를 위 URL에 사용

응답 형식: `json`, `sdmx` (DSD/Generic/StructureSpecific), `csv`, `xls`.

> `run_kosis_stats.py bigdata --format` 은 텍스트 응답인 `json`, `sdmx`, `csv` 만 지원한다. `xls` 는 KOSIS가 바이너리 Excel 파일로 응답하므로 helper의 텍스트 출력 경로로 다루지 않는다. xls가 필요하면 KOSIS 웹 화면에서 직접 다운로드하거나, 추후 `--output PATH` 바이너리 모드가 추가되면 그때 사용한다.

---

## 4. 응답 포맷 주의

KOSIS API는 가끔 **비표준 JSON**을 반환한다. 키에 따옴표가 없는 경우가 있다.

```javascript
// 비표준 (KOSIS 원본)
{ORG_ID:"101", TBL_ID:"DT_1YL20631"}

// 표준 JSON으로 보정
{"ORG_ID":"101", "TBL_ID":"DT_1YL20631"}
```

`run_kosis_stats.py` 의 `fix_unquoted_keys()` 가 자동으로 보정한다.

`format=json&jsonVD=Y` 조합을 권장한다 (`jsonVD=Y` 는 verbatim 응답).

---

## 5. 에러 코드

KOSIS는 에러 시 `{"err": "<코드>", "errMsg": "<메시지>"}` 또는 `{"errCode": "<코드>", "errMsg": "<메시지>"}` 형태로 응답한다.

| 코드 | KOSIS 메시지 | 카테고리 | 권장 액션 |
|------|---|---|---|
| 10 | 인증키 누락 | auth | URL의 `?apiKey=` 확인 |
| 11 | 인증키 기간만료 | auth | https://kosis.kr/openapi/ 에서 갱신 |
| 20 | 필수요청변수 누락 | input | 필수 파라미터 확인 |
| 21 | 잘못된 요청변수 | input | `orgId`/`tblId`/기간 형식 재확인 |
| 30 | 조회결과 없음 | query | 키워드/기간/분류 완화 |
| 31 | 조회결과 초과 | query | 기간·지역·항목을 분할 |
| 40 | 호출가능건수 제한 | rate_limit | 분당 1,000건 한도 — 잠시 대기 |
| 41 | 호출가능ROW수 제한 | rate_limit | 1회 40,000셀 한도 — 쿼리 분할 |
| 42 | 사용자별 이용 제한 | rate_limit | KOSIS 운영팀 문의 |
| 50 | 서버오류 | server | 1~2초 대기 후 재시도 |

`run_kosis_stats.py` 는 위 코드를 감지해 사람용 힌트와 함께 stderr로 출력하고 exit 2 로 종료한다.

---

## 6. 안전 사용 가이드

- 인증키는 절대 저장소에 커밋하지 않는다. `examples/secrets.env.example` 참고.
- 호출은 분당 1,000건 한도 안에서 한다. 반복 폴링이 필요한 경우 호출 간 sleep을 둔다.
- 대용량 자료를 받을 때는 `userStatsId` 등록부터 사용자에게 안내한다. 자동 등록·웹 자동화는 하지 않는다.
- 응답에 개인식별정보가 포함될 일은 없지만, 비공개 검토 자료에 인증키가 echo되지 않도록 `--dry-run` 출력에서도 키는 `<DRY-RUN>` 으로 대체한다.

scripts/run_kosis_stats.py

#!/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"
LIST_URL = "https://kosis.kr/openapi/statisticsList.do"
EXPLAIN_URL = "https://kosis.kr/openapi/statisticsExplData.do"
INDICATOR_URL = "https://kosis.kr/openapi/pkNumberService.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"}

# statisticsList.do service view codes
VIEW_CODES = {
    "MT_ZTITLE", "MT_OTITLE", "MT_GTITLE01", "MT_GTITLE02",
    "MT_CHOSUN_TITLE", "MT_HANKUK_TITLE", "MT_STOP_TITLE",
    "MT_RTITLE", "MT_BUKHAN", "MT_TM1_TITLE", "MT_TM2_TITLE", "MT_ETITLE",
}

# pkNumberService.do sub-services
INDICATOR_SERVICES = {"1", "2", "3"}

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": "조회 결과가 없습니다. data 호출이면 먼저 `--prd-se` 오지정을 의심하세요 — 표의 실제 수록주기와 다르면 기간/분류가 맞아도 항상 이 에러가 납니다. 연 단위로 보이는 총조사·센서스류는 다년주기 `F` 또는 부정기 `IR` 인 경우가 많으니 `explain --org-id <ORG> --table-id <TBL>` 로 조사주기를 확인하거나 좁은 슬라이스로 `Y` → `F` → `IR` → `M`/`Q`/`S` 순으로 프로브하세요(`search` 응답의 `PRD_SE` 는 코드 체계가 달라 그대로 쓸 수 없습니다). search 호출이면 키워드를 더 짧게(예: '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 parse_view_code(value: str) -> str:
    upper = value.strip().upper()
    if upper not in VIEW_CODES:
        raise argparse.ArgumentTypeError(
            "must be one of: " + ", ".join(sorted(VIEW_CODES))
        )
    return upper


def parse_indicator_service(value: str) -> str:
    if value not in INDICATOR_SERVICES:
        raise argparse.ArgumentTypeError(
            "must be one of: " + ", ".join(sorted(INDICATOR_SERVICES))
        )
    return value


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).",
    )

    list_cmd = sub.add_parser(
        "list",
        help="Browse statistical table categories (statisticsList.do).",
    )
    _add_common_flags(list_cmd)
    list_cmd.add_argument(
        "--vw-cd",
        type=parse_view_code,
        required=True,
        help="Service view code: MT_ZTITLE(국내통계주제별) MT_OTITLE(기관별) "
        "MT_GTITLE01/02(e-지방지표) MT_RTITLE(국제통계) MT_BUKHAN(북한통계) etc.",
    )
    list_cmd.add_argument(
        "--parent-id",
        default="",
        help="Parent list ID for sub-categories (sent as parentListId). Empty string = top-level.",
    )

    explain = sub.add_parser(
        "explain",
        help="Fetch statistical survey description (statisticsExplData.do).",
    )
    _add_common_flags(explain)
    explain.add_argument(
        "--stat-id",
        help="Statistics survey ID (alternative to --org-id/--table-id).",
    )
    explain.add_argument("--org-id", help="Organization ID (with --table-id).")
    explain.add_argument("--table-id", help="KOSIS table ID (with --org-id).")
    explain.add_argument(
        "--meta-itm",
        default="All",
        choices=["All", *_EXPLAIN_FIELD_LABELS.keys()] if "_EXPLAIN_FIELD_LABELS" in globals() else None,
        help="Requested field: All(전체) or one field name (statsNm, statsKind, writingPurps, ...). Default All.",
    )

    indicator = sub.add_parser(
        "indicator",
        help="Fetch key indicator metadata (pkNumberService.do).",
    )
    _add_common_flags(indicator)
    indicator.add_argument(
        "--service",
        type=parse_indicator_service,
        default="1",
        help="Sub-service: 1(개념) 2(산정방법·출처) 3(전체설명). Default 1.",
    )
    indicator.add_argument(
        "--jipyo-id",
        required=True,
        help="Indicator ID (e.g. 160).",
    )
    indicator.add_argument(
        "--page-no",
        type=int,
        default=1,
        help="Page number (default 1).",
    )
    indicator.add_argument(
        "--num-of-rows",
        type=int,
        default=10,
        help="Rows per page (default 10).",
    )

    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_list_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    return {
        "method": "getList",
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "vwCd": args.vw_cd,
        "parentListId": args.parent_id,
    }


def build_explain_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    params: dict[str, str] = {
        "method": "getList",
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "metaItm": args.meta_itm,
    }
    if args.stat_id:
        params["statId"] = args.stat_id
    elif args.org_id and args.table_id:
        params["orgId"] = args.org_id
        params["tblId"] = args.table_id
    else:
        raise SystemExit(
            "explain requires either --stat-id or both --org-id and --table-id."
        )
    return params


def build_indicator_params(api_key: str, args: argparse.Namespace) -> dict[str, str]:
    return {
        "method": "getList",
        "apiKey": api_key,
        "format": "json",
        "jsonVD": "Y",
        "service": args.service,
        "serviceDetail": "pkNotion" if args.service == "1" else "pkCalcSource"
        if args.service == "2" else "pkAll",
        "jipyoId": args.jipyo_id,
        "pageNo": str(args.page_no),
        "numOfRows": str(args.num_of_rows),
    }


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 is_proxy_not_configured_body(body: str) -> bool:
    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        return False
    return isinstance(payload, dict) and payload.get("error") == "upstream_not_configured"


def fetch_text(url: str, timeout: int, *, via_proxy: bool = False) -> 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")
        if exc.code == 503 and is_proxy_not_configured_body(body):
            raise KosisError("503", "k-skill-proxy에 필요한 KOSIS API 키가 설정되어 있지 않습니다. 운영자에게 문의하세요.") from exc
        raise KosisError(str(exc.code), f"HTTP {exc.code}: {body[:200]}") from exc
    except urllib.error.URLError as exc:
        if not via_proxy:
            raise KosisError(None, f"network error: {exc.reason}") from exc
        raise KosisError(
            None,
            "설정된 k-skill-proxy 서버가 응답하지 않습니다. "
            f"잠시 후 재시도하거나 운영자에게 문의하세요. (상세: {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", via_proxy: bool = False) -> Any:
    text = fetch_text(url, timeout, via_proxy=via_proxy)
    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)


_EXPLAIN_FIELD_LABELS: dict[str, str] = {
    "statsNm": "조사명",
    "statsKind": "작성유형",
    "statsEnd": "통계종류",
    "statsContinue": "계속여부",
    "basisLaw": "법적근거",
    "writingPurps": "조사목적",
    "examinPd": "조사기간",
    "statsPeriod": "조사주기",
    "writingSystem": "조사체계",
    "writingTel": "연락처",
    "statsField": "통계분야·실태",
    "examinObjrange": "조사대상범위",
    "examinObjArea": "조사대상지역",
    "josaUnit": "조사단위·규모",
    "applyGroup": "적용분류",
    "josaItm": "조사항목",
    "pubPeriod": "공표주기",
    "pubExtent": "공표범위",
    "pubDate": "공표시기",
    "publictMth": "공표방법/URL",
    "examinTrgetPd": "조사대상기간/기준시점",
    "dataUserNote": "자료이용시 유의사항",
    "mainTermExpl": "주요 용어해설",
    "dataCollectMth": "자료 수집방법",
    "examinHistory": "조사연혁",
    "confmNo": "승인번호",
    "confmDt": "승인일자",
}


def render_list_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "조회 결과가 없습니다. --parent-id 를 비워 최상위 카테고리를 보거나, "
            "다른 --vw-cd (서비스뷰 코드)를 시도하세요."
        )
    lines: list[str] = []
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        list_id = entry.get("LIST_ID", "")
        list_nm = entry.get("LIST_NM", "?")
        org_id = entry.get("ORG_ID", "")
        tbl_id = entry.get("TBL_ID", "")
        tbl_nm = entry.get("TBL_NM", "")
        send_de = entry.get("SEND_DE", "")
        if tbl_id:
            lines.append(f"- [{list_id}] {list_nm} → [{org_id}/{tbl_id}] {tbl_nm} (갱신: {send_de})")
        else:
            lines.append(f"- [{list_id}] {list_nm}")
    lines.append(
        "\nNext: 하위 카테고리는 `list --vw-cd <코드> --parent-id <LIST_ID>`. "
        "통계표는 `meta --table-id <TBL_ID>` 또는 `data --table-id <TBL_ID> ...`."
    )
    return "\n".join(lines)


def render_explain_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "설명 정보가 없습니다. statId 또는 orgId+tblId 를 재확인하세요. "
            "`search --query <키워드>` 로 정확한 ID를 찾을 수 있습니다."
        )
    lines: list[str] = []
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        lines.append("---")
        for field, label in _EXPLAIN_FIELD_LABELS.items():
            value = entry.get(field, "")
            if value:
                lines.append(f"{label}: {value}")
    if len(lines) == 1 and lines[0] == "---":
        return "설명 정보 필드가 모두 비어 있습니다. 다른 --meta-itm 을 시도하세요."
    return "\n".join(lines)


def render_indicator_text(payload: Any) -> str:
    if not isinstance(payload, list) or not payload:
        return (
            "지표 정보가 없습니다. --jipyo-id 를 확인하세요. "
            "KOSIS 개발가이드에서 사용 가능한 지표ID 목록을 볼 수 있습니다."
        )
    lines: list[str] = []
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        jipyo_id = entry.get("statJipyoId", entry.get("jipyoId", "?"))
        jipyo_nm = entry.get("statJipyoNm", entry.get("jipyoNm", "?"))
        title = entry.get("jipyoExplan", "")
        details = [
            ("개념", entry.get("jipyoExplan1", entry.get("jipyoCalc", ""))),
            ("산정방법", entry.get("jipyoExplan2", "")),
            ("출처", entry.get("jipyoExplan3", "")),
        ]
        lines.append(f"- [{jipyo_id}] {jipyo_nm}")
        if title:
            lines.append(f"  제목: {title}")
        for label, value in details:
            if not value:
                continue
            display = value[:500] if len(value) > 500 else value
            lines.append(f"  {label}: {display}")
            if len(value) > 500:
                lines.append(f"  ... ({len(value) - 500}자 생략, --json으로 전체)")
    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)
    if command == "list":
        return render_list_text(payload)
    if command == "explain":
        return render_explain_text(payload)
    if command == "indicator":
        return render_indicator_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,
        "list": LIST_URL,
        "explain": EXPLAIN_URL,
        "indicator": INDICATOR_URL,
    }[command]


def should_use_proxy(args: argparse.Namespace) -> bool:
    return args.command in {"search", "meta", "data", "list", "explain", "indicator"} 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",
        "list": "/v1/kosis/list",
        "explain": "/v1/kosis/explain",
        "indicator": "/v1/kosis/indicator",
    }[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,
        "list": build_list_params,
        "explain": build_explain_params,
        "indicator": build_indicator_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, via_proxy=use_proxy)
    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())

skill.json

{
  "name": "kosis-stats",
  "description": "국가데이터처가 운영하는 KOSIS(국가통계포털, kosis.kr) Open API로 한국 공식 통계표를 검색하고 메타데이터·데이터·대용량 자료를 조회한다. Use when the user asks for 한국 공식 통계 (인구, 가구, 물가, 고용 등) 수치 조회, not for analysis or visualization.",
  "profiles": [
    "proxy",
    "lookup"
  ],
  "frontmatter": "name: kosis-stats\ndescription: 국가데이터처가 운영하는 KOSIS(국가통계포털, kosis.kr) Open API로 한국 공식 통계표를 검색하고 메타데이터·데이터·대용량 자료를 조회한다. Use when the user asks for 한국 공식 통계 (인구, 가구, 물가, 고용 등) 수치 조회, not for analysis or visualization.\nlicense: MIT\nmetadata:\n  category: data\n  locale: ko-KR\n  phase: v1"
}

SKILL.md

---
name: kosis-stats
description: 국가데이터처가 운영하는 KOSIS(국가통계포털, kosis.kr) Open API로 한국 공식 통계표를 검색하고 메타데이터·데이터·대용량 자료를 조회한다. Use when the user asks for 한국 공식 통계 (인구, 가구, 물가, 고용 등) 수치 조회, not for analysis or visualization.
license: MIT
metadata:
  category: data
  locale: ko-KR
  phase: v1
---

# kosis-stats

<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->

## Get the full instructions (required first step)

Run this and follow its output as the primary instructions for this skill:

```bash
npx -y @nomadamas/k-skill@0 instruct kosis-stats
```

The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:

```bash
npx -y @nomadamas/k-skill@0 files kosis-stats
```

Keep the CLI and every coding-agent skill install current (including Vercel Agent Skills) with:

```bash
npx -y @nomadamas/k-skill@0 update
```

If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/kosis-stats/instruction.md.

## Hard rules even without the CLI

- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.

tests/__init__.py

tests/fixtures/data_response.json

[
  {
    "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인 가구 비율"
  }
]

tests/fixtures/meta_response.json

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

tests/fixtures/search_response.json

[
  {
    "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인 가구"
  }
]

tests/test_run_kosis_stats.py

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",
            ])

    def test_list_subcommand_parses_vw_cd(self):
        args = helper.parse_args(["list", "--vw-cd", "MT_ZTITLE"])
        self.assertEqual(args.command, "list")
        self.assertEqual(args.vw_cd, "MT_ZTITLE")
        self.assertEqual(args.parent_id, "")

    def test_list_invalid_vw_cd_rejected(self):
        with self.assertRaises(SystemExit):
            helper.parse_args(["list", "--vw-cd", "INVALID_CODE"])

    def test_explain_requires_stat_id_or_org_tbl(self):
        args = helper.parse_args(["explain", "--stat-id", "A10120100413104843"])
        self.assertEqual(args.stat_id, "A10120100413104843")
        args2 = helper.parse_args(["explain", "--org-id", "101", "--table-id", "DT_1IN0001"])
        self.assertEqual(args2.org_id, "101")
        self.assertEqual(args2.table_id, "DT_1IN0001")

    def test_indicator_defaults_service_to_1(self):
        args = helper.parse_args(["indicator", "--jipyo-id", "160"])
        self.assertEqual(args.service, "1")
        self.assertEqual(args.jipyo_id, "160")
        self.assertEqual(args.page_no, 1)
        self.assertEqual(args.num_of_rows, 10)

    def test_indicator_invalid_service_rejected(self):
        with self.assertRaises(SystemExit):
            helper.parse_args(["indicator", "--jipyo-id", "160", "--service", "9"])


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)

    def test_list_params_include_vw_cd_and_parent_id(self):
        args = helper.parse_args(["list", "--vw-cd", "MT_ZTITLE", "--parent-id", "sub123"])
        params = helper.build_list_params("KEY", args)
        self.assertEqual(params["vwCd"], "MT_ZTITLE")
        self.assertEqual(params["parentListId"], "sub123")
        self.assertEqual(params["method"], "getList")

    def test_explain_params_with_org_and_table(self):
        args = helper.parse_args(["explain", "--org-id", "101", "--table-id", "DT_1IN0001"])
        params = helper.build_explain_params("KEY", args)
        self.assertEqual(params["orgId"], "101")
        self.assertEqual(params["tblId"], "DT_1IN0001")
        self.assertEqual(params["metaItm"], "All")

    def test_explain_params_with_stat_id(self):
        args = helper.parse_args(["explain", "--stat-id", "A10120100413104843"])
        params = helper.build_explain_params("KEY", args)
        self.assertEqual(params["statId"], "A10120100413104843")
        self.assertNotIn("orgId", params)

    def test_explain_without_any_id_exits(self):
        args = helper.parse_args(["explain"])
        with self.assertRaises(SystemExit):
            helper.build_explain_params("KEY", args)

    def test_indicator_params_include_service_and_jipyo_id(self):
        args = helper.parse_args(["indicator", "--jipyo-id", "160", "--service", "2"])
        params = helper.build_indicator_params("KEY", args)
        self.assertEqual(params["jipyoId"], "160")
        self.assertEqual(params["service"], "2")
        self.assertEqual(params["serviceDetail"], "pkCalcSource")
        self.assertEqual(params["pageNo"], "1")
        self.assertEqual(params["numOfRows"], "10")


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_proxy_not_configured_body_requires_matching_error_code(self):
        self.assertTrue(helper.is_proxy_not_configured_body(
            '{"error":"upstream_not_configured","message":"missing key"}'
        ))
        self.assertFalse(helper.is_proxy_not_configured_body(
            '{"error":"maintenance","message":"try later"}'
        ))
        self.assertFalse(helper.is_proxy_not_configured_body("temporarily unavailable"))

    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_code_30_hint_includes_prd_se_misconfiguration(self):
        err = helper.detect_kosis_error({"err": "30", "errMsg": "결과 없음"})
        self.assertIsNotNone(err)
        self.assertIn("--prd-se", str(err))
        self.assertIn("PRD_SE", 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)

    def test_list_empty_suggests_parent_id_or_vw_cd(self):
        text = helper.render_list_text([])
        self.assertIn("--parent-id", text)
        self.assertIn("--vw-cd", text)

    def test_list_text_shows_category_entries(self):
        payload = [
            {"LIST_ID": "L1", "LIST_NM": "인구/가구", "ORG_ID": "", "TBL_ID": "", "TBL_NM": "", "SEND_DE": ""},
            {"LIST_ID": "L2", "LIST_NM": "경제/산업", "ORG_ID": "101", "TBL_ID": "DT_X", "TBL_NM": "GDP", "SEND_DE": "20240101"},
        ]
        text = helper.render_list_text(payload)
        self.assertIn("인구/가구", text)
        self.assertIn("GDP", text)
        self.assertIn("DT_X", text)

    def test_explain_empty_suggests_search(self):
        text = helper.render_explain_text([])
        self.assertIn("search", text)

    def test_explain_text_shows_fields(self):
        payload = [{"statsNm": "인구총조사", "writingPurps": "인구 파악", "statsPeriod": "5년"}]
        text = helper.render_explain_text(payload)
        self.assertIn("인구총조사", text)
        self.assertIn("인구 파악", text)
        self.assertIn("조사주기", text)

    def test_indicator_empty_suggests_jipyo_id(self):
        text = helper.render_indicator_text([])
        self.assertIn("--jipyo-id", text)

    def test_indicator_text_shows_concept(self):
        payload = [{"statJipyoId": "160", "statJipyoNm": "인구밀도", "jipyoExplan": "인구밀도 설명", "jipyoExplan1": "1km²당 인구수"}]
        text = helper.render_indicator_text(payload)
        self.assertIn("160", text)
        self.assertIn("인구밀도", text)
        self.assertIn("1km²당 인구수", text)

    def test_indicator_text_shows_calculation_and_source(self):
        payload = [{
            "statJipyoId": "160",
            "statJipyoNm": "인구밀도",
            "jipyoExplan2": "총인구를 면적으로 나눔",
            "jipyoExplan3": "KOSIS",
        }]
        text = helper.render_indicator_text(payload)
        self.assertIn("산정방법", text)
        self.assertIn("총인구를 면적으로 나눔", text)
        self.assertIn("출처", text)
        self.assertIn("KOSIS", 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)

    def test_list_dry_run_uses_proxy_and_no_api_key(self):
        args = helper.parse_args(["list", "--vw-cd", "MT_ZTITLE", "--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.assertIn("/v1/kosis/list", out)
        self.assertIn("statisticsList.do", out)

    def test_explain_dry_run_uses_proxy(self):
        args = helper.parse_args(["explain", "--org-id", "101", "--table-id", "DT_1IN0001", "--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.assertIn("/v1/kosis/explain", out)
        self.assertIn("statisticsExplData.do", out)

    def test_indicator_dry_run_uses_proxy(self):
        args = helper.parse_args(["indicator", "--jipyo-id", "160", "--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.assertIn("/v1/kosis/indicator", out)
        self.assertIn("pkNumberService.do", 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()