365 lines
15 KiB
Python
Executable File
365 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Capture local Streamlit browser evidence with Chrome and a synthetic DB.
|
|
|
|
This script is intentionally local-only. It creates a synthetic fixture database,
|
|
starts Streamlit against that database, and uses an installed Chrome binary to
|
|
save page screenshots and DOM snapshots. It must not target the operating ledger
|
|
DB or real financial statement files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import websockets
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
DEFAULT_PAGES = ["요약", "분류 규칙", "확인필요 큐", "Import"]
|
|
DEFAULT_CHROME = "/usr/bin/google-chrome-stable"
|
|
DEFAULT_STREAMLIT = Path(".venv/bin/streamlit")
|
|
DEFAULT_OUTPUT_ROOT = Path("docs/batches/bat-014_20260703_adapter-surface-narrowing-baseline-refresh/evidence")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Capture Streamlit browser screenshots and DOM snapshots with a synthetic DB.")
|
|
parser.add_argument("--chrome", default=os.getenv("CHROME_BIN", DEFAULT_CHROME), help="Chrome executable path")
|
|
parser.add_argument("--streamlit", default=str(DEFAULT_STREAMLIT), help="Streamlit executable path")
|
|
parser.add_argument("--port", type=int, default=18501, help="Local Streamlit port")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Local Streamlit host")
|
|
parser.add_argument("--output-dir", default="", help="Evidence output directory. Defaults under docs/batches/BAT-004/evidence/run-<timestamp>.")
|
|
parser.add_argument("--page", action="append", dest="pages", help="Page to capture. Repeatable. Defaults to core E2E evidence pages.")
|
|
parser.add_argument("--keep-server", action="store_true", help="Leave Streamlit running after capture for manual inspection.")
|
|
return parser.parse_args()
|
|
|
|
|
|
def slugify_page(name: str) -> str:
|
|
mapping = {
|
|
"분류 규칙": "rules",
|
|
"확인필요 큐": "transfer-review",
|
|
"Import": "import",
|
|
"계좌": "accounts",
|
|
"요약": "overview",
|
|
"거래내역": "transactions",
|
|
"미분류": "uncategorized",
|
|
"카테고리": "categories",
|
|
}
|
|
return mapping.get(name, urllib.parse.quote(name, safe="").replace("%", "").lower() or "page")
|
|
|
|
|
|
def require_executable(path: str, label: str) -> str:
|
|
resolved = shutil.which(path) if os.path.sep not in path else path
|
|
if not resolved or not Path(resolved).exists():
|
|
raise SystemExit(f"{label} executable not found: {path}")
|
|
return resolved
|
|
|
|
|
|
def assert_synthetic_db_path(path: Path) -> None:
|
|
resolved = path.expanduser().resolve()
|
|
operating = Path.home() / ".local" / "share" / "jarca" / "ledger.sqlite3"
|
|
if resolved == operating.resolve():
|
|
raise SystemExit("Refusing to use the operating ledger DB for browser evidence")
|
|
if not str(resolved).endswith("fixture.sqlite3"):
|
|
raise SystemExit(f"Synthetic evidence DB must end with fixture.sqlite3: {resolved}")
|
|
|
|
|
|
def seed_fixture_db(db_path: Path) -> dict[str, Any]:
|
|
assert_synthetic_db_path(db_path)
|
|
if db_path.exists():
|
|
db_path.unlink()
|
|
init_db(str(db_path))
|
|
service.seed_default_ledger_categories(db_path=str(db_path))
|
|
service.seed_default_category_rules(db_path=str(db_path))
|
|
|
|
account = service.create_ledger_account(
|
|
"Synthetic Evidence Checking",
|
|
institution="Synthetic Bank",
|
|
account_number_masked="SYNTH-0000",
|
|
holder_name="Synthetic Holder",
|
|
last_balance=1_250_000,
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
statement_status="partial_import",
|
|
liquidity_bucket="cash",
|
|
memo="browser evidence synthetic fixture",
|
|
db_path=str(db_path),
|
|
)["account"]
|
|
account_id = account["id"]
|
|
service.create_ledger_account(
|
|
"Synthetic Investment Account",
|
|
type="investment",
|
|
institution="Synthetic Securities",
|
|
last_balance=5_000_000,
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
liquidity_bucket="investment",
|
|
memo="browser evidence investment asset",
|
|
db_path=str(db_path),
|
|
)
|
|
service.create_ledger_account(
|
|
"Synthetic Mortgage Balance",
|
|
type="loan",
|
|
institution="Synthetic Bank",
|
|
last_balance=2_000_000,
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
liquidity_bucket="loan",
|
|
memo="browser evidence liability",
|
|
db_path=str(db_path),
|
|
)
|
|
service.create_ledger_account(
|
|
"Synthetic Appraisal Pending",
|
|
type="other",
|
|
institution="Manual",
|
|
liquidity_bucket="real_estate",
|
|
memo="browser evidence missing last_balance warning",
|
|
db_path=str(db_path),
|
|
)
|
|
service.create_ledger_budget("2026-06", "식비", 20_000, memo="browser evidence budget", db_path=str(db_path))
|
|
|
|
tx_specs = [
|
|
dict(transaction_date="2026-06-25", type="income", amount=2_000_000, merchant="SYNTHETIC_SALARY", category="수입", subcategory="급여", source="manual", memo="browser evidence income"),
|
|
dict(transaction_date="2026-06-26", type="expense", amount=35_000, merchant="SYNTHETIC_DINING", category="식비", subcategory="외식", source="manual", memo="browser evidence expense"),
|
|
dict(transaction_date="2026-06-27", type="expense", amount=18_500, merchant="SYNTHETIC_MART", category="", subcategory="", source="manual", memo="browser evidence uncategorized", raw_summary="synthetic grocery"),
|
|
dict(transaction_date="2026-06-28", type="expense", amount=4_800, merchant="SYNTHETIC_CAFE", category="", subcategory="", source="manual", memo="browser evidence uncategorized", raw_summary="synthetic cafe"),
|
|
dict(transaction_date="2026-06-29", type="transfer", amount=120_000, merchant="SYNTHETIC_TRANSFER", category="이체", subcategory="확인필요", source="kb_bank_pdf", counterparty="SYNTHETIC_TOSS", memo="cross_transfer_review: synthetic browser evidence", raw_summary="synthetic transfer", balance_after=1_250_000),
|
|
]
|
|
transactions = []
|
|
for index, spec in enumerate(tx_specs, start=1):
|
|
result = service.create_ledger_transaction(account_id=account_id, external_id=f"browser-evidence-{index}", db_path=str(db_path), **spec)
|
|
if not result.get("success"):
|
|
raise SystemExit(f"Fixture transaction seed failed: {result}")
|
|
transactions.append(result["transaction"]["id"])
|
|
|
|
return {"account_id": account_id, "transaction_ids": transactions, "db_path": str(db_path)}
|
|
|
|
|
|
def wait_for_http(url: str, timeout_seconds: int = 45) -> None:
|
|
deadline = time.time() + timeout_seconds
|
|
last_error: str | None = None
|
|
while time.time() < deadline:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=2) as response:
|
|
if response.status == 200:
|
|
return
|
|
last_error = f"HTTP {response.status}"
|
|
except Exception as exc: # noqa: BLE001 - diagnostics returned in manifest
|
|
last_error = str(exc)
|
|
time.sleep(1)
|
|
raise SystemExit(f"Streamlit did not become ready at {url}: {last_error}")
|
|
|
|
|
|
def read_json_url(url: str, timeout: int = 2) -> Any:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
|
|
|
|
def wait_for_devtools(port: int, timeout_seconds: int = 15) -> str:
|
|
deadline = time.time() + timeout_seconds
|
|
last_error: str | None = None
|
|
while time.time() < deadline:
|
|
try:
|
|
targets = read_json_url(f"http://127.0.0.1:{port}/json/list")
|
|
for target in targets:
|
|
if target.get("type") == "page" and target.get("webSocketDebuggerUrl"):
|
|
return target["webSocketDebuggerUrl"]
|
|
last_error = "no page target"
|
|
except Exception as exc: # noqa: BLE001 - Chrome startup diagnostics
|
|
last_error = str(exc)
|
|
time.sleep(0.25)
|
|
raise SystemExit(f"Chrome DevTools endpoint did not become ready on port {port}: {last_error}")
|
|
|
|
|
|
def remove_tree_with_retries(path: Path, attempts: int = 5, delay_seconds: float = 0.5) -> None:
|
|
for attempt in range(1, attempts + 1):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
if not path.exists():
|
|
return
|
|
time.sleep(delay_seconds * attempt)
|
|
if path.exists():
|
|
raise SystemExit(f"temporary Chrome profile cleanup failed: {path}")
|
|
|
|
|
|
class CdpClient:
|
|
def __init__(self, ws: websockets.ClientConnection) -> None:
|
|
self.ws = ws
|
|
self.next_id = 1
|
|
|
|
async def call(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
command_id = self.next_id
|
|
self.next_id += 1
|
|
await self.ws.send(json.dumps({"id": command_id, "method": method, "params": params or {}}))
|
|
while True:
|
|
raw = await self.ws.recv()
|
|
payload = json.loads(raw)
|
|
if payload.get("id") == command_id:
|
|
if "error" in payload:
|
|
raise RuntimeError(f"CDP {method} failed: {payload['error']}")
|
|
return payload.get("result", {})
|
|
|
|
|
|
async def cdp_capture(ws_url: str, url: str, expected_text: str, screenshot: Path, snapshot: Path) -> dict[str, Any]:
|
|
async with websockets.connect(ws_url, max_size=16 * 1024 * 1024) as ws:
|
|
cdp = CdpClient(ws)
|
|
await cdp.call("Page.enable")
|
|
await cdp.call("Runtime.enable")
|
|
await cdp.call(
|
|
"Emulation.setDeviceMetricsOverride",
|
|
{"width": 1440, "height": 1400, "deviceScaleFactor": 1, "mobile": False},
|
|
)
|
|
await cdp.call("Page.navigate", {"url": url})
|
|
|
|
inner_text = ""
|
|
deadline = time.time() + 45
|
|
while time.time() < deadline:
|
|
result = await cdp.call(
|
|
"Runtime.evaluate",
|
|
{"expression": "document.body ? document.body.innerText : ''", "returnByValue": True},
|
|
)
|
|
inner_text = result.get("result", {}).get("value", "") or ""
|
|
if expected_text in inner_text:
|
|
break
|
|
await asyncio.sleep(1)
|
|
if expected_text not in inner_text:
|
|
raise SystemExit(f"Browser page did not render expected text {expected_text!r}; last text starts with {inner_text[:200]!r}")
|
|
|
|
html_result = await cdp.call(
|
|
"Runtime.evaluate",
|
|
{"expression": "document.documentElement.outerHTML", "returnByValue": True},
|
|
)
|
|
snapshot.write_text(html_result.get("result", {}).get("value", ""), encoding="utf-8")
|
|
image_result = await cdp.call("Page.captureScreenshot", {"format": "png", "fromSurface": True, "captureBeyondViewport": True})
|
|
screenshot.write_bytes(base64.b64decode(image_result["data"]))
|
|
return {"inner_text_head": inner_text[:500]}
|
|
|
|
|
|
def capture_page(chrome: str, base_url: str, page: str, output_dir: Path, devtools_port: int) -> dict[str, Any]:
|
|
slug = slugify_page(page)
|
|
url = f"{base_url}/?page={urllib.parse.quote(page)}"
|
|
screenshot = output_dir / f"{slug}.png"
|
|
snapshot = output_dir / f"{slug}.dom.html"
|
|
|
|
profile_dir = output_dir / f"chrome-profile-{slug}"
|
|
chrome_cmd = [
|
|
chrome,
|
|
"--headless=new",
|
|
"--no-sandbox",
|
|
"--disable-dev-shm-usage",
|
|
"--disable-gpu",
|
|
"--disable-software-rasterizer",
|
|
"--hide-scrollbars",
|
|
f"--remote-debugging-port={devtools_port}",
|
|
f"--user-data-dir={profile_dir}",
|
|
"--window-size=1440,1400",
|
|
"about:blank",
|
|
]
|
|
chrome_proc = subprocess.Popen(chrome_cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
try:
|
|
ws_url = wait_for_devtools(devtools_port)
|
|
cdp_info = asyncio.run(cdp_capture(ws_url, url, page, screenshot, snapshot))
|
|
finally:
|
|
if chrome_proc.poll() is None:
|
|
chrome_proc.terminate()
|
|
try:
|
|
chrome_proc.wait(timeout=8)
|
|
except subprocess.TimeoutExpired:
|
|
chrome_proc.kill()
|
|
chrome_proc.wait(timeout=5)
|
|
remove_tree_with_retries(profile_dir)
|
|
|
|
if not screenshot.exists() or screenshot.stat().st_size <= 0:
|
|
raise SystemExit(f"Screenshot was not created for {page}: {screenshot}")
|
|
if page not in snapshot.read_text(encoding="utf-8", errors="ignore"):
|
|
raise SystemExit(f"DOM snapshot does not contain expected page title {page!r}: {snapshot}")
|
|
|
|
return {
|
|
"page": page,
|
|
"url": url,
|
|
"screenshot": str(screenshot),
|
|
"screenshot_bytes": screenshot.stat().st_size,
|
|
"snapshot": str(snapshot),
|
|
"snapshot_bytes": snapshot.stat().st_size,
|
|
"devtools_port": devtools_port,
|
|
"inner_text_head": cdp_info.get("inner_text_head", ""),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
chrome = require_executable(args.chrome, "Chrome")
|
|
streamlit = require_executable(args.streamlit, "Streamlit")
|
|
run_id = datetime.now().strftime("run-%Y%m%d-%H%M%S")
|
|
output_dir = Path(args.output_dir) if args.output_dir else DEFAULT_OUTPUT_ROOT / run_id
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
db_path = output_dir / "fixture.sqlite3"
|
|
fixture = seed_fixture_db(db_path)
|
|
|
|
env = os.environ.copy()
|
|
env["JARCA_LEDGER_DB_PATH"] = str(db_path.resolve())
|
|
env["STREAMLIT_BROWSER_GATHER_USAGE_STATS"] = "false"
|
|
base_url = f"http://{args.host}:{args.port}"
|
|
server_cmd = [
|
|
streamlit,
|
|
"run",
|
|
"dashboard/app.py",
|
|
"--server.address",
|
|
args.host,
|
|
"--server.port",
|
|
str(args.port),
|
|
"--server.headless",
|
|
"true",
|
|
"--browser.gatherUsageStats",
|
|
"false",
|
|
]
|
|
|
|
server = subprocess.Popen(server_cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, start_new_session=True)
|
|
manifest: dict[str, Any] = {
|
|
"success": False,
|
|
"base_url": base_url,
|
|
"output_dir": str(output_dir.resolve()),
|
|
"fixture": fixture,
|
|
"chrome": chrome,
|
|
"streamlit": streamlit,
|
|
"pages": [],
|
|
"server_command": server_cmd,
|
|
}
|
|
try:
|
|
wait_for_http(base_url)
|
|
for index, page in enumerate(args.pages or DEFAULT_PAGES):
|
|
manifest["pages"].append(capture_page(chrome, base_url, page, output_dir, devtools_port=19222 + index))
|
|
manifest["success"] = True
|
|
print(json.dumps(manifest, ensure_ascii=False, indent=2))
|
|
return 0
|
|
finally:
|
|
log_tail = ""
|
|
if server.stdout is not None:
|
|
try:
|
|
time.sleep(0.5)
|
|
if server.poll() is not None:
|
|
log_tail = server.stdout.read()[-4000:]
|
|
except Exception:
|
|
log_tail = ""
|
|
manifest["server_log_tail"] = log_tail
|
|
(output_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
if not args.keep_server and server.poll() is None:
|
|
os.killpg(server.pid, signal.SIGTERM)
|
|
try:
|
|
server.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(server.pid, signal.SIGKILL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|