Route NTS business checks through the proxy

Add the NTS business registration skill and proxy endpoints so agents can verify business-number status and authenticity without exposing data.go.kr keys to users.\n\nConstraint: data.go.kr publicDataPk=15081808 requires a server-side API key, so the route belongs behind k-skill-proxy.\nRejected: caller-supplied service keys | would violate the proxy credential boundary and duplicate user setup.\nConfidence: high\nScope-risk: moderate\nDirective: Keep future NTS fields normalized at the proxy boundary and never accept client serviceKey overrides.\nTested: PYTHONPATH=.:scripts python3 -m unittest scripts.test_nts_business_registration; npm run test --workspace k-skill-proxy -- --test-name-pattern 'NTS business'; buildServer smoke inject; npm run ci\nNot-tested: live data.go.kr request, because this session has no production DATA_GO_KR_API_KEY authority.
This commit is contained in:
Jeffrey (Dongkyu) Kim 2026-05-14 21:40:26 +09:00
commit cd3366a9dc
13 changed files with 1125 additions and 5 deletions

View file

@ -0,0 +1,5 @@
---
"k-skill-proxy": patch
---
Add National Tax Service business registration status and authenticity proxy routes.

View file

@ -38,6 +38,7 @@ Claude Code, Codex, OpenCode, OpenClaw/ClawHub 등 각종 코딩 에이전트
| 한국 법령 검색 | `korean-law-search` | 한국 법령/조문/판례/유권해석 검색 | 불필요 | [한국 법령 검색 가이드](docs/features/korean-law-search.md) |
| 등기부등본 자동화 | `iros-registry-automation` | 인터넷등기소(IROS)에서 법인/부동산 등기부등본 장바구니, 수동 결제 후 열람·저장 흐름을 보조 | 필요(수동 로그인·결제/TouchEn) | [등기부등본 자동화 가이드](docs/features/iros-registry-automation.md) |
| 법인등기 신청 컨설팅 | `corporate-registration-consulting` | 일반 영리 주식회사 발기설립 기준으로 법인명·이사·주소 등 사용자 결정사항을 받아 표준 정관, 설립등기 첨부서류, 등록면허세·과밀억제권역 중과 체크, rhwp 기반 HWP 양식 순차 작성 흐름을 참고용으로 안내 | 불필요 | [법인등기 신청 컨설팅 가이드](docs/features/corporate-registration-consulting.md) |
| 사업자등록정보 확인 | `nts-business-registration` | 국세청 사업자등록번호 상태조회와 사업자등록정보 진위확인(공공데이터포털 API, 프록시 경유) | 불필요 | [사업자등록정보 확인 가이드](docs/features/nts-business-registration.md) |
| 한국 사업자 장부 자동화 | `korean-jangbu-for` | `kimlawtech/korean-jangbu-for` 기반 카드·은행·영수증·세금계산서 입력 → 표준 거래내역·계정과목·세무사 전달 CSV·경영 리포트 생성 thin wrapper | 선택사항(CODEF BYOK 자동 수집 시 필요) | [한국 사업자 장부 자동화 가이드](docs/features/korean-jangbu-for.md) |
| 한국 개인정보처리방침·이용약관 자동 생성 | `korean-privacy-terms` | Next.js 프로젝트에 개인정보보호법·약관규제법·전자상거래법 기반 개인정보처리방침/이용약관/쿠키 배너/동의 모달을 생성하는 `kimlawtech/korean-privacy-terms` (Apache-2.0) thin wrapper | 불필요 | [한국 개인정보처리방침·이용약관 자동 생성 가이드](docs/features/korean-privacy-terms.md) |
| 한국 부동산 실거래가 조회 | `real-estate-search` | 아파트/오피스텔/빌라/단독주택 실거래가·전월세·지역코드 조회 | 불필요 | [한국 부동산 실거래가 조회 가이드](docs/features/real-estate-search.md) |
@ -145,6 +146,7 @@ Claude Code, Codex, OpenCode, OpenClaw/ClawHub 등 각종 코딩 에이전트
- [한강 수위 정보 가이드](docs/features/han-river-water-level.md)
- [한국 법령 검색 가이드](docs/features/korean-law-search.md)
- [한국 개인정보처리방침·이용약관 자동 생성 가이드](docs/features/korean-privacy-terms.md)
- [사업자등록정보 확인 가이드](docs/features/nts-business-registration.md)
- [한국 사업자 장부 자동화 가이드](docs/features/korean-jangbu-for.md)
- [한국 부동산 실거래가 조회 가이드](docs/features/real-estate-search.md)
- [개별공시지가 조회 가이드](docs/features/gongsijiga-search.md)

View file

@ -0,0 +1,43 @@
# 국세청 사업자등록정보 진위확인 및 상태조회
`nts-business-registration` 스킬은 공공데이터포털의 **국세청_사업자등록정보 진위확인 및 상태조회 서비스**를 `k-skill-proxy` 경유로 호출한다.
## 제공 기능
- 사업자등록번호 상태조회: `POST /v1/nts-business/status`
- 사업자등록정보 진위확인: `POST /v1/nts-business/validate`
## 인증/시크릿
사용자 로컬 시크릿은 필요 없다. upstream `DATA_GO_KR_API_KEY`는 프록시 서버에만 둔다.
self-host 프록시를 쓰는 경우에만 `KSKILL_PROXY_BASE_URL`을 설정한다. 비우면 hosted proxy(`https://k-skill-proxy.nomadamas.org`)를 사용한다.
## 예시
```bash
python3 nts-business-registration/scripts/nts_business_registration.py status \
--b-no 123-45-67890
```
```bash
python3 nts-business-registration/scripts/nts_business_registration.py validate \
--business-json '{"b_no":"123-45-67890","start_dt":"2020-01-31","p_nm":"홍길동","b_nm":"테스트상사"}'
```
## 입력 제한
- 사업자등록번호는 숫자 10자리여야 한다. 하이픈은 자동 제거한다.
- 상태조회/진위확인은 한 번에 최대 100건까지 보낸다.
- 진위확인은 `b_no`, `start_dt`, `p_nm`이 필수다.
- 선택 필드: `p_nm2`, `b_nm`, `corp_no`, `b_sector`, `b_type`, `b_adr`
## 실패 모드
- `400 bad_request`: 입력 형식 오류 또는 필수 필드 누락
- `503 upstream_not_configured`: 프록시 서버에 `DATA_GO_KR_API_KEY` 없음
- upstream 인증/활용신청 오류: 공공데이터포털 키가 해당 서비스에 승인되지 않았거나 오류 상태
## 공식 출처
- 공공데이터포털: <https://www.data.go.kr/tcs/dss/selectApiDataDetailView.do?publicDataPk=15081808>

View file

