"""汇总与输出：统计（成功率、p50/p95）、rich 表格、JSON/CSV 原子写。（参考答案）
原子写的意思：先写同目录的临时文件，再 os.replace(临时, 目标)——
中途断电/报错也不会留下半个文件。
"""
import csv
import json
import math
import os
from collections import Counter
from collections.abc import Sequence
from pathlib import Path
from rich.table import Table
from aggregator.models import FetchResult, Summary
def percentile(values: Sequence[float], pct: float) -> float:
"""取百分位（最近秩法：向上取整索引），空序列返回 0.0。
pct 不在 0..100 抛 ValueError("pct 必须在 0..100 之间")。
例：percentile([1, 2, 3, 4], 50) -> 2.0；percentile([], 95) -> 0.0
"""
if not 0 <= pct <= 100:
raise ValueError("pct 必须在 0..100 之间")
if not values:
return 0.0
ordered = sorted(values)
rank = max(1, math.ceil(pct / 100 * len(ordered)))
return float(ordered[rank - 1])
def summarize(results: Sequence[FetchResult], total_elapsed_ms: float = 0.0) -> Summary:
"""把一批结果汇总成 Summary：成功数、失败数、成功率、p50/p95、错误分类计数。
success_rate 用 0..1 的小数（空结果为 0.0）；
errors 是 {错误标签: 次数}，只统计失败的那些。
例：3 成功 1 失败 -> Summary(total=4, ok=3, success_rate=0.75, ...)
"""
ok = [r for r in results if r.ok]
failed = [r for r in results if not r.ok]
latencies = [r.elapsed_ms for r in results]
return Summary(
total=len(results),
ok=len(ok),
failed=len(failed),
success_rate=(len(ok) / len(results)) if results else 0.0,
p50_ms=percentile(latencies, 50),
p95_ms=percentile(latencies, 95),
total_elapsed_ms=total_elapsed_ms,
errors=dict(Counter(r.error or "unknown" for r in failed)),
)
def results_table(results: Sequence[FetchResult], limit: int = 10) -> Table:
"""做一张 rich 表格：列 url / status / elapsed_ms / error，最多显示 limit 行。
例：results_table(results, limit=5).row_count -> 5
"""
table = Table(
title=f"抓取结果（前 {min(limit, len(results))} / {len(results)} 条）"
)
for column in ("url", "status", "elapsed_ms", "error"):
table.add_column(column, justify="right" if column != "url" else "left")
for result in list(results)[:limit]:
row = result.row()
table.add_row(row["url"], row["status"], row["elapsed_ms"], row["error"])
return table
def summary_table(summary: Summary) -> Table:
"""做一张两列（指标 / 值）的 rich 表格，展示 Summary 的所有字段。"""
table = Table(title="汇总")
table.add_column("指标")
table.add_column("值", justify="right")
for key, value in summary.model_dump().items():
if isinstance(value, float):
text = f"{value:.2%}" if key == "success_rate" else f"{value:.1f}"
elif isinstance(value, dict):
text = ", ".join(f"{k}×{v}" for k, v in value.items()) or "-"
else:
text = str(value)
table.add_row(key, text)
return table
def _atomic_write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(text, encoding="utf-8", newline="")
os.replace(tmp, path)
def write_json(path: Path, summary: Summary, results: Sequence[FetchResult]) -> None:
"""原子写 JSON：{"summary": {...}, "results": [...]}。
提示：summary.model_dump()、[r.model_dump() for r in results]；
先写 path.with_suffix(path.suffix + ".tmp")，再 os.replace 换上去。
"""
payload = {
"summary": summary.model_dump(),
"results": [r.model_dump() for r in results],
}
_atomic_write_text(path, json.dumps(payload, ensure_ascii=False, indent=2))
def write_csv(path: Path, results: Sequence[FetchResult]) -> None:
"""原子写 CSV：表头 url,status,elapsed_ms,error（用 FetchResult.row()）。
注意 newline=""，否则 Windows 上每行之间会多一个空行。
"""
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
with tmp.open("w", encoding="utf-8", newline="") as fp:
writer = csv.DictWriter(fp, fieldnames=["url", "status", "elapsed_ms", "error"])
writer.writeheader()
writer.writerows(r.row() for r in results)
os.replace(tmp, path)