kis_trader/remove/legacy_root/test_kiwoom_token.py

#!/usr/bin/env python3
"""
키움 OpenAPI 토큰 발급 진단 스크립트
====================================

DB(env_config 최신 스냅샷)에 저장된 키움 키 4종 (MOCK·REAL × KEY·SECRET)을
자동으로 읽어 모의/실전 도메인 각각에 ``POST /oauth2/token`` 을 던지고,
응답·키 메타정보(길이·공백포함·앞뒤 일부)를 한 번에 보여 준다.

키움 토큰 엔드포인트 (kis_ws.KiwoomTokenManager._request_new_token 와 동일):
  - 실전: https://api.kiwoom.com/oauth2/token
  - 모의: https://mockapi.kiwoom.com/oauth2/token
  - body: {"grant_type":"client_credentials","appkey":..., "secretkey":...}

사용법
------

DB 키로 자동 테스트::

    python3 test_kiwoom_token.py

인터랙티브 입력 (키에 특수문자/공백 있을 때 가장 안전)::

    python3 test_kiwoom_token.py --manual
    python3 test_kiwoom_token.py --manual --mode real

CLI 인자로 직접 입력 (키 값은 반드시 ``"…"`` 로 감싸세요)::

    python3 test_kiwoom_token.py --key "<APPKEY>" --secret "<SECRET>" --mode mock
    python3 test_kiwoom_token.py --key="<APPKEY>" --secret="<SECRET>" --mode real

자주 보는 에러 코드
-------------------

- 8001: App Key / Secret 검증 실패 → 키 자체가 틀렸거나 OpenAPI 미신청 / 만료
- 1700: 요청 개수 초과 (au10001 rate limit) → 1분 정도 기다렸다가 재시도
- 9999: 서버 일시 오류 → 재시도

이 스크립트는 토큰 발급만 하므로 계좌번호는 사용하지 않는다 (키움 시세 API 정책).
"""

from __future__ import annotations

import argparse
import getpass
import json
import sys
from pathlib import Path
from typing import Optional, Tuple

import requests

SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))


# ──────────────────────────────────────────────────────────────────────
# 도메인 / 모드
# ──────────────────────────────────────────────────────────────────────
DOMAINS = {
    "mock": "mockapi.kiwoom.com",
    "real": "api.kiwoom.com",
}


# ──────────────────────────────────────────────────────────────────────
# 키 메타정보 진단
# ──────────────────────────────────────────────────────────────────────
def _safe_preview(s: str, head: int = 8, tail: int = 4) -> str:
    """키를 가운데를 가린 형태로 표시 (앞8자 …끝4자)."""
    if not s:
        return "(empty)"
    if len(s) <= head + tail:
        return repr(s)
    return f"{s[:head]}…{s[-tail:]}"


def _has_whitespace(s: str) -> bool:
    return any(c in s for c in (" ", "\t", "\n", "\r", "\v", "\f"))


def diagnose_key(label: str, value: Optional[str]) -> bool:
    """키 값 메타 출력. 사용 가능하면 True."""
    v = (value or "")
    raw_len = len(v)
    stripped = v.strip()
    has_ws = _has_whitespace(v)
    print(f"  • {label:28s} len={raw_len:3d}  preview={_safe_preview(stripped)}"
          f"  whitespace={'YES⚠️ ' if has_ws else 'no '}"
          f"  {'(empty)' if not stripped else ''}")
    return bool(stripped)


# ──────────────────────────────────────────────────────────────────────
# 토큰 발급
# ──────────────────────────────────────────────────────────────────────
def request_token(
    appkey: str,
    secretkey: str,
    mode: str,
    timeout: int = 10,
) -> Tuple[bool, dict]:
    """
    키움 토큰 발급 단일 호출.

    Returns:
        (success, response_data_dict)
        success=True 이면 ``response_data_dict`` 안에 token 이 들어 있음.
    """
    if mode not in DOMAINS:
        raise ValueError(f"mode must be 'mock' or 'real', got {mode!r}")

    url = f"https://{DOMAINS[mode]}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "appkey":     appkey,
        "secretkey":  secretkey,
    }

    try:
        resp = requests.post(url, json=payload, timeout=timeout)
    except requests.RequestException as e:
        return False, {"_exception": str(e), "_url": url}

    try:
        data = resp.json()
    except ValueError:
        return False, {"_status": resp.status_code, "_text": resp.text[:500]}

    data["_status"] = resp.status_code
    token = (data.get("token") or data.get("access_token") or "").strip()
    return bool(token), data


