"""运行报告：单次 :func:`pyflowx.run` 的类型化、可查询结果。
报告是执行后的唯一事实来源。它通过 ``report["name"]`` 暴露单任务结果
（类型为 ``Any``，因为映射异构）、汇总统计，以及整次运行是否成功的标志。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Iterator
from .task import TaskResult, TaskStatus
[文档]
@dataclass
class RunReport:
"""工作流运行的聚合结果。
属性
----
results:
任务名 -> :class:`TaskResult` 的映射。插入顺序与任务完成顺序一致。
success:
当且仅当所有非跳过任务都以 ``SUCCESS`` 结束时为 ``True``。
"""
results: dict[str, TaskResult[Any]] = field(default_factory=dict)
success: bool = True
# ---- 类型化访问 --------------------------------------------------- #
[文档]
def __getitem__(self, name: str) -> Any:
"""返回任务 ``name`` 的*值*（而非 TaskResult）。
任务不在本次运行中则抛出 ``KeyError``。未达到 SUCCESS 的任务
返回 ``None``。
"""
return self.results[name].value
[文档]
def result_of(self, name: str) -> TaskResult[Any]:
"""返回 ``name`` 的完整 :class:`TaskResult`。"""
return self.results[name]
def __contains__(self, name: Any) -> bool:
return name in self.results
def __iter__(self) -> Iterator[str]:
return iter(self.results)
def __len__(self) -> int:
return len(self.results)
# ---- 汇总 --------------------------------------------------------- #
[文档]
def summary(self) -> dict[str, Any]:
"""用于日志/仪表盘的紧凑统计字典。"""
counts: dict[str, int] = {}
total_duration = 0.0
for r in self.results.values():
counts[r.status.value] = counts.get(r.status.value, 0) + 1
if r.duration is not None:
total_duration += r.duration
return {
"success": self.success,
"total_tasks": len(self.results),
"by_status": counts,
"total_duration_seconds": round(total_duration, 6),
}
[文档]
def failed_tasks(self) -> list[str]:
"""以 FAILED 状态结束的任务名列表。"""
return [name for name, r in self.results.items() if r.status == TaskStatus.FAILED]
[文档]
def succeeded_tasks(self) -> list[str]:
"""以 SUCCESS 状态结束的任务名列表。"""
return [name for name, r in self.results.items() if r.status == TaskStatus.SUCCESS]
[文档]
def skipped_tasks(self) -> list[str]:
"""以 SKIPPED 状态结束的任务名列表。"""
return [name for name, r in self.results.items() if r.status == TaskStatus.SKIPPED]
[文档]
def tasks_by_status(self, status: TaskStatus) -> list[str]:
"""返回指定状态的任务名列表。"""
return [name for name, r in self.results.items() if r.status == status]
[文档]
def durations(self) -> dict[str, float]:
"""任务名 -> 执行时长（秒）。无时长记录的为 0.0。"""
return {name: (r.duration or 0.0) for name, r in self.results.items()}
[文档]
def describe(self) -> str:
"""用于调试的人类可读多行报告。"""
lines: list[str] = [f"RunReport(success={self.success})"]
for name, r in self.results.items():
dur = f"{r.duration:.3f}s" if r.duration is not None else "-"
err = f" error={r.error!r}" if r.error else ""
lines.append(f" {name}: {r.status.value} ({dur} attempts={r.attempts}){err}")
return "\n".join(lines)