270 lines
9.2 KiB
Python
270 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""LS증권 OpenAPI 포털에서 TR 스펙을 수집해 data/ls-openapi-spec.json 을 만든다.
|
|
|
|
실제 시세/주문 REST 호출이 아니다. 문서 메타(가이드)만 가져온다.
|
|
요청 간격은 포털 부하를 피하기 위해 sleep 한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
BASE = "https://openapi.ls-sec.co.kr"
|
|
# 포털 API 가이드 URL에 쓰이는 고정 group_id (apiservice 쿼리)
|
|
DEFAULT_DOCS_GROUP_ID = "73142d9f-1983-48d2-8543-89b75535d34c"
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DEFAULT_OUT = os.path.join(SCRIPT_DIR, "data", "ls-openapi-spec.json")
|
|
|
|
# propertyType 코드 → 짧은 타입 라벨 (문서 가독용)
|
|
TYPE_LABEL = {
|
|
"A0001": "string",
|
|
"A0003": "object",
|
|
"A0004": "number",
|
|
"A0005": "array",
|
|
}
|
|
|
|
|
|
def _http_get_json(url: str, timeout: float = 45.0) -> Any:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": "kis_bot-ls-docs-mcp/0.1 (docs-fetch-only)",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def _strip_html(text: str) -> str:
|
|
if not text:
|
|
return ""
|
|
t = re.sub(r"<br\s*/?>", "\n", str(text), flags=re.I)
|
|
t = re.sub(r"<[^>]+>", "", t)
|
|
t = t.replace(" ", " ").replace("&", "&")
|
|
return t.strip()
|
|
|
|
|
|
def _clean_field_code(raw: str) -> str:
|
|
s = (raw or "").replace(" ", " ").strip()
|
|
s = re.sub(r"^\s*-\s*", "", s)
|
|
return s.strip()
|
|
|
|
|
|
def _category_from_group_name(name: str) -> str:
|
|
n = (name or "").strip()
|
|
if n.startswith("[") and "]" in n:
|
|
major = n[1 : n.index("]")]
|
|
minor = n[n.index("]") + 1 :].strip()
|
|
if minor:
|
|
return f"{major}_{minor}".replace(" ", "")
|
|
return major
|
|
return n or "기타"
|
|
|
|
|
|
def _parse_example(ex: Any) -> Any:
|
|
if ex is None:
|
|
return None
|
|
if isinstance(ex, (dict, list)):
|
|
return ex
|
|
if isinstance(ex, str):
|
|
s = ex.strip()
|
|
if not s:
|
|
return None
|
|
try:
|
|
return json.loads(s)
|
|
except Exception:
|
|
return s
|
|
return ex
|
|
|
|
|
|
def _normalize_props(props: list[dict[str, Any]], body_types: set[str]) -> list[dict[str, Any]]:
|
|
out: list[dict[str, Any]] = []
|
|
for p in props:
|
|
if not isinstance(p, dict):
|
|
continue
|
|
bt = str(p.get("bodyType") or "")
|
|
if bt not in body_types:
|
|
continue
|
|
code = _clean_field_code(str(p.get("propertyCd") or ""))
|
|
if not code:
|
|
continue
|
|
ptype = str(p.get("propertyType") or "")
|
|
out.append(
|
|
{
|
|
"itemId": code,
|
|
"itemNm": _strip_html(str(p.get("propertyNm") or "")),
|
|
"type": TYPE_LABEL.get(ptype, ptype),
|
|
"length": p.get("propertyLength") or "",
|
|
"required": str(p.get("requireYn") or ""),
|
|
"order": str(p.get("propertyOrder") or ""),
|
|
"bodyType": bt,
|
|
"desc": _strip_html(str(p.get("description") or "")),
|
|
}
|
|
)
|
|
out.sort(key=lambda x: (x.get("order") or "", x.get("itemId") or ""))
|
|
return out
|
|
|
|
|
|
def fetch_all(*, sleep_s: float, out_path: str, docs_group_id: str) -> dict[str, Any]:
|
|
list_url = f"{BASE}/api/apis/public?page=0&size=500"
|
|
print(f"[fetch] groups: {list_url}", flush=True)
|
|
groups = _http_get_json(list_url)
|
|
if not isinstance(groups, list):
|
|
raise RuntimeError("public API list is not a list")
|
|
print(f"[fetch] groups={len(groups)}", flush=True)
|
|
time.sleep(sleep_s)
|
|
|
|
apis: dict[str, Any] = {}
|
|
dupes: list[str] = []
|
|
errors: list[dict[str, str]] = []
|
|
|
|
for gi, g in enumerate(groups, 1):
|
|
if not isinstance(g, dict):
|
|
continue
|
|
group_id = str(g.get("id") or "").strip()
|
|
group_name = str(g.get("name") or "").strip()
|
|
access_url = str(g.get("accessUrl") or "").strip()
|
|
method = str(g.get("httpMethod") or "POST").strip() or "POST"
|
|
domain = str(g.get("domain") or f"{BASE}:8080").strip()
|
|
mock_domain = str(g.get("simulatedDomain") or "").strip()
|
|
protocol = str(g.get("protocolType") or "REST").strip() or "REST"
|
|
content_type = str(g.get("contentType") or "").strip()
|
|
description = _strip_html(str(g.get("description") or ""))
|
|
category = _category_from_group_name(group_name)
|
|
docs_url = (
|
|
f"{BASE}/apiservice?group_id={docs_group_id}&api_id={group_id}"
|
|
if group_id
|
|
else ""
|
|
)
|
|
|
|
print(f"[fetch] ({gi}/{len(groups)}) {group_name} {group_id}", flush=True)
|
|
try:
|
|
trs = _http_get_json(f"{BASE}/api/apis/guide/tr/{group_id}")
|
|
except Exception as e:
|
|
errors.append({"group": group_name, "stage": "tr_list", "error": str(e)})
|
|
print(f" ! tr_list fail: {e}", flush=True)
|
|
time.sleep(sleep_s)
|
|
continue
|
|
if not isinstance(trs, list):
|
|
errors.append({"group": group_name, "stage": "tr_list", "error": "not a list"})
|
|
time.sleep(sleep_s)
|
|
continue
|
|
time.sleep(sleep_s)
|
|
|
|
for tr in trs:
|
|
if not isinstance(tr, dict):
|
|
continue
|
|
tr_id = str(tr.get("id") or "").strip()
|
|
tr_code = str(tr.get("trCode") or "").strip()
|
|
tr_name = str(tr.get("trName") or "").strip()
|
|
if not tr_code:
|
|
continue
|
|
tps = str(tr.get("transactionPerSec") or "").strip()
|
|
|
|
props: list[dict[str, Any]] = []
|
|
if tr_id:
|
|
try:
|
|
raw_props = _http_get_json(f"{BASE}/api/apis/guide/tr/property/{tr_id}")
|
|
if isinstance(raw_props, list):
|
|
props = [x for x in raw_props if isinstance(x, dict)]
|
|
except Exception as e:
|
|
errors.append(
|
|
{
|
|
"group": group_name,
|
|
"tr": tr_code,
|
|
"stage": "property",
|
|
"error": str(e),
|
|
}
|
|
)
|
|
print(f" ! property fail {tr_code}: {e}", flush=True)
|
|
time.sleep(sleep_s)
|
|
|
|
entry = {
|
|
"apiId": tr_code,
|
|
"apiNm": tr_name or tr_code,
|
|
"trCode": tr_code,
|
|
"trName": tr_name,
|
|
"trId": tr_id,
|
|
"method": method,
|
|
"url": access_url,
|
|
"domain": domain,
|
|
"mockDomain": mock_domain,
|
|
"protocolType": protocol,
|
|
"format": "JSON",
|
|
"contentType": content_type,
|
|
"category": category,
|
|
"groupName": group_name,
|
|
"groupId": group_id,
|
|
"transactionPerSec": tps,
|
|
"docsUrl": docs_url,
|
|
"description": description,
|
|
"requestIo": _normalize_props(props, {"req_h", "req_b"}),
|
|
"responseIo": _normalize_props(props, {"res_h", "res_b"}),
|
|
"requestExample": _parse_example(tr.get("reqExample")),
|
|
"responseExample": _parse_example(tr.get("resExample")),
|
|
}
|
|
|
|
store_id = tr_code
|
|
if store_id in apis:
|
|
# 동일 TR이 여러 그룹에 있으면 groupId로 구분
|
|
alt = f"{tr_code}@{group_id[:8]}"
|
|
dupes.append(f"{tr_code} -> {alt} ({group_name})")
|
|
store_id = alt
|
|
entry["apiId"] = store_id
|
|
apis[store_id] = entry
|
|
print(f" + {tr_code} ({tr_name}) fields={len(props)}", flush=True)
|
|
|
|
meta = {
|
|
"fetchedAt": datetime.now(timezone.utc).isoformat(),
|
|
"source": BASE,
|
|
"groupCount": len(groups),
|
|
"trCount": len(apis),
|
|
"docsGroupId": docs_group_id,
|
|
"duplicateKeys": dupes,
|
|
"errors": errors,
|
|
}
|
|
out = {"_meta": meta, **apis}
|
|
os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump(out, f, ensure_ascii=False, indent=2)
|
|
f.write("\n")
|
|
print(f"[done] wrote {out_path} trs={len(apis)} errors={len(errors)}", flush=True)
|
|
return meta
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Fetch LS OpenAPI docs catalog")
|
|
p.add_argument("--out", default=DEFAULT_OUT, help="output JSON path")
|
|
p.add_argument(
|
|
"--sleep",
|
|
type=float,
|
|
default=float(os.environ.get("LS_DOCS_FETCH_SLEEP", "0.12")),
|
|
help="seconds between HTTP calls (default 0.12)",
|
|
)
|
|
p.add_argument(
|
|
"--docs-group-id",
|
|
default=DEFAULT_DOCS_GROUP_ID,
|
|
help="apiservice group_id for docsUrl",
|
|
)
|
|
args = p.parse_args()
|
|
try:
|
|
fetch_all(sleep_s=max(0.05, args.sleep), out_path=args.out, docs_group_id=args.docs_group_id)
|
|
except urllib.error.URLError as e:
|
|
print(f"FATAL: {e}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|