def explain_error(data: dict) -> str:
    """에러 응답 → 사람이 읽는 진단 메시지."""
    if "_exception" in data:
        return f"네트워크 예외: {data['_exception']}"

    code = str(data.get("return_code", "")).strip()
    msg  = str(data.get("return_msg",  "")).strip()

    if "8001" in msg or code == "3":
        return ("[8001] App Key / Secret 검증 실패\n"
                "       → 키 자체가 잘못되었거나 OpenAPI 신청이 안 된 상태.\n"
                "       → openapi.kiwoom.com 에서 모의/실전 OpenAPI 신청 여부 확인.\n"
                "       → 키를 사이트에서 다시 복사해 DB 재등록 (앞뒤 공백 주의).")
    if "1700" in msg or code == "5":
        return ("[1700] 요청 개수 초과 (au10001 rate limit)\n"
                "       → 1분 정도 기다렸다가 재시도. 키는 정상일 수 있음.")
    if "9999" in msg:
        return "[9999] 서버 일시 오류 → 재시도"
    return f"기타 오류: code={code} msg={msg}"


def run_test(appkey: str, secretkey: str, mode: str, label: str) -> None:
    """단일 모드 테스트 + 결과 출력."""
    print(f"\n── [{label}] mode={mode}  domain={DOMAINS[mode]} ──")

    if not appkey or not secretkey:
        print(f"  ⏭  키 없음 (appkey={'∅' if not appkey else 'OK'}, "
              f"secretkey={'∅' if not secretkey else 'OK'}) — 건너뜀")
        return

    ok, data = request_token(appkey.strip(), secretkey.strip(), mode)
    if ok:
        token   = data.get("token") or data.get("access_token")
        exp_dt  = data.get("expires_dt", "?")
        exp_in  = data.get("expires_in", "?")
        print(f"  ✅ 발급 성공!  token={_safe_preview(token, 8, 4)}  "
              f"expires_dt={exp_dt}  expires_in={exp_in}")
    else:
        status = data.pop("_status", "?")
        print(f"  ❌ 발급 실패  HTTP={status}")
        print(f"     응답: {json.dumps(data, ensure_ascii=False)}")
        print(f"     진단: {explain_error(data)}")


# ──────────────────────────────────────────────────────────────────────
# DB 키 로드
# ──────────────────────────────────────────────────────────────────────
def load_keys_from_db() -> dict:
    """
    DB(env_config 최신 스냅샷)에서 키움 4개 키 + 레거시 폴백 + KIS_MOCK 로드.
    """
    try:
        from database import TradeDB  # type: ignore
    except ImportError as e:
        print(f"⚠️  database 모듈 import 실패: {e}")
        return {}

    db_path = SCRIPT_DIR / "quant_bot.db"
    try:
        db = TradeDB(db_path=str(db_path))
    except TypeError:
        db = TradeDB()

    latest = db.get_latest_env()
    snap = (latest or {}).get("snapshot") or {}
    db.close()

    return {
        "KIWOOM_APP_KEY_MOCK":    str(snap.get("KIWOOM_APP_KEY_MOCK",    "") or ""),
        "KIWOOM_APP_SECRET_MOCK": str(snap.get("KIWOOM_APP_SECRET_MOCK", "") or ""),
        "KIWOOM_APP_KEY_REAL":    str(snap.get("KIWOOM_APP_KEY_REAL",    "") or ""),
        "KIWOOM_APP_SECRET_REAL": str(snap.get("KIWOOM_APP_SECRET_REAL", "") or ""),
        # 레거시 폴백
        "KIWOOM_APP_KEY":         str(snap.get("KIWOOM_APP_KEY",         "") or ""),
        "KIWOOM_APP_SECRET":      str(snap.get("KIWOOM_APP_SECRET",      "") or ""),
        "KIS_MOCK":               str(snap.get("KIS_MOCK", "true") or "true"),
    }


# ──────────────────────────────────────────────────────────────────────
# main
# ──────────────────────────────────────────────────────────────────────
def _prompt_manual_keys() -> Tuple[str, str]:
    """
    인터랙티브 입력 — 키에 ``--`` ``=`` ``$`` ``"`` 같은 특수문자가 있어도
    셸 escaping 영향 없이 안전하게 받음. secretkey 는 getpass 로 화면 가림.
    """
    print("\n📝 인터랙티브 입력 (Enter 로 확정, secretkey 는 화면에 안 찍힘)")
    appkey    = input("  appkey   : ").strip()
    secretkey = getpass.getpass("  secretkey: ").strip()
    if not appkey or not secretkey:
        print("❌ 빈 값은 허용되지 않습니다.")
        sys.exit(2)
    return appkey, secretkey


