Update wehago matching logic and exclude reports
This commit is contained in:
b17301
@@ -0,0 +1,162 @@
|
||||
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())
|
||||
Reference in New Issue
Block a user