"""证据落盘:HTML 快照、截图与 meta.json
规格 §6.1:每个任务一个子目录,按步骤序号命名文件。report 里的 evidence_ref
填 `{task_id}/{序号}-{步骤名}`,不含扩展名。
证据**只留本地**(约束 4),网关只存 evidence_ref 这个相对路径字符串。
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
class EvidenceStore:
"""把每一步的页面快照、截图与 meta 落到 {evidence_dir}/{task_id}/ 下"""
def __init__(self, evidence_dir: Path):
self._root = evidence_dir
def step_dir(self, task_id: str) -> Path:
path = self._root / task_id
path.mkdir(parents=True, exist_ok=True)
return path
def write_step(
self,
task_id: str,
step_no: int,
step_name: str,
*,
html: str | None = None,
png: bytes | None = None,
meta: dict[str, Any] | None = None,
) -> str:
"""写入一步的证据文件
返回相对 evidence_dir 的引用路径(不含扩展名),用于 report 的 evidence_ref
字段。文件命名:`{序号:02d}-{步骤名}.{html|png|meta.json}`。
"""
step_dir = self.step_dir(task_id)
stem = f"{step_no:02d}-{step_name}"
rel_stem = f"{task_id}/{stem}"
if html is not None:
(step_dir / f"{stem}.html").write_text(html, encoding="utf-8")
if png is not None:
(step_dir / f"{stem}.png").write_bytes(png)
if meta is not None:
(step_dir / f"{stem}.meta.json").write_text(
json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8"
)
return rel_stem