@ -0,0 +1,116 @@
---
name: nts-business-registration
description: 국세청 사업자등록정보 진위확인 및 사업자등록 상태조회를 공공데이터포털 API(k-skill-proxy 경유)로 수행한다.
license: MIT
metadata:
category: business
locale: ko-KR
phase: v1
---
# 국세청 사업자등록정보 진위확인 및 상태조회
## What this skill does
공공데이터포털의 **국세청_사업자등록정보 진위확인 및 상태조회 서비스**를 `k-skill-proxy` 경유로 호출해 다음을 확인한다.
- `status`: 사업자등록번호 기준 상태조회 (`계속사업자`, `휴업자`, `폐업자`, 과세유형 등 upstream 응답 그대로 포함)
- `validate`: 사업자등록번호 + 개업일자 + 대표자명(및 선택 필드) 기준 진위확인
## When to use
- "이 사업자등록번호가 계속사업자인지 확인해줘"
- "사업자등록번호 상태조회해줘"
- "사업자등록번호, 개업일, 대표자명으로 진위확인해줘"
- 거래처 등록 전 공식 NTS/공공데이터포털 기준 확인이 필요할 때
## Prerequisites
- 인터넷 연결
- `python3`
- 설치된 skill payload 안에 `scripts/nts_business_registration.py` helper 포함
- hosted/self-host `k-skill-proxy``/v1/nts-business/status`, `/v1/nts-business/validate` route 접근 가능
## Credential requirements
- 사용자 측 필수 시크릿 없음.
- `KSKILL_PROXY_BASE_URL` — self-host·별도 프록시를 쓸 때만 설정. 비우면 기본 hosted `https://k-skill-proxy.nomadamas.org` 를 사용한다.
- `DATA_GO_KR_API_KEY` 는 프록시 운영 서버 환경에만 둔다. 공공데이터포털에서 `국세청_사업자등록정보 진위확인 및 상태조회 서비스` 활용신청이 되어 있어야 한다.
## Official surfaces
- 공공데이터포털 문서: `https://www.data.go.kr/tcs/dss/selectApiDataDetailView.do?publicDataPk=15081808`
- 상태조회 upstream: `POST https://api.odcloud.kr/api/nts-businessman/v1/status?serviceKey=...`
- 진위확인 upstream: `POST https://api.odcloud.kr/api/nts-businessman/v1/validate?serviceKey=...`
- 프록시 route: `POST /v1/nts-business/status`, `POST /v1/nts-business/validate`
## Inputs
### 상태조회
- `b_no`: 사업자등록번호 10자리. 하이픈은 허용되며 helper/proxy가 숫자만 남긴다.
- 한 요청은 최대 100개까지 보낸다.
### 진위확인
필수:
- `b_no`: 사업자등록번호 10자리
- `start_dt`: 개업일자 `YYYYMMDD` (하이픈/점 허용)
- `p_nm`: 대표자 성명
선택:
- `p_nm2`: 대표자 성명2
- `b_nm`: 상호
- `corp_no`: 법인등록번호
- `b_sector`: 주업태명
- `b_type`: 주종목명
- `b_adr`: 사업장주소
## Workflow
1. 사용자 입력에서 사업자등록번호는 숫자 10자리인지 확인한다.
2. 상태조회만 필요하면 `status`를 호출한다.
3. 진위확인은 최소 `b_no`, `start_dt`, `p_nm`이 있을 때만 호출한다.
4. 개인정보/거래처 정보는 필요한 필드만 보내고, 프록시 응답을 그대로 보존하되 핵심 상태/진위 결과를 짧게 요약한다.
5. upstream이 `upstream_not_configured`, 활용신청 미승인, 인증키 오류 등을 반환하면 설정/승인 문제로 안내한다.
## CLI examples
```bash
python3 scripts/nts_business_registration.py status \
--b-no 123-45-67890
```
```bash
python3 scripts/nts_business_registration.py validate \
--business-json '{"b_no":"123-45-67890","start_dt":"2020-01-31","p_nm":"홍길동","b_nm":"테스트상사"}'
```
## Direct proxy examples
```bash
curl -fsS -X POST "$KSKILL_PROXY_BASE_URL/v1/nts-business/status" \
-H 'content-type: application/json' \
-d '{"b_no":["123-45-67890"]}'
```
```bash
curl -fsS -X POST "$KSKILL_PROXY_BASE_URL/v1/nts-business/validate" \
-H 'content-type: application/json' \
-d '{"businesses":[{"b_no":"123-45-67890","start_dt":"20200131","p_nm":"홍길동"}]}'
```
## Failure modes
- `400 bad_request`: 사업자등록번호가 10자리가 아니거나 진위확인 필수 필드가 빠짐.
- `503 upstream_not_configured`: 프록시 서버에 `DATA_GO_KR_API_KEY`가 없음.
- upstream 인증/활용신청 오류: API 키가 해당 서비스에 승인되지 않았거나 만료/오류 상태.
- 빈 결과 또는 진위불일치: 공식 응답의 `valid`, `valid_msg`, `b_stt` 값을 그대로 근거로 설명한다.
## Done when
- 상태조회는 공식 응답의 `b_stt`, `b_stt_cd`, `tax_type` 등 핵심 필드를 확인했다.
- 진위확인은 `valid`, `valid_msg` 결과를 확인했다.
- API 키는 사용자에게 요구하지 않고 프록시 서버에만 둔다는 점을 지켰다.

View file

