hyowons
306 lines
12 KiB
Python
306 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""후잉 OpenAPI 클라이언트 — 잔액(bs.json) 조회 + 거래(entries.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 re
|
|
import secrets
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timedelta
|
|
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 api_key_from_cred() -> str:
|
|
"""credentials 에서 API 키 문자열 1회 생성."""
|
|
api_cfg = json.loads(CRED_PATH.read_text())["api"]
|
|
return build_api_key(api_cfg["app_id"], api_cfg["token"], api_cfg["signature"])
|
|
|
|
|
|
def api_request(method: str, endpoint: str, api_key: str,
|
|
params: dict | None = None, body: dict | None = None) -> dict:
|
|
"""후잉 OpenAPI 호출. GET 은 쿼리스트링, PUT 은 본문으로 파라미터를 보낸다.
|
|
|
|
⚠️ DELETE 는 쓸 수 없다. 후잉 서버가 DELETE 요청의 파라미터를 쿼리·본문(urlencoded/JSON)·
|
|
쿠키·경로 어디에서도 읽지 못해 항상 `section_id parameter is required` 로 실패한다
|
|
(2026-08-04 전수 실측. 같은 본문을 PUT 으로 보내면 정상이라 후잉 쪽 문제다).
|
|
거래를 지우는 대신 PUT 으로 금액을 줄이는 이유가 이것이다.
|
|
"""
|
|
url = f"{BASE}/{endpoint}"
|
|
if params:
|
|
url += "?" + urllib.parse.urlencode(params)
|
|
data = None
|
|
if body:
|
|
# 후잉은 '+' 를 공백으로 디코드하지 않는다. quote_via=quote 로 공백을 %20 으로 보낸다.
|
|
data = urllib.parse.urlencode(body, quote_via=urllib.parse.quote).encode()
|
|
req = urllib.request.Request(url, data=data, method=method, headers={"X-API-KEY": api_key})
|
|
if data:
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
parsed = json.loads(resp.read().decode("utf-8"))
|
|
if parsed.get("code") != 200:
|
|
raise RuntimeError(
|
|
f"후잉 API error {parsed.get('code')}: {parsed.get('message')} "
|
|
f"{parsed.get('error_parameters') or ''} ({method} {endpoint})"
|
|
)
|
|
return parsed["results"]
|
|
|
|
|
|
def _section_list(api_key: str) -> list:
|
|
sections = api_request("GET", "sections.json", api_key)
|
|
if isinstance(sections, dict):
|
|
return sections.get("sections") or sections.get("rows") or list(sections.values())
|
|
return sections
|
|
|
|
|
|
def _name_to_account_id(api_key: str, section_id) -> dict[str, str]:
|
|
raw = api_request("GET", "accounts.json", api_key, {"section_id": section_id})
|
|
out: dict[str, str] = {}
|
|
for acc_list in raw.values():
|
|
if not isinstance(acc_list, list):
|
|
continue
|
|
for a in acc_list:
|
|
aid = str(a.get("account_id"))
|
|
out[a.get("title") or aid] = aid
|
|
return out
|
|
|
|
|
|
def _memo_has_merchant(memo: str, merchant: str) -> bool:
|
|
"""memo 에 merchant 가 토큰 단위로 들어있는지.
|
|
|
|
단순 substring 이면 '쿠팡' 이 '쿠팡페이주' 승인까지 잡아 엉뚱한 거래를 감액한다
|
|
(2026-08-04 실제로 두 상호가 같은 날 같은 금액으로 있었다). 카드 SMS 는 상호를
|
|
줄바꿈·공백으로 구분하므로 앞뒤가 공백이거나 문자열 끝일 때만 인정한다.
|
|
"""
|
|
if not memo or not merchant:
|
|
return False
|
|
for m in re.finditer(re.escape(merchant), memo):
|
|
before = memo[m.start() - 1] if m.start() > 0 else " "
|
|
after = memo[m.end()] if m.end() < len(memo) else " "
|
|
if before.isspace() and after.isspace():
|
|
return True
|
|
return False
|
|
|
|
|
|
def find_card_entries(card_account: str, money: int, merchant: str,
|
|
on_date: str, window_days: int = 30) -> list[dict]:
|
|
"""카드 취소의 원본 승인 거래 후보를 찾는다.
|
|
|
|
card_account 로 결제된 거래 중 memo 에 merchant 가 들어있고 금액이 취소액 이상인 것을
|
|
[on_date - window_days, on_date] 구간에서 찾는다. 금액 일치(전액취소)를 우선 반환하고,
|
|
없으면 초과분(부분취소 후보)을 반환한다.
|
|
|
|
⚠️ item 이 아니라 memo 로 매칭한다 — merchant_map 의 contains 룰이 item 을 재작성해서
|
|
(동행복권→'복권') item 은 상호와 다를 수 있지만, memo 는 항상 승인 SMS 원문이다.
|
|
⚠️ 판단은 호출측이 한다. 정확히 1건일 때만 손대는 게 안전하다.
|
|
"""
|
|
api_key = api_key_from_cred()
|
|
start = (datetime.strptime(on_date, "%Y%m%d") - timedelta(days=window_days)).strftime("%Y%m%d")
|
|
exact, over = [], []
|
|
for sec in _section_list(api_key):
|
|
sid = sec.get("section_id") or sec.get("id")
|
|
card_id = _name_to_account_id(api_key, sid).get(card_account)
|
|
if not card_id:
|
|
continue
|
|
rows = api_request("GET", "entries.json", api_key, {
|
|
"section_id": sid, "start_date": start, "end_date": on_date,
|
|
})
|
|
for row in (rows.get("rows") or []):
|
|
if str(row.get("r_account_id")) != card_id:
|
|
continue
|
|
if not _memo_has_merchant(row.get("memo") or "", merchant):
|
|
continue
|
|
row_money = row.get("money") or 0
|
|
if row_money == money:
|
|
exact.append({**row, "section_id": sid})
|
|
elif row_money > money:
|
|
over.append({**row, "section_id": sid})
|
|
return exact or over
|
|
|
|
|
|
def set_entry_money(entry: dict, money: int) -> None:
|
|
"""거래 금액을 money 로 바꾼다(PUT). money=0 도 후잉이 받는다(2026-08-04 실측).
|
|
|
|
조회한 나머지 필드를 그대로 되돌려보내 보존한다 — 부분 PUT 의 필드 보존 동작에
|
|
기대지 않는 편이 안전하다. entry_date 는 조회 시 '20260804.0000' 형태라 소수부를 뗀다.
|
|
"""
|
|
api_request("PUT", f"entries/{entry['entry_id']}.json", api_key_from_cred(), body={
|
|
"section_id": entry["section_id"],
|
|
"entry_date": str(entry.get("entry_date") or "").split(".")[0],
|
|
"l_account_id": entry.get("l_account_id"),
|
|
"r_account_id": entry.get("r_account_id"),
|
|
"item": entry.get("item") or "",
|
|
"money": str(money),
|
|
"memo": entry.get("memo") or "",
|
|
})
|
|
|
|
|
|
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())
|