def main() -> int:
    p = argparse.ArgumentParser(
        description="키움 OpenAPI 토큰 발급 진단",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    p.add_argument("--key",    help='직접 입력할 appkey (값은 "…" 로 감싸길 권장)')
    p.add_argument("--secret", help='직접 입력할 secretkey (값은 "…" 로 감싸길 권장)')
    p.add_argument("--manual", action="store_true",
                   help="인터랙티브 입력 모드 (키에 특수문자 있을 때 권장)")
    p.add_argument("--mode",   choices=("mock", "real", "both"), default="both",
                   help="테스트 도메인 (default: both)")
    args = p.parse_args()

    # ── 인터랙티브 모드 ────────────────────────────────────────────
    if args.manual:
        print("=" * 70)
        print("🔧 인터랙티브 입력 모드 (DB 무시)")
        print("=" * 70)
        appkey, secretkey = _prompt_manual_keys()
        diagnose_key("입력 APPKEY",    appkey)
        diagnose_key("입력 SECRETKEY", secretkey)
        modes = ("mock", "real") if args.mode == "both" else (args.mode,)
        for m in modes:
            run_test(appkey, secretkey, m, label=f"입력키→{m.upper()}")
        return 0

    # ── CLI 인자 직접 입력 모드 ─────────────────────────────────────
    if args.key or args.secret:
        if not (args.key and args.secret):
            print("❌ --key 와 --secret 은 함께 지정해야 합니다.\n"
                  "   특수문자 때문에 까다롭다면 --manual 을 사용하세요.\n"
                  "   예시: python3 test_kiwoom_token.py --manual --mode real")
            return 2
        print("=" * 70)
        print("🔧 CLI 인자 입력 모드 (DB 무시)")
        print("=" * 70)
        diagnose_key("입력 APPKEY",    args.key)
        diagnose_key("입력 SECRETKEY", args.secret)
        modes = ("mock", "real") if args.mode == "both" else (args.mode,)
        for m in modes:
            run_test(args.key, args.secret, m, label=f"입력키→{m.upper()}")
        return 0

    # ── DB 자동 모드 ──────────────────────────────────────────────
    print("=" * 70)
    print("🔍 키움 토큰 진단 — DB env_config 최신 스냅샷")
    print("=" * 70)

    keys = load_keys_from_db()
    if not keys:
        print("❌ DB 에서 키를 불러오지 못했습니다.")
        return 1

    print(f"\nKIS_MOCK = {keys['KIS_MOCK']!r}  "
          f"(코드 정책: 시세는 항상 실키, 단 본 스크립트는 둘 다 시도)")
    print("\n📦 DB에 저장된 키 메타:")
    diagnose_key("KIWOOM_APP_KEY_MOCK",    keys["KIWOOM_APP_KEY_MOCK"])
    diagnose_key("KIWOOM_APP_SECRET_MOCK", keys["KIWOOM_APP_SECRET_MOCK"])
    diagnose_key("KIWOOM_APP_KEY_REAL",    keys["KIWOOM_APP_KEY_REAL"])
    diagnose_key("KIWOOM_APP_SECRET_REAL", keys["KIWOOM_APP_SECRET_REAL"])
    if keys["KIWOOM_APP_KEY"] or keys["KIWOOM_APP_SECRET"]:
        print("  (레거시 폴백)")
        diagnose_key("KIWOOM_APP_KEY",    keys["KIWOOM_APP_KEY"])
        diagnose_key("KIWOOM_APP_SECRET", keys["KIWOOM_APP_SECRET"])

    # ── 호출 ─────────────────────────────────────────────────────
    if args.mode in ("mock", "both"):
        run_test(
            keys["KIWOOM_APP_KEY_MOCK"],
            keys["KIWOOM_APP_SECRET_MOCK"],
            "mock",
            label="MOCK키→모의도메인",
        )

    if args.mode in ("real", "both"):
        run_test(
            keys["KIWOOM_APP_KEY_REAL"],
            keys["KIWOOM_APP_SECRET_REAL"],
            "real",
            label="REAL키→실전도메인",
        )

    # 레거시 키도 있으면 둘 다 찔러본다 (어느 도메인에서 통하는지 확인)
    if keys["KIWOOM_APP_KEY"] and keys["KIWOOM_APP_SECRET"]:
        if args.mode in ("mock", "both"):
            run_test(
                keys["KIWOOM_APP_KEY"], keys["KIWOOM_APP_SECRET"],
                "mock", label="LEGACY키→모의도메인",
            )
        if args.mode in ("real", "both"):
            run_test(
                keys["KIWOOM_APP_KEY"], keys["KIWOOM_APP_SECRET"],
                "real", label="LEGACY키→실전도메인",
            )

    print("\n" + "=" * 70)
    print("✅ 진단 끝")
    print("=" * 70)
    print("팁:")
    print("  • whitespace=YES 가 보이면 → DB 에 공백 섞임. 키 다시 등록 필요.")
    print("  • 8001 만 계속 → openapi.kiwoom.com 에서 OpenAPI 신청 여부 확인.")
    print("  • 1700 → 1분 후 재시도 (rate limit).")
    print("  • 둘 다 실패해도 매매에는 영향 없음 — 갭보정만 비활성됨.")
    return 0


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