154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
"""Shared ACCEPT report for every RPA flow (any platform / page).
|
|
|
|
New flows (ORDER_FETCH, REBATE_*, M5, ...) must use this instead of
|
|
reading screenshots. Screenshots stay on disk for humans; this text is
|
|
the agent oracle.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
|
|
from errors import NodeError
|
|
|
|
PACKAGE_ROOT = Path(__file__).resolve().parent
|
|
SCREENSHOT_ROOT = PACKAGE_ROOT / "screenshots"
|
|
|
|
|
|
class HasScreenshot(Protocol):
|
|
def screenshot(self, name: str) -> str: ...
|
|
|
|
|
|
@dataclass
|
|
class AcceptLine:
|
|
step: str
|
|
title: str
|
|
ok: bool
|
|
facts: dict[str, str]
|
|
mismatches: list[str]
|
|
|
|
|
|
def safe_flow_name(flow: str) -> str:
|
|
text = (flow or "default").strip() or "default"
|
|
return "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in text)
|
|
|
|
|
|
def artifact_dir(flow: str) -> Path:
|
|
return SCREENSHOT_ROOT / safe_flow_name(flow)
|
|
|
|
|
|
def evidence_shot(browser: HasScreenshot, name: str, *, failed: bool = False) -> str:
|
|
"""Save a screenshot only on failure. Set RPA_SCREENSHOT=always to keep every frame."""
|
|
mode = (os.environ.get("RPA_SCREENSHOT") or "fail").strip().lower()
|
|
if not failed and mode != "always":
|
|
return ""
|
|
return browser.screenshot(name)
|
|
|
|
|
|
def accept_log_path(flow: str = "default") -> Path:
|
|
override = os.environ.get("RPA_ACCEPT_LOG")
|
|
if override:
|
|
return Path(override)
|
|
return artifact_dir(flow) / "ACCEPT.txt"
|
|
|
|
|
|
def field_match(actual: str, expected: str) -> bool:
|
|
actual_text = (actual or "").strip()
|
|
expected_text = (expected or "").strip()
|
|
if not expected_text:
|
|
return True
|
|
if actual_text == expected_text:
|
|
return True
|
|
compact_a = actual_text.replace(" ", "").replace(",", "")
|
|
compact_e = expected_text.replace(" ", "").replace(",", "")
|
|
if compact_a == compact_e or compact_a.replace("-", "") == compact_e.replace("-", ""):
|
|
return True
|
|
try:
|
|
return float(compact_a) == float(compact_e)
|
|
except ValueError:
|
|
pass
|
|
yes = {"是", "1", "Y", "YES", "TRUE"}
|
|
no = {"否", "2", "0", "N", "NO", "FALSE"}
|
|
if compact_a.upper() in yes and compact_e.upper() in yes:
|
|
return True
|
|
if compact_a.upper() in no and compact_e.upper() in no:
|
|
return True
|
|
return compact_e in compact_a or compact_a in compact_e
|
|
|
|
|
|
def format_accept(line: AcceptLine) -> str:
|
|
status = "ok" if line.ok else "FAIL"
|
|
facts = " ".join(f"{key}={value}" for key, value in line.facts.items())
|
|
extra = f" {' | '.join(line.mismatches)}" if line.mismatches else ""
|
|
return f"ACCEPT {line.step} {status} {line.title} {facts}{extra}".rstrip()
|
|
|
|
|
|
@dataclass
|
|
class AcceptanceReport:
|
|
"""DOM / page-state checklist. One instance per flow run."""
|
|
|
|
flow: str = "default"
|
|
lines: list[AcceptLine] = field(default_factory=list)
|
|
|
|
def reset(self) -> None:
|
|
self.lines = []
|
|
|
|
def check(
|
|
self,
|
|
step: str,
|
|
title: str,
|
|
pairs: list[tuple[str, object, object]],
|
|
*,
|
|
browser: HasScreenshot | None = None,
|
|
screenshot_on_fail: str = "",
|
|
) -> None:
|
|
facts: dict[str, str] = {}
|
|
mismatches: list[str] = []
|
|
for name, actual, expected in pairs:
|
|
actual_text = "" if actual is None else str(actual).strip()
|
|
expected_text = "" if expected is None else str(expected).strip()
|
|
facts[name] = actual_text or "(empty)"
|
|
if expected_text and not field_match(actual_text, expected_text):
|
|
mismatches.append(f"{name} expected={expected_text!r} actual={actual_text!r}")
|
|
ok = not mismatches
|
|
self.lines.append(AcceptLine(step, title, ok, facts, mismatches))
|
|
print(format_accept(self.lines[-1]), flush=True)
|
|
self.dump()
|
|
if ok:
|
|
return
|
|
if browser is not None and screenshot_on_fail:
|
|
evidence_shot(browser, screenshot_on_fail, failed=True)
|
|
raise NodeError(f"ACCEPT {step} FAIL " + "; ".join(mismatches))
|
|
|
|
def note(
|
|
self,
|
|
step: str,
|
|
title: str,
|
|
facts: dict[str, object],
|
|
*,
|
|
ok: bool = True,
|
|
raise_error: bool = True,
|
|
) -> None:
|
|
rendered = {key: ("" if value is None else str(value).strip()) or "(empty)" for key, value in facts.items()}
|
|
self.lines.append(AcceptLine(step, title, ok, rendered, [] if ok else ["failed"]))
|
|
print(format_accept(self.lines[-1]), flush=True)
|
|
self.dump()
|
|
if not ok and raise_error:
|
|
raise NodeError(f"ACCEPT {step} FAIL {title} " + " ".join(f"{k}={v}" for k, v in rendered.items()))
|
|
|
|
def dump(self) -> None:
|
|
path = accept_log_path(self.flow)
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(self.summary_text(), encoding="utf-8")
|
|
except OSError as exc:
|
|
print(f"ACCEPT log write skipped: {exc}", flush=True)
|
|
|
|
def summary_text(self) -> str:
|
|
passed = sum(1 for line in self.lines if line.ok)
|
|
rows = [f"ACCEPT SUMMARY {self.flow} {passed}/{len(self.lines)} ok"]
|
|
rows.extend(format_accept(line) for line in self.lines)
|
|
return "\n".join(rows) + "\n"
|