AutoRpa/rpa_contract/accept.py

# 页面验收报告。每步写一行 ACCEPT，汇总进 ACCEPT.txt。
#
# 验收看这份文本，不看截图。截图只在失败时留，给人查页面。

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]


# 把流程名收成目录名。空名用 default，其余只留字母、数字、横线和底线。
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)


# 这个流程的截图和 ACCEPT.txt 放在哪。
def artifact_dir(flow: str) -> Path:
    return SCREENSHOT_ROOT / safe_flow_name(flow)


# 默认只在失败时截图。RPA_SCREENSHOT=always 时每步都留。
def evidence_shot(browser: HasScreenshot, name: str, *, failed: bool = False) -> str:
    mode = (os.environ.get("RPA_SCREENSHOT") or "fail").strip().lower()
    if not failed and mode != "always":
        return ""
    return browser.screenshot(name)


# ACCEPT.txt 的路径。RPA_ACCEPT_LOG 可改到别处，单测用这个钉住文件。
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


# 把一步收成控制台一行：ACCEPT、步骤、ok 或 FAIL、标题和读到的字段。
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()


# 一次页面流的验收单。每步打印并覆写 ACCEPT.txt，对不上就抛 NodeError。
@dataclass
class AcceptanceReport:
    flow: str = "default"
    lines: list[AcceptLine] = field(default_factory=list)

    # 清空已记的步骤，下一次页面流重新计。
    def reset(self) -> None:
        self.lines = []

    # 核对一组字段。对不上就记 FAIL、失败时截图，并抛 NodeError 停下这一步。
    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))

    # 记一步事实，不和期望逐字段比。ok 为假且 raise_error 时抛 NodeError。
    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()))

    # 把当前汇总写进 ACCEPT.txt。写失败只打印，不把页面流打成另一次错误。
    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)

    # 汇总文本：第一行是通过数，后面是每步的 ACCEPT 行。
    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"