"""Structured report for intelligent evaluation (票据 04). Two concerns live here so they stay co-located: * **Validation** — pydantic models describing the report contract OpenClaw submits. A malformed body yields a 422 before any state change happens. * **Rendering** — ``render_report_markdown`` is a pure function (dict in, string out) with no DB or I/O, so it unit-tests trivially. """ from typing import Any, Optional from pydantic import BaseModel, Field, model_validator class ReportEvidence(BaseModel): session_id: str = "" turn_index: Optional[int] = None user_said: str = "" assistant_replied: str = "" class ReportFinding(BaseModel): issue: str = Field(min_length=1) severity: str = Field(min_length=1) dimension: str = Field(min_length=1) evidence: list[ReportEvidence] = Field(default_factory=list) suggestion: Optional[str] = None related_sop: Optional[str] = None class ReportHighlight(BaseModel): description: str = "" dimension: Optional[str] = None class ReportModel(BaseModel): summary: str = Field(min_length=1) scores: Optional[dict[str, Any]] = None findings: list[ReportFinding] = Field(default_factory=list) highlights: list[ReportHighlight] = Field(default_factory=list) priority_recommendations: list[str] = Field(default_factory=list) @model_validator(mode="after") def _require_findings(self) -> "ReportModel": if not self.findings: raise ValueError("findings 不能为空") return self _SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2} def _severity_rank(severity: str) -> int: return _SEVERITY_ORDER.get((severity or "").lower(), 99) def render_report_markdown(report: dict[str, Any], *, name: str = "", eval_id: str = "") -> str: """Render a validated report dict to Markdown. Pure function, no I/O.""" validated = ReportModel.model_validate(report) lines: list[str] = [] title = name or "智能评估报告" lines.append(f"# {title}") lines.append("") if eval_id: lines.append(f"> 评估 ID:`{eval_id}`") lines.append("") lines.append("## 总体概述") lines.append("") lines.append(validated.summary.strip()) lines.append("") if validated.scores: lines.append("## 维度评分") lines.append("") lines.append("| 维度 | 分数 |") lines.append("| --- | --- |") for dimension, score in validated.scores.items(): lines.append(f"| {dimension} | {score} |") lines.append("") lines.append("## 问题发现") lines.append("") findings = sorted(validated.findings, key=lambda f: _severity_rank(f.severity)) for index, finding in enumerate(findings, start=1): lines.append(f"### {index}. {finding.issue}") lines.append("") lines.append(f"- **严重程度**:{finding.severity}") lines.append(f"- **维度**:{finding.dimension}") if finding.suggestion: lines.append(f"- **建议**:{finding.suggestion}") if finding.related_sop: lines.append(f"- **关联 SOP**:{finding.related_sop}") if finding.evidence: lines.append("") lines.append("**证据**:") lines.append("") for ev in finding.evidence: header_bits = [] if ev.session_id: header_bits.append(f"会话 `{ev.session_id}`") if ev.turn_index is not None: header_bits.append(f"第 {ev.turn_index} 轮") header = "(" + ",".join(header_bits) + ")" if header_bits else "" lines.append(f"> {header}") if ev.user_said: lines.append(f"> **用户**:{ev.user_said}") if ev.assistant_replied: lines.append(f"> **对象**:{ev.assistant_replied}") lines.append("") lines.append("") if validated.highlights: lines.append("## 亮点") lines.append("") for highlight in validated.highlights: suffix = f"({highlight.dimension})" if highlight.dimension else "" lines.append(f"- {highlight.description}{suffix}") lines.append("") if validated.priority_recommendations: lines.append("## 优先改进建议") lines.append("") for recommendation in validated.priority_recommendations: lines.append(f"- {recommendation}") lines.append("") return "\n".join(lines).rstrip() + "\n"