openclaw/agents/budget/workspace/skills/whooing-sync/scripts/whooing_balance.py

#!/usr/bin/env python3
"""후잉 OpenAPI 잔액(bs.json) 조회.

Usage:
    whooing_balance.py                    # 모든 섹션의 현재 잔액
    whooing_balance.py --section-id 1     # 특정 섹션
    whooing_balance.py --as-of 2026-04-23 # 특정 날짜 기준
    whooing_balance.py --json             # 원시 JSON 출력
"""
from __future__ import annotations

import argparse
import json
import secrets
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path

CRED_PATH = Path("/Users/snowoyh/.openclaw/credentials/whooing.json")
BASE = "https://whooing.com/api"


def build_api_key(app_id, token, signature) -> str:
    nounce = secrets.token_hex(20)
    ts = int(time.time())
    return f"app_id={app_id},token={token},signiture={signature},nounce={nounce},timestamp={ts}"


def api_get(endpoint: str, api_key: str, params: dict | None = None) -> dict:
    url = f"{BASE}/{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"X-API-KEY": api_key})
    with urllib.request.urlopen(req, timeout=15) as resp:
        body = resp.read().decode("utf-8")
    data = json.loads(body)
    if data.get("code") != 200:
        raise RuntimeError(f"후잉 API error {data.get('code')}: {data.get('message')} (endpoint={endpoint})")
    return data["results"]


def fmt_won(n: int) -> str:
    return f"{n:,}원"


def fetch_asset_balances(account_names: list[str]) -> dict[str, int]:
    """주어진 자산 계정명들의 후잉 잔액을 dict로 반환 (호환용 별칭)."""
    return fetch_balances(account_names, sides=("assets",))


def fetch_balances(account_names: list[str], sides: tuple = ("assets", "liabilities")) -> dict[str, int]:
    """주어진 계정명들의 후잉 잔액을 dict로 반환.

    sides 에 'assets' / 'liabilities' / 'capital' 중 원하는 것만 포함.
    name -> money. 미발견 계정은 dict에 포함하지 않음.
    """
    cred = json.loads(CRED_PATH.read_text())
    api_cfg = cred["api"]

    def key() -> str:
        return build_api_key(api_cfg["app_id"], api_cfg["token"], api_cfg["signature"])

    end_date = time.strftime("%Y%m%d")
    start_date = "19000101"

    sections = api_get("sections.json", key())
    if isinstance(sections, dict):
        section_list = sections.get("sections") or sections.get("rows") or list(sections.values())
    else:
        section_list = sections

    wanted = set(account_names)
    out: dict[str, int] = {}
    for sec in section_list:
        sid = sec.get("section_id") or sec.get("id")
        accounts_raw = api_get("accounts.json", key(), {"section_id": sid})
        id_to_name: dict[str, str] = {}
        for acc_list in accounts_raw.values():
            if not isinstance(acc_list, list):
                continue
            for a in acc_list:
                id_to_name[str(a.get("account_id"))] = a.get("title") or str(a.get("account_id"))
        bs = api_get("bs.json", key(), {
            "section_id": sid, "start_date": start_date, "end_date": end_date,
        })
        for side in sides:
            for row in (bs.get(side) or {}).get("accounts", []) or []:
                name = id_to_name.get(str(row.get("account_id")), "")
                if name in wanted:
                    out[name] = row.get("money", 0)
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--section-id", help="섹션 id. 미지정 시 모든 섹션 조회.")
    ap.add_argument("--as-of", help="기준일 YYYY-MM-DD. 기본값: 오늘.")
    ap.add_argument("--json", action="store_true", help="원시 JSON 출력")
    args = ap.parse_args()

    cred = json.loads(CRED_PATH.read_text())
    api_cfg = cred.get("api")
    if not api_cfg:
        print("error: credentials/whooing.json 에 'api' 블록이 없습니다.", file=sys.stderr)
        return 2

    def key() -> str:
        return build_api_key(api_cfg["app_id"], api_cfg["token"], api_cfg["signature"])

    end_date = (args.as_of or time.strftime("%Y-%m-%d")).replace("-", "")
    start_date = "19000101"

    sections = api_get("sections.json", key())
    if isinstance(sections, dict):
        section_list = sections.get("sections") or sections.get("rows") or list(sections.values())
    else:
        section_list = sections

    if args.section_id:
        section_list = [s for s in section_list if str(s.get("section_id")) == str(args.section_id)]

    output: dict = {"as_of": end_date, "sections": []}

    for sec in section_list:
        sid = sec.get("section_id") or sec.get("id")
        title = sec.get("title") or sec.get("name") or str(sid)

        accounts_raw = api_get("accounts.json", key(), {"section_id": sid})
        id_to_name: dict[str, str] = {}
        for acc_type, acc_list in accounts_raw.items():
            if not isinstance(acc_list, list):
                continue
            for a in acc_list:
                id_to_name[str(a.get("account_id"))] = a.get("title") or str(a.get("account_id"))

        bs = api_get("bs.json", key(), {
            "section_id": sid,
            "start_date": start_date,
            "end_date": end_date,
        })

        sec_out = {"section_id": sid, "title": title, "groups": {}}
        for key_name, ko in [("assets", "자산"), ("liabilities", "부채"), ("capital", "자본")]:
            group = bs.get(key_name) or {}
            total = group.get("total", 0)
            items = []
            for row in group.get("accounts", []) or []:
                money = row.get("money", 0)
                if money == 0:
                    continue
                items.append({
                    "account_id": row.get("account_id"),
                    "name": id_to_name.get(str(row.get("account_id")), str(row.get("account_id"))),
                    "money": money,
                })
            items.sort(key=lambda x: abs(x["money"]), reverse=True)
            sec_out["groups"][ko] = {"total": total, "items": items}
        output["sections"].append(sec_out)

    if args.json:
        print(json.dumps(output, ensure_ascii=False, indent=2))
        return 0

    print(f"## 후잉 잔액 (기준일 {end_date[:4]}-{end_date[4:6]}-{end_date[6:]})\n")
    for sec_out in output["sections"]:
        print(f"### [{sec_out['section_id']}] {sec_out['title']}")
        for ko in ("자산", "부채", "자본"):
            g = sec_out["groups"][ko]
            print(f"- **{ko} 합계:** {fmt_won(g['total'])}")
            for it in g["items"]:
                print(f"  - {it['name']}: {fmt_won(it['money'])}")
        print()
    return 0


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