HM_project_Viewer_Board/scripts/inspect_wehago_ledger_api_9225.py

from __future__ import annotations

import argparse
import json
import time
from datetime import datetime
from pathlib import Path
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from selenium import webdriver
from selenium.common.exceptions import WebDriverException
from selenium.webdriver.chrome.options import Options

import refresh_hanmac_wehago_ledgers as refresh
import wehago_data_download_2022_work as wehago


OUTPUT_DIR = Path(__file__).resolve().parents[1] / "runtime_diagnostics"
SENSITIVE_KEYS = {
    "access_token",
    "authorization",
    "cookie",
    "password",
    "refresh_token",
    "session",
    "token",
}


def redact_url(url: str) -> str:
    parts = urlsplit(url)
    query = [
        (key, "<redacted>" if any(secret in key.lower() for secret in SENSITIVE_KEYS) else value)
        for key, value in parse_qsl(parts.query, keep_blank_values=True)
    ]
    return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))


def redact_payload(value: object) -> object:
    if isinstance(value, dict):
        return {
            key: "<redacted>" if any(secret in key.lower() for secret in SENSITIVE_KEYS) else redact_payload(item)
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [redact_payload(item) for item in value]
    return value


def safe_post_data(raw: str | None) -> object:
    if not raw:
        return None
    try:
        return redact_payload(json.loads(raw))
    except json.JSONDecodeError:
        pairs = parse_qsl(raw, keep_blank_values=True)
        if pairs:
            return redact_payload(dict(pairs))
        return raw[:4000]


def drain_network(driver: webdriver.Chrome) -> list[dict[str, object]]:
    entries: list[dict[str, object]] = []
    for item in driver.get_log("performance"):
        try:
            message = json.loads(item["message"])["message"]
        except (KeyError, TypeError, json.JSONDecodeError):
            continue
        method = message.get("method")
        params = message.get("params") or {}
        if method == "Network.requestWillBeSent":
            request = params.get("request") or {}
            url = str(request.get("url") or "")
            if url.startswith("http"):
                entries.append(
                    {
                        "event": "request",
                        "request_id": params.get("requestId"),
                        "method": request.get("method"),
                        "url": redact_url(url),
                        "header_names": sorted((request.get("headers") or {}).keys()),
                        "post_data": safe_post_data(request.get("postData")),
                    }
                )
        elif method == "Network.responseReceived":
            response = params.get("response") or {}
            url = str(response.get("url") or "")
            mime_type = str(response.get("mimeType") or "")
            if not url.startswith("http") or "json" not in mime_type.lower():
                continue
            body: object = None
            try:
                raw_body = driver.execute_cdp_cmd(
                    "Network.getResponseBody",
                    {"requestId": params.get("requestId")},
                ).get("body", "")
                body = redact_payload(json.loads(raw_body))
            except (WebDriverException, json.JSONDecodeError, AttributeError):
                body = None
            entries.append(
                {
                    "event": "response",
                    "request_id": params.get("requestId"),
                    "status": response.get("status"),
                    "url": redact_url(url),
                    "mime_type": mime_type,
                    "body": body,
                }
            )
    return entries


def main() -> int:
    parser = argparse.ArgumentParser(description="로그인된 WEHAGO 계정별원장 화면의 API 요청을 안전하게 진단합니다.")
    parser.add_argument("--year", type=int, default=2024, choices=refresh.YEARS)
    parser.add_argument("--accounts", nargs="+", default=["223", "231"])
    parser.add_argument("--debugger-address", default="127.0.0.1:9225")
    args = parser.parse_args()

    options = Options()
    options.add_experimental_option("debuggerAddress", args.debugger_address)
    options.set_capability("goog:loggingPrefs", {"performance": "ALL"})
    driver = webdriver.Chrome(options=options)
    driver.set_page_load_timeout(30)
    driver.set_script_timeout(15)
    driver.execute_cdp_cmd("Network.enable", {})

    target_by_code = {account.code: account for account in wehago.ACCOUNTS}
    captures: list[dict[str, object]] = []
    try:
        driver.get(refresh.HANMAC_LEDGER_URL.format(year=args.year, gisu=refresh.GISU_BY_YEAR[args.year]))
        wehago.ensure_ledger_data_loaded(driver, wehago.ACCOUNTS)
        drain_network(driver)

        for code in args.accounts:
            account = target_by_code.get(code, wehago.Account(code, ""))
            before = wehago.detail_grid_signature(driver)
            wehago.select_account_from_left_list(driver, account)
            wehago.assert_selected_account(driver, account)
            time.sleep(1.5)
            after = wehago.detail_grid_signature(driver)
            captures.append(
                {
                    "account_code": account.code,
                    "account_name": account.name,
                    "detail_changed": bool(after and after != before),
                    "detail_signature": after[:500],
                    "network": drain_network(driver),
                }
            )
    finally:
        driver.quit()

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    output = OUTPUT_DIR / f"wehago_ledger_api_{args.year}_{datetime.now():%Y%m%d_%H%M%S}.json"
    output.write_text(json.dumps(captures, ensure_ascii=False, indent=2), encoding="utf-8")
    print(output)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())