@ -0,0 +1,183 @@
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import re
import sys
import urllib.error
import urllib.request
from typing import Any
PROXY_BASE_URL_ENV_VAR = "KSKILL_PROXY_BASE_URL"
DEFAULT_PROXY_BASE_URL = "https://k-skill-proxy.nomadamas.org"
BATCH_LIMIT = 100
class ApiError(RuntimeError):
def __init__(self, message: str, *, status_code: int | None = None, url: str | None = None):
super().__init__(message)
self.status_code = status_code
self.url = url
def _text_or_none(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def resolve_proxy_base_url(explicit_base_url: str | None = None, env: dict[str, str] | None = None) -> str:
env = os.environ if env is None else env
candidate = _text_or_none(explicit_base_url or env.get(PROXY_BASE_URL_ENV_VAR))
if candidate and candidate.casefold() in {"off", "false", "0", "disable", "disabled", "none"}:
raise ValueError("KSKILL_PROXY_BASE_URL 가 비활성화되어 있습니다.")
if candidate and candidate != "replace-me":
return candidate.rstrip("/")
return DEFAULT_PROXY_BASE_URL
def normalize_business_number(value: Any) -> str:
raw = _text_or_none(value)
if not raw:
raise ValueError("사업자등록번호(b_no)를 입력하세요.")
normalized = re.sub(r"\D", "", raw)
if not re.fullmatch(r"\d{10}", normalized):
raise ValueError("사업자등록번호는 숫자 10자리여야 합니다.")
return normalized
def normalize_start_date(value: Any) -> str:
raw = _text_or_none(value)
if not raw:
raise ValueError("개업일자(start_dt)를 YYYYMMDD 형식으로 입력하세요.")
normalized = re.sub(r"\D", "", raw)
if not re.fullmatch(r"\d{8}", normalized):
raise ValueError("개업일자는 YYYYMMDD 형식이어야 합니다.")
try:
dt.date(int(normalized[:4]), int(normalized[4:6]), int(normalized[6:8]))
except ValueError as error:
raise ValueError("개업일자는 유효한 날짜여야 합니다.") from error
return normalized
def build_status_payload(business_numbers: list[Any]) -> dict[str, list[str]]:
numbers = [normalize_business_number(value) for value in business_numbers]
numbers = list(dict.fromkeys(numbers))
if not numbers:
raise ValueError("사업자등록번호를 1개 이상 입력하세요.")
if len(numbers) > BATCH_LIMIT:
raise ValueError("한 번에 조회할 수 있는 사업자등록번호는 100개까지입니다.")
return {"b_no": numbers}
def build_validate_business(**kwargs: Any) -> dict[str, str]:
p_nm = _text_or_none(kwargs.get("p_nm"))
if not p_nm:
raise ValueError("대표자 성명(p_nm)을 입력하세요.")
business = {
"b_no": normalize_business_number(kwargs.get("b_no")),
"start_dt": normalize_start_date(kwargs.get("start_dt")),
"p_nm": p_nm,
}
for key in ("p_nm2", "b_nm", "b_sector", "b_type", "b_adr"):
value = _text_or_none(kwargs.get(key))
if value:
business[key] = value
corp_no = _text_or_none(kwargs.get("corp_no"))
if corp_no:
business["corp_no"] = re.sub(r"\D", "", corp_no)
return business
def build_validate_payload(businesses: list[dict[str, Any]]) -> dict[str, list[dict[str, str]]]:
if not businesses:
raise ValueError("진위확인 대상 businesses를 1개 이상 입력하세요.")
if len(businesses) > BATCH_LIMIT:
raise ValueError("한 번에 진위확인할 수 있는 사업자는 100개까지입니다.")
return {"businesses": [build_validate_business(**business) for business in businesses]}
def read_json_response(request: urllib.request.Request) -> dict[str, Any]:
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
try:
payload = json.loads(body)
except json.JSONDecodeError:
payload = None
if isinstance(payload, dict) and payload.get("message"):
raise ApiError(str(payload["message"]), status_code=error.code, url=getattr(error, "url", None)) from error
raise ApiError(f"NTS business proxy request failed with HTTP {error.code}", status_code=error.code, url=getattr(error, "url", None)) from error
except urllib.error.URLError as error:
raise ApiError(f"NTS business proxy request failed: {error.reason}") from error
def _post_json(path: str, payload: dict[str, Any], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
request = urllib.request.Request(
f"{resolved_base_url}{path}",
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "k-skill-nts-business-registration/1.0",
},
method="POST",
)
return read_json(request)
def query_status(business_numbers: list[Any], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
return _post_json("/v1/nts-business/status", build_status_payload(business_numbers), base_url=base_url, read_json=read_json)
def validate_businesses(businesses: list[dict[str, Any]], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
return _post_json("/v1/nts-business/validate", build_validate_payload(businesses), base_url=base_url, read_json=read_json)
def _parse_business_json(value: str) -> dict[str, Any]:
payload = json.loads(value)
if not isinstance(payload, dict):
raise argparse.ArgumentTypeError("business JSON must be an object")
return payload
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="NTS business registration status/authenticity helper")
subparsers = parser.add_subparsers(dest="command", required=True)
status = subparsers.add_parser("status", help="사업자등록번호 상태조회")
status.add_argument("--b-no", action="append", required=True, help="사업자등록번호(10자리; 하이픈 허용). 여러 번 지정 가능")
status.add_argument("--proxy-base-url")
validate = subparsers.add_parser("validate", help="사업자등록정보 진위확인")
validate.add_argument("--business-json", action="append", type=_parse_business_json, required=True, help='예: {"b_no":"1234567890","start_dt":"20200101","p_nm":"홍길동"}')
validate.add_argument("--proxy-base-url")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "status":
print(json.dumps(query_status(args.b_no, base_url=args.proxy_base_url), ensure_ascii=False, indent=2))
return 0
if args.command == "validate":
print(json.dumps(validate_businesses(args.business_json, base_url=args.proxy_base_url), ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
return 1
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -10,9 +10,9 @@
"scripts": {
"build": "npm run build --workspaces --if-present",
"build:manus-bundle": "node scripts/build-manus-bundle.js",
"lint": "node --check scripts/skill-docs.test.js scripts/korean_character_count.js scripts/test_korean_character_count.js scripts/build-manus-bundle.js scripts/test_build_manus_bundle.js && python3 -m py_compile scripts/k_skill_cleaner.py scripts/test_k_skill_cleaner.py corporate-registration-consulting/scripts/fill_official_hwp.py k-skill-cleaner/scripts/k_skill_cleaner.py scripts/fine_dust.py scripts/test_fine_dust.py scripts/ktx_booking.py scripts/test_ktx_booking.py scripts/sillok_search.py scripts/test_sillok_search.py scripts/korean_spell_check.py scripts/test_korean_spell_check.py scripts/patent_search.py scripts/test_patent_search.py scripts/mfds_drug_safety.py scripts/test_mfds_drug_safety.py scripts/mfds_food_safety.py scripts/test_mfds_food_safety.py scripts/zipcode_search.py scripts/test_zipcode_search.py scripts/subway_lost_property.py scripts/test_subway_lost_property.py scripts/geeknews_search.py scripts/test_geeknews_search.py scripts/test_naver_blog_search.py scripts/test_korean_slang_writing.py scripts/kakaotalk_mac.py scripts/test_kakaotalk_mac.py scripts/test_coupang_partners_mcp_wrapper.py scripts/ticket_availability.py scripts/test_ticket_availability.py ticket-availability/scripts/ticket_availability.py coupang-product-search/scripts/coupang_partners_mcp.py kakaotalk-mac/scripts/kakaotalk_mac.py naver-blog-research/scripts/_naver_http.py naver-blog-research/scripts/naver_search.py naver-blog-research/scripts/naver_read.py naver-blog-research/scripts/naver_download_images.py korean-slang-writing/scripts/_slang_http.py korean-slang-writing/scripts/slang_search.py korean-slang-writing/scripts/slang_lookup.py korean-scholarship-search/scripts/scholarship_filter.py korean-scholarship-search/scripts/test_scholarship_filter.py korean-scholarship-search/scripts/university_search_plan.py danawa-price-search/scripts/danawa_search.py kosis-stats/scripts/run_kosis_stats.py kosis-stats/tests/test_run_kosis_stats.py intercity-bus-booking/scripts/intercity_bus_search.py daangn-used-goods-search/scripts/daangn_used_goods.py daangn-realty-search/scripts/daangn_realty.py daangn-jobs-search/scripts/daangn_jobs.py daangn-cars-search/scripts/daangn_cars.py && npm run lint --workspaces --if-present && ./scripts/validate-skills.sh",
"lint": "node --check scripts/skill-docs.test.js scripts/korean_character_count.js scripts/test_korean_character_count.js scripts/build-manus-bundle.js scripts/test_build_manus_bundle.js && python3 -m py_compile scripts/k_skill_cleaner.py scripts/test_k_skill_cleaner.py corporate-registration-consulting/scripts/fill_official_hwp.py k-skill-cleaner/scripts/k_skill_cleaner.py scripts/fine_dust.py scripts/test_fine_dust.py scripts/ktx_booking.py scripts/test_ktx_booking.py scripts/sillok_search.py scripts/test_sillok_search.py scripts/korean_spell_check.py scripts/test_korean_spell_check.py scripts/patent_search.py scripts/test_patent_search.py scripts/mfds_drug_safety.py scripts/test_mfds_drug_safety.py scripts/nts_business_registration.py scripts/test_nts_business_registration.py scripts/mfds_food_safety.py scripts/test_mfds_food_safety.py scripts/zipcode_search.py scripts/test_zipcode_search.py scripts/subway_lost_property.py scripts/test_subway_lost_property.py scripts/geeknews_search.py scripts/test_geeknews_search.py nts-business-registration/scripts/nts_business_registration.py scripts/test_naver_blog_search.py scripts/test_korean_slang_writing.py scripts/kakaotalk_mac.py scripts/test_kakaotalk_mac.py scripts/test_coupang_partners_mcp_wrapper.py scripts/ticket_availability.py scripts/test_ticket_availability.py ticket-availability/scripts/ticket_availability.py coupang-product-search/scripts/coupang_partners_mcp.py kakaotalk-mac/scripts/kakaotalk_mac.py naver-blog-research/scripts/_naver_http.py naver-blog-research/scripts/naver_search.py naver-blog-research/scripts/naver_read.py naver-blog-research/scripts/naver_download_images.py korean-slang-writing/scripts/_slang_http.py korean-slang-writing/scripts/slang_search.py korean-slang-writing/scripts/slang_lookup.py korean-scholarship-search/scripts/scholarship_filter.py korean-scholarship-search/scripts/test_scholarship_filter.py korean-scholarship-search/scripts/university_search_plan.py danawa-price-search/scripts/danawa_search.py kosis-stats/scripts/run_kosis_stats.py kosis-stats/tests/test_run_kosis_stats.py intercity-bus-booking/scripts/intercity_bus_search.py daangn-used-goods-search/scripts/daangn_used_goods.py daangn-realty-search/scripts/daangn_realty.py daangn-jobs-search/scripts/daangn_jobs.py daangn-cars-search/scripts/daangn_cars.py && npm run lint --workspaces --if-present && ./scripts/validate-skills.sh",
"typecheck": "tsc --noEmit",
"test": "node --test scripts/skill-docs.test.js scripts/test_korean_character_count.js scripts/test_build_manus_bundle.js && PYTHONPATH=.:scripts python3 -m unittest scripts.test_k_skill_cleaner scripts.test_fine_dust scripts.test_ktx_booking scripts.test_sillok_search scripts.test_korean_spell_check scripts.test_patent_search scripts.test_mfds_drug_safety scripts.test_mfds_food_safety scripts.test_zipcode_search scripts.test_subway_lost_property scripts.test_geeknews_search scripts.test_naver_blog_search scripts.test_korean_slang_writing scripts.test_kakaotalk_mac scripts.test_coupang_partners_mcp_wrapper scripts.test_ticket_availability && PYTHONPATH=.:scripts:korean-scholarship-search/scripts python3 -m unittest discover -s korean-scholarship-search/scripts -p 'test_scholarship_filter.py' && PYTHONPATH=.:scripts:kosis-stats/scripts python3 -m unittest discover -s kosis-stats/tests -p 'test_run_kosis_stats.py' && npm run test --workspaces --if-present && ./scripts/validate-skills.sh",
"test": "node --test scripts/skill-docs.test.js scripts/test_korean_character_count.js scripts/test_build_manus_bundle.js && PYTHONPATH=.:scripts python3 -m unittest scripts.test_k_skill_cleaner scripts.test_fine_dust scripts.test_ktx_booking scripts.test_sillok_search scripts.test_korean_spell_check scripts.test_patent_search scripts.test_mfds_drug_safety scripts.test_nts_business_registration scripts.test_mfds_food_safety scripts.test_zipcode_search scripts.test_subway_lost_property scripts.test_geeknews_search scripts.test_naver_blog_search scripts.test_korean_slang_writing scripts.test_kakaotalk_mac scripts.test_coupang_partners_mcp_wrapper scripts.test_ticket_availability && PYTHONPATH=.:scripts:korean-scholarship-search/scripts python3 -m unittest discover -s korean-scholarship-search/scripts -p 'test_scholarship_filter.py' && PYTHONPATH=.:scripts:kosis-stats/scripts python3 -m unittest discover -s kosis-stats/tests -p 'test_run_kosis_stats.py' && npm run test --workspaces --if-present && ./scripts/validate-skills.sh",
"pack:dry-run": "npm pack --workspace k-lotto --dry-run && npm pack --workspace daiso-product-search --dry-run && npm pack --workspace market-kurly-search --dry-run && npm pack --workspace blue-ribbon-nearby --dry-run && npm pack --workspace kakao-bar-nearby --dry-run && npm pack --workspace cheap-gas-nearby --dry-run && npm pack --workspace public-restroom-nearby --dry-run && npm pack --workspace parking-lot-search --dry-run && npm pack --workspace court-auction-notice-search --dry-run && npm pack --workspace donation-place-search --dry-run && npm pack --workspace gongsijiga-search --dry-run && npm pack --workspace kbl-results --dry-run && npm pack --workspace kleague-results --dry-run && npm pack --workspace lck-analytics --dry-run && npm pack --workspace toss-securities --dry-run && npm pack --workspace hipass-receipt --dry-run && npm pack --workspace used-car-price-search --dry-run && npm pack --workspace k-skill-rhwp --dry-run && npm pack --workspace korean-marathon-schedule --dry-run && npm pack --workspace gangnamunni-clinic-search --dry-run",
"ci": "npm run lint && npm run typecheck && npm run test && npm run pack:dry-run",
"version-packages": "changeset version",

View file

@ -13,6 +13,8 @@
- `GET /v1/parking-lots/search` — 전국주차장정보표준데이터 기반 근처 공영주차장 검색(`DATA_GO_KR_API_KEY`)
- `GET /v1/neis/school-search` — 나이스 학교기본정보(교육청명·학교명 검색)
- `GET /v1/neis/school-meal` — 나이스 급식식단정보(일자별 메뉴)
- `POST /v1/nts-business/status` — 국세청 사업자등록 상태조회(`DATA_GO_KR_API_KEY`)
- `POST /v1/nts-business/validate` — 국세청 사업자등록정보 진위확인(`DATA_GO_KR_API_KEY`)
- `GET /v1/mfds/drug-safety/lookup` — 식약처 의약품개요정보(e약은요) + 안전상비의약품 정보(`DATA_GO_KR_API_KEY`)
- `GET /v1/mfds/food-safety/search` — 식약처 부적합 식품 + 식품안전나라 회수 정보(`DATA_GO_KR_API_KEY`, 선택적 `FOODSAFETYKOREA_API_KEY`)
- `GET /v1/korean-stock/search`
@ -60,7 +62,7 @@
- `KSKILL_PROXY_CACHE_TTL_MS` — 기본 `300000`
- `KSKILL_PROXY_RATE_LIMIT_WINDOW_MS` — 기본 `60000`
- `KSKILL_PROXY_RATE_LIMIT_MAX` — 기본 `60`
- `DATA_GO_KR_API_KEY` - 공공데이터포털 에서 쓰이는 API 인증키 (`household-waste`, `parking-lots`, `real-estate`, `mfds-drug-safety`, `mfds-food-safety`, `lh-notice`). 각 서비스는 공공데이터포털에서 별도 "활용신청" 승인이 필요하다. 키를 발급받은 뒤에는 [LH 임대공고문 정보](https://www.data.go.kr/data/15058530/openapi.do) 페이지에서도 활용신청을 눌러 동일 키를 활성화해야 `lh-notice` 라우트가 성공한다. 미활성 상태에서는 upstream이 HTTP 403 Forbidden을 돌려주고 proxy는 `upstream_error`로 변환한다.
- `DATA_GO_KR_API_KEY` - 공공데이터포털 에서 쓰이는 API 인증키 (`household-waste`, `parking-lots`, `real-estate`, `nts-business`, `mfds-drug-safety`, `mfds-food-safety`, `lh-notice`). 각 서비스는 공공데이터포털에서 별도 "활용신청" 승인이 필요하다. 키를 발급받은 뒤에는 [LH 임대공고문 정보](https://www.data.go.kr/data/15058530/openapi.do) 페이지에서도 활용신청을 눌러 동일 키를 활성화해야 `lh-notice` 라우트가 성공한다. 미활성 상태에서는 upstream이 HTTP 403 Forbidden을 돌려주고 proxy는 `upstream_error`로 변환한다.
기본 정책은 **무료 API 공개 프록시 = 무인증** 이다. 대신 endpoint scope 를 좁게 유지하고, cache + rate limit 으로 남용을 늦춘다.
@ -72,6 +74,15 @@ node packages/k-skill-proxy/src/server.js
환경변수(`AIR_KOREA_OPEN_API_KEY` 등)가 이미 설정되어 있거나 `~/.config/k-skill/secrets.env`를 source한 상태에서 실행한다.
국세청 사업자등록 상태조회 예시:
```bash
curl -fsS -X POST 'http://127.0.0.1:4020/v1/nts-business/status' \
-H 'content-type: application/json' \
-d '{"b_no":["123-45-67890"]}'
```
서울 지하철 도착정보 예시:
```bash

View file

@ -9,7 +9,7 @@
"node": ">=18"
},
"scripts": {
"lint": "node --check src/airkorea.js && node --check src/bluer.js && node --check src/hrfco.js && node --check src/krx-stock.js && node --check src/lh-notice.js && node --check src/mfds.js && node --check src/molit.js && node --check src/naver-news.js && node --check src/naver-shopping.js && node --check src/parking-lots.js && node --check src/region-lookup.js && node --check src/server.js && node --check test/airkorea.test.js && node --check test/hrfco.test.js && node --check test/lh-notice.test.js && node --check test/molit.test.js && node --check test/naver-news.test.js && node --check test/naver-shopping.test.js && node --check test/region-lookup.test.js && node --check test/server.test.js",
"lint": "node --check src/airkorea.js && node --check src/bluer.js && node --check src/hrfco.js && node --check src/krx-stock.js && node --check src/lh-notice.js && node --check src/mfds.js && node --check src/molit.js && node --check src/naver-news.js && node --check src/naver-shopping.js && node --check src/nts-business.js && node --check src/parking-lots.js && node --check src/region-lookup.js && node --check src/server.js && node --check test/airkorea.test.js && node --check test/hrfco.test.js && node --check test/lh-notice.test.js && node --check test/molit.test.js && node --check test/naver-news.test.js && node --check test/naver-shopping.test.js && node --check test/region-lookup.test.js && node --check test/server.test.js",
"test": "node --test"
},
"dependencies": {

View file

@ -0,0 +1,178 @@
const NTS_BUSINESSMAN_UPSTREAM_BASE_URL = "https://api.odcloud.kr/api/nts-businessman/v1";
const NTS_BATCH_LIMIT = 100;
const NTS_BUSINESS_OPERATIONS = new Set(["status", "validate"]);
const NTS_VALIDATE_OPTIONAL_TEXT_FIELDS = ["p_nm2", "b_nm", "b_sector", "b_type", "b_adr"];
function trimOrNull(value) {
if (value === undefined || value === null) {
return null;
}
const trimmed = String(value).trim();
return trimmed ? trimmed : null;
}
function normalizeBusinessNumber(value) {
const raw = trimOrNull(value);
if (!raw) {
throw new Error("Provide business registration number (b_no). business registration number must be 10 digits.");
}
const normalized = raw.replace(/[^0-9]/g, "");
if (!/^\d{10}$/.test(normalized)) {
throw new Error("Provide valid business registration number (b_no) as 10 digits.");
}
return normalized;
}
function normalizeNtsBusinessNumbers(value) {
const rawValues = Array.isArray(value) ? value : String(value ?? "").split(",");
const numbers = rawValues
.flatMap((entry) => (Array.isArray(entry) ? entry : [entry]))
.map((entry) => trimOrNull(entry))
.filter(Boolean)
.map(normalizeBusinessNumber);
const unique = [...new Set(numbers)];
if (unique.length === 0) {
throw new Error("Provide b_no as one or more business registration numbers.");
}
if (unique.length > NTS_BATCH_LIMIT) {
throw new Error(`Provide up to ${NTS_BATCH_LIMIT} business registration numbers per request.`);
}
return unique;
}
function normalizeNtsStartDate(value) {
const raw = trimOrNull(value);
if (!raw) {
throw new Error("Provide start_dt as YYYYMMDD.");
}
const normalized = raw.replace(/[^0-9]/g, "");
if (!/^\d{8}$/.test(normalized)) {
throw new Error("Provide start_dt as YYYYMMDD.");
}
const year = Number.parseInt(normalized.slice(0, 4), 10);
const month = Number.parseInt(normalized.slice(4, 6), 10);
const day = Number.parseInt(normalized.slice(6, 8), 10);
const date = new Date(Date.UTC(year, month - 1, day));
if (date.getUTCFullYear() !== year || date.getUTCMonth() !== month - 1 || date.getUTCDate() !== day) {
throw new Error("Provide start_dt as a valid YYYYMMDD date.");
}
return normalized;
}
function normalizeOptionalDigits(value, label) {
const raw = trimOrNull(value);
if (!raw) {
return null;
}
const normalized = raw.replace(/[^0-9]/g, "");
if (!normalized) {
throw new Error(`Provide valid ${label}.`);
}
return normalized;
}
function normalizeNtsBusinessStatusQuery(body = {}) {
return {
b_no: normalizeNtsBusinessNumbers(body.b_no ?? body.business_numbers ?? body.businessNumbers)
};
}
function normalizeNtsBusinessValidateItem(item) {
if (!item || typeof item !== "object" || Array.isArray(item)) {
throw new Error("Each business must be an object.");
}
const pNm = trimOrNull(item.p_nm ?? item.owner_name ?? item.ownerName ?? item.representative_name);
if (!pNm) {
throw new Error("Provide p_nm (representative name) for each business.");
}
const normalized = {
b_no: normalizeBusinessNumber(item.b_no ?? item.business_number ?? item.businessNumber),
start_dt: normalizeNtsStartDate(item.start_dt ?? item.startDate ?? item.opening_date),
p_nm: pNm
};
for (const key of NTS_VALIDATE_OPTIONAL_TEXT_FIELDS) {
const value = trimOrNull(item[key]);
if (value) {
normalized[key] = value;
}
}
const corpNo = normalizeOptionalDigits(item.corp_no ?? item.corpNo, "corp_no");
if (corpNo) {
normalized.corp_no = corpNo;
}
return normalized;
}
function normalizeNtsBusinessValidateQuery(body = {}) {
const businesses = body.businesses;
if (!Array.isArray(businesses) || businesses.length === 0) {
throw new Error("Provide businesses as a non-empty array.");
}
if (businesses.length > NTS_BATCH_LIMIT) {
throw new Error(`Provide up to ${NTS_BATCH_LIMIT} businesses per request.`);
}
return {
businesses: businesses.map(normalizeNtsBusinessValidateItem)
};
}
async function proxyNtsBusinessRequest({ operation, payload, serviceKey, fetchImpl = global.fetch }) {
if (!serviceKey) {
return {
statusCode: 503,
contentType: "application/json; charset=utf-8",
body: JSON.stringify({
error: "upstream_not_configured",
message: "DATA_GO_KR_API_KEY is not configured on the proxy server."
})
};
}
if (!NTS_BUSINESS_OPERATIONS.has(operation)) {
return {
statusCode: 404,
contentType: "application/json; charset=utf-8",
body: JSON.stringify({
error: "not_found",
message: "That NTS business route is not exposed by this proxy."
})
};
}
const url = new URL(`${NTS_BUSINESSMAN_UPSTREAM_BASE_URL}/${operation}`);
url.searchParams.set("serviceKey", serviceKey);
const response = await fetchImpl(url, {
method: "POST",
headers: {
accept: "application/json",
"content-type": "application/json"
},
body: JSON.stringify(payload),
signal: AbortSignal.timeout(20000)
});
return {
statusCode: response.status,
contentType: response.headers.get("content-type") || "application/json; charset=utf-8",
body: await response.text()
};
}
module.exports = {
normalizeBusinessNumber,
normalizeNtsBusinessNumbers,
normalizeNtsBusinessStatusQuery,
normalizeNtsBusinessValidateItem,
normalizeNtsBusinessValidateQuery,
normalizeNtsStartDate,
proxyNtsBusinessRequest
};

View file

@ -22,6 +22,11 @@ const {
const { fetchTransactions, VALID_ASSET_TYPES, VALID_DEAL_TYPES } = require("./molit");
const { fetchNaverNewsSearch, normalizeNaverNewsSearchQuery } = require("./naver-news");
const { fetchNaverShoppingSearch, normalizeNaverShoppingSearchQuery } = require("./naver-shopping");
const {
normalizeNtsBusinessStatusQuery,
normalizeNtsBusinessValidateQuery,
proxyNtsBusinessRequest
} = require("./nts-business");
const { fetchNearbyParkingLots } = require("./parking-lots");
const { searchRegionCode } = require("./region-lookup");
const { resolveEducationOfficeFromNaturalLanguage } = require("./neis-office-codes");
@ -1560,7 +1565,8 @@ function buildServer({ env = process.env, provider = null, now = () => new Date(
kosisConfigured: Boolean(config.kosisApiKey),
naverShoppingConfigured: true,
naverSearchApiConfigured: naverSearchKeysPresent,
naverNewsApiConfigured: naverSearchKeysPresent
naverNewsApiConfigured: naverSearchKeysPresent,
ntsBusinessConfigured: Boolean(config.molitApiKey)
},
auth: {
tokenRequired: false
@ -2635,6 +2641,127 @@ function buildServer({ env = process.env, provider = null, now = () => new Date(
return payload;
});
async function handleNtsBusinessRoute({ operation, route, normalizer, request, reply }) {
let normalized;
try {
normalized = normalizer(request.body || {});
} catch (error) {
reply.code(400);
return {
error: "bad_request",
message: error.message
};
}
const cacheKey = makeCacheKey({
route,
...normalized
});
const cached = cache.get(cacheKey);
if (cached) {
return {
...cached,
proxy: {
...cached.proxy,
cache: {
hit: true,
ttl_ms: config.cacheTtlMs
}
}
};
}
let upstream;
try {
upstream = await proxyNtsBusinessRequest({
operation,
payload: normalized,
serviceKey: config.molitApiKey
});
} catch (error) {
reply.code(502);
return {
error: "proxy_error",
message: error.message,
proxy: {
name: config.proxyName,
cache: {
hit: false,
ttl_ms: config.cacheTtlMs
}
}
};
}
let parsed;
try {
parsed = JSON.parse(upstream.body);
} catch {
reply.code(upstream.statusCode >= 400 ? upstream.statusCode : 502);
return {
error: "upstream_invalid_response",
message: "NTS business upstream did not return valid JSON.",
upstream_status: upstream.statusCode,
proxy: {
name: config.proxyName,
cache: {
hit: false,
ttl_ms: config.cacheTtlMs
}
}
};
}
if (upstream.statusCode < 200 || upstream.statusCode >= 300 || parsed.error) {
reply.code(upstream.statusCode >= 400 ? upstream.statusCode : 502);
return {
...parsed,
proxy: {
name: config.proxyName,
cache: {
hit: false,
ttl_ms: config.cacheTtlMs
},
requested_at: new Date().toISOString()
}
};
}
const payload = {
...parsed,
query: normalized,
proxy: {
name: config.proxyName,
cache: {
hit: false,
ttl_ms: config.cacheTtlMs
},
requested_at: new Date().toISOString()
}
};
cache.set(cacheKey, payload, config.cacheTtlMs);
return payload;
}
app.post("/v1/nts-business/status", async (request, reply) => handleNtsBusinessRoute({
operation: "status",
route: "nts-business-status",
normalizer: normalizeNtsBusinessStatusQuery,
request,
reply
}));
app.post("/v1/nts-business/validate", async (request, reply) => handleNtsBusinessRoute({
operation: "validate",
route: "nts-business-validate",
normalizer: normalizeNtsBusinessValidateQuery,
request,
reply
}));
app.get("/v1/mfds/drug-safety/lookup", async (request, reply) => {
let normalized;
@ -3850,6 +3977,8 @@ module.exports = {
normalizeNeisSchoolMealQuery,
normalizeNeisSchoolSearchQuery,
normalizeNaverShoppingSearchQuery,
normalizeNtsBusinessStatusQuery,
normalizeNtsBusinessValidateQuery,
normalizeParkingLotSearchQuery,
normalizeRealEstateQuery,
normalizeRegionCodeQuery,

View file

@ -15,6 +15,8 @@ const {
normalizeKosisDataQuery,
normalizeKosisMetaQuery,
normalizeKosisSearchQuery,
normalizeNtsBusinessStatusQuery,
normalizeNtsBusinessValidateQuery,
proxyAirKoreaRequest,
proxyData4LibraryRequest,
proxyHrfcoWaterLevelRequest,
@ -151,6 +153,176 @@ test("food-safety search does not cache upstream failures so transient errors se
assert.equal(recallCalls.length, 2, "upstream hit on first (fail) and second (recovered) - third served from cache");
});
test("NTS business normalizers validate status and authenticity payloads", () => {
const tooManyBusinessNumbers = Array.from({ length: 101 }, (_, index) => String(index).padStart(10, "0"));
assert.deepEqual(normalizeNtsBusinessStatusQuery({ b_no: "123-45-67890, 9876543210" }), {
b_no: ["1234567890", "9876543210"]
});
assert.deepEqual(
normalizeNtsBusinessValidateQuery({
businesses: [
{
b_no: "123-45-67890",
start_dt: "2020-01-31",
p_nm: "홍길동",
b_nm: "테스트상사",
corp_no: "110111-1234567"
}
]
}),
{
businesses: [
{
b_no: "1234567890",
start_dt: "20200131",
p_nm: "홍길동",
b_nm: "테스트상사",
corp_no: "1101111234567"
}
]
}
);
assert.throws(() => normalizeNtsBusinessStatusQuery({ b_no: "123" }), /business registration number/);
assert.throws(
() => normalizeNtsBusinessValidateQuery({ businesses: [{ b_no: "1234567890", p_nm: "홍길동" }] }),
/start_dt/
);
assert.throws(
() => normalizeNtsBusinessStatusQuery({ b_no: tooManyBusinessNumbers }),
/up to 100/
);
});
test("NTS business status route proxies POST body with service key server-side", async (t) => {
const originalFetch = global.fetch;
const calls = [];
global.fetch = async (url, options) => {
calls.push({ url: String(url), options });
return new Response(
JSON.stringify({ status_code: "OK", request_cnt: 1, data: [{ b_no: "1234567890", b_stt: "계속사업자" }] }),
{ status: 200, headers: { "content-type": "application/json;charset=UTF-8" } }
);
};
const app = buildServer({ env: { DATA_GO_KR_API_KEY: "data-go-key" } });
t.after(async () => {
global.fetch = originalFetch;
await app.close();
});
const response = await app.inject({
method: "POST",
url: "/v1/nts-business/status",
payload: { b_no: ["123-45-67890"] }
});
const body = response.json();
assert.equal(response.statusCode, 200);
assert.equal(body.data[0].b_stt, "계속사업자");
assert.equal(body.proxy.cache.hit, false);
assert.match(calls[0].url, /\/nts-businessman\/v1\/status\?serviceKey=data-go-key$/);
assert.deepEqual(JSON.parse(calls[0].options.body), { b_no: ["1234567890"] });
assert.equal(calls[0].options.method, "POST");
assert.equal(calls[0].options.headers["content-type"], "application/json");
const cached = await app.inject({
method: "POST",
url: "/v1/nts-business/status",
payload: { b_no: ["1234567890"] }
});
const cachedBody = cached.json();
assert.equal(cached.statusCode, 200);
assert.equal(cachedBody.proxy.cache.hit, true);
assert.equal(calls.length, 1);
});
test("NTS business validate route normalizes businesses and reports missing key", async (t) => {
const missingKeyApp = buildServer();
t.after(async () => {
await missingKeyApp.close();
});
const unavailable = await missingKeyApp.inject({
method: "POST",
url: "/v1/nts-business/validate",
payload: { businesses: [{ b_no: "1234567890", start_dt: "20200101", p_nm: "홍길동" }] }
});
const unavailableBody = unavailable.json();
assert.equal(unavailable.statusCode, 503);
assert.equal(unavailableBody.error, "upstream_not_configured");
const originalFetch = global.fetch;
const calls = [];
global.fetch = async (url, options) => {
calls.push({ url: String(url), options });
return new Response(
JSON.stringify({ status_code: "OK", valid_cnt: 1, data: [{ b_no: "1234567890", valid: "01", valid_msg: "확인할 수 있습니다." }] }),
{ status: 200, headers: { "content-type": "application/json;charset=UTF-8" } }
);
};
const app = buildServer({ env: { DATA_GO_KR_API_KEY: "data-go-key" } });
t.after(async () => {
global.fetch = originalFetch;
await app.close();
});
const response = await app.inject({
method: "POST",
url: "/v1/nts-business/validate",
payload: { businesses: [{ b_no: "123-45-67890", start_dt: "2020.01.01", p_nm: "홍길동", p_nm2: "", b_adr: "서울" }] }
});
const body = response.json();
assert.equal(response.statusCode, 200);
assert.equal(body.data[0].valid, "01");
assert.match(calls[0].url, /\/nts-businessman\/v1\/validate\?serviceKey=data-go-key$/);
assert.deepEqual(JSON.parse(calls[0].options.body), {
businesses: [{ b_no: "1234567890", start_dt: "20200101", p_nm: "홍길동", b_adr: "서울" }]
});
});
test("NTS business route maps upstream fetch failures to 502 without caching", async (t) => {
const originalFetch = global.fetch;
let calls = 0;
global.fetch = async () => {
calls += 1;
throw new Error("network down");
};
const app = buildServer({ env: { DATA_GO_KR_API_KEY: "data-go-key" } });
t.after(async () => {
global.fetch = originalFetch;
await app.close();
});
const first = await app.inject({
method: "POST",
url: "/v1/nts-business/status",
payload: { b_no: ["1234567890"] }
});
const firstBody = first.json();
assert.equal(first.statusCode, 502);
assert.equal(firstBody.error, "proxy_error");
assert.match(firstBody.message, /network down/);
const second = await app.inject({
method: "POST",
url: "/v1/nts-business/status",
payload: { b_no: ["1234567890"] }
});
assert.equal(second.statusCode, 502);
assert.equal(calls, 2, "fetch failures must not be cached");
});
test("health endpoint stays public and reports auth/upstream status", async (t) => {
const app = buildServer({
provider: async () => {

View file

@ -0,0 +1,183 @@
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import re
import sys
import urllib.error
import urllib.request
from typing import Any
PROXY_BASE_URL_ENV_VAR = "KSKILL_PROXY_BASE_URL"
DEFAULT_PROXY_BASE_URL = "https://k-skill-proxy.nomadamas.org"
BATCH_LIMIT = 100
class ApiError(RuntimeError):
def __init__(self, message: str, *, status_code: int | None = None, url: str | None = None):
super().__init__(message)
self.status_code = status_code
self.url = url
def _text_or_none(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def resolve_proxy_base_url(explicit_base_url: str | None = None, env: dict[str, str] | None = None) -> str:
env = os.environ if env is None else env
candidate = _text_or_none(explicit_base_url or env.get(PROXY_BASE_URL_ENV_VAR))
if candidate and candidate.casefold() in {"off", "false", "0", "disable", "disabled", "none"}:
raise ValueError("KSKILL_PROXY_BASE_URL 가 비활성화되어 있습니다.")
if candidate and candidate != "replace-me":
return candidate.rstrip("/")
return DEFAULT_PROXY_BASE_URL
def normalize_business_number(value: Any) -> str:
raw = _text_or_none(value)
if not raw:
raise ValueError("사업자등록번호(b_no)를 입력하세요.")
normalized = re.sub(r"\D", "", raw)
if not re.fullmatch(r"\d{10}", normalized):
raise ValueError("사업자등록번호는 숫자 10자리여야 합니다.")
return normalized
def normalize_start_date(value: Any) -> str:
raw = _text_or_none(value)
if not raw:
raise ValueError("개업일자(start_dt)를 YYYYMMDD 형식으로 입력하세요.")
normalized = re.sub(r"\D", "", raw)
if not re.fullmatch(r"\d{8}", normalized):
raise ValueError("개업일자는 YYYYMMDD 형식이어야 합니다.")
try:
dt.date(int(normalized[:4]), int(normalized[4:6]), int(normalized[6:8]))
except ValueError as error:
raise ValueError("개업일자는 유효한 날짜여야 합니다.") from error
return normalized
def build_status_payload(business_numbers: list[Any]) -> dict[str, list[str]]:
numbers = [normalize_business_number(value) for value in business_numbers]
numbers = list(dict.fromkeys(numbers))
if not numbers:
raise ValueError("사업자등록번호를 1개 이상 입력하세요.")
if len(numbers) > BATCH_LIMIT:
raise ValueError("한 번에 조회할 수 있는 사업자등록번호는 100개까지입니다.")
return {"b_no": numbers}
def build_validate_business(**kwargs: Any) -> dict[str, str]:
p_nm = _text_or_none(kwargs.get("p_nm"))
if not p_nm:
raise ValueError("대표자 성명(p_nm)을 입력하세요.")
business = {
"b_no": normalize_business_number(kwargs.get("b_no")),
"start_dt": normalize_start_date(kwargs.get("start_dt")),
"p_nm": p_nm,
}
for key in ("p_nm2", "b_nm", "b_sector", "b_type", "b_adr"):
value = _text_or_none(kwargs.get(key))
if value:
business[key] = value
corp_no = _text_or_none(kwargs.get("corp_no"))
if corp_no:
business["corp_no"] = re.sub(r"\D", "", corp_no)
return business
def build_validate_payload(businesses: list[dict[str, Any]]) -> dict[str, list[dict[str, str]]]:
if not businesses:
raise ValueError("진위확인 대상 businesses를 1개 이상 입력하세요.")
if len(businesses) > BATCH_LIMIT:
raise ValueError("한 번에 진위확인할 수 있는 사업자는 100개까지입니다.")
return {"businesses": [build_validate_business(**business) for business in businesses]}
def read_json_response(request: urllib.request.Request) -> dict[str, Any]:
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
try:
payload = json.loads(body)
except json.JSONDecodeError:
payload = None
if isinstance(payload, dict) and payload.get("message"):
raise ApiError(str(payload["message"]), status_code=error.code, url=getattr(error, "url", None)) from error
raise ApiError(f"NTS business proxy request failed with HTTP {error.code}", status_code=error.code, url=getattr(error, "url", None)) from error
except urllib.error.URLError as error:
raise ApiError(f"NTS business proxy request failed: {error.reason}") from error
def _post_json(path: str, payload: dict[str, Any], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
request = urllib.request.Request(
f"{resolved_base_url}{path}",
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "k-skill-nts-business-registration/1.0",
},
method="POST",
)
return read_json(request)
def query_status(business_numbers: list[Any], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
return _post_json("/v1/nts-business/status", build_status_payload(business_numbers), base_url=base_url, read_json=read_json)
def validate_businesses(businesses: list[dict[str, Any]], *, base_url: str | None = None, read_json: Any = read_json_response) -> dict[str, Any]:
return _post_json("/v1/nts-business/validate", build_validate_payload(businesses), base_url=base_url, read_json=read_json)
def _parse_business_json(value: str) -> dict[str, Any]:
payload = json.loads(value)
if not isinstance(payload, dict):
raise argparse.ArgumentTypeError("business JSON must be an object")
return payload
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="NTS business registration status/authenticity helper")
subparsers = parser.add_subparsers(dest="command", required=True)
status = subparsers.add_parser("status", help="사업자등록번호 상태조회")
status.add_argument("--b-no", action="append", required=True, help="사업자등록번호(10자리; 하이픈 허용). 여러 번 지정 가능")
status.add_argument("--proxy-base-url")
validate = subparsers.add_parser("validate", help="사업자등록정보 진위확인")
validate.add_argument("--business-json", action="append", type=_parse_business_json, required=True, help='예: {"b_no":"1234567890","start_dt":"20200101","p_nm":"홍길동"}')
validate.add_argument("--proxy-base-url")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.command == "status":
print(json.dumps(query_status(args.b_no, base_url=args.proxy_base_url), ensure_ascii=False, indent=2))
return 0
if args.command == "validate":
print(json.dumps(validate_businesses(args.business_json, base_url=args.proxy_base_url), ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
return 1
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,98 @@
import json
import unittest
import urllib.error
from scripts.nts_business_registration import (
ApiError,
build_status_payload,
build_validate_business,
normalize_business_number,
normalize_start_date,
query_status,
resolve_proxy_base_url,
validate_businesses,
)
class NtsBusinessNormalizationTest(unittest.TestCase):
def test_normalize_business_number_keeps_ten_digits_only(self):
self.assertEqual(normalize_business_number("123-45-67890"), "1234567890")
with self.assertRaisesRegex(ValueError, "사업자등록번호"):
normalize_business_number("123")
def test_normalize_start_date_accepts_common_date_separators(self):
self.assertEqual(normalize_start_date("2020-01-31"), "20200131")
self.assertEqual(normalize_start_date("2020.01.31"), "20200131")
with self.assertRaisesRegex(ValueError, "개업일자"):
normalize_start_date("2020-13-01")
def test_build_status_payload_limits_batch_size(self):
self.assertEqual(build_status_payload(["123-45-67890"]), {"b_no": ["1234567890"]})
with self.assertRaisesRegex(ValueError, "100개"):
build_status_payload([f"{index:010d}" for index in range(101)])
def test_build_validate_business_trims_optional_fields(self):
business = build_validate_business(
b_no="123-45-67890",
start_dt="2020-01-31",
p_nm=" 홍길동 ",
b_nm="테스트상사",
corp_no="110111-1234567",
p_nm2="",
)
self.assertEqual(
business,
{
"b_no": "1234567890",
"start_dt": "20200131",
"p_nm": "홍길동",
"b_nm": "테스트상사",
"corp_no": "1101111234567",
},
)
class NtsBusinessProxyTest(unittest.TestCase):
def test_query_status_posts_to_proxy_route(self):
captured = {}
def fake_read_json(request):
captured["url"] = request.full_url
captured["data"] = json.loads(request.data.decode("utf-8"))
captured["method"] = request.get_method()
return {"data": [{"b_no": "1234567890", "b_stt": "계속사업자"}]}
payload = query_status(["123-45-67890"], base_url="https://proxy.example.com", read_json=fake_read_json)
self.assertEqual(payload["data"][0]["b_stt"], "계속사업자")
self.assertEqual(captured["url"], "https://proxy.example.com/v1/nts-business/status")
self.assertEqual(captured["data"], {"b_no": ["1234567890"]})
self.assertEqual(captured["method"], "POST")
def test_validate_businesses_posts_to_proxy_route(self):
captured = {}
def fake_read_json(request):
captured["url"] = request.full_url
captured["data"] = json.loads(request.data.decode("utf-8"))
return {"data": [{"valid": "01"}]}
payload = validate_businesses(
[{"b_no": "1234567890", "start_dt": "20200101", "p_nm": "홍길동"}],
base_url="https://proxy.example.com/",
read_json=fake_read_json,
)
self.assertEqual(payload["data"][0]["valid"], "01")
self.assertEqual(captured["url"], "https://proxy.example.com/v1/nts-business/validate")
self.assertEqual(captured["data"], {"businesses": [{"b_no": "1234567890", "start_dt": "20200101", "p_nm": "홍길동"}]})
def test_resolve_proxy_base_url_defaults_to_hosted_proxy(self):
self.assertEqual(resolve_proxy_base_url(None, env={}), "https://k-skill-proxy.nomadamas.org")
self.assertEqual(resolve_proxy_base_url(None, env={"KSKILL_PROXY_BASE_URL": "https://proxy.example.com/"}), "https://proxy.example.com")
with self.assertRaisesRegex(ValueError, "KSKILL_PROXY_BASE_URL"):
resolve_proxy_base_url(None, env={"KSKILL_PROXY_BASE_URL": "off"})
if __name__ == "__main__":
unittest.main()