Spaces:

Kevinshh
/

Preformu

Running

Preformu / core /services /visualization.py
feat: 意图保真(intent-fidelity) + 描述性梳理技能 + 相容性引擎升级; 修复转置宽表解析/CQA对账/澄清交互/功能切换串显; .gitignore 排除专利与机密Demo数据
0e6887b 4 months ago
28.2 kB
"""自适应报告可视化：数据形态驱动的图表推荐引擎（adaptive-report-visualization）。
本模块是**纯 Python、确定性、不接收 svc/LLM 句柄、不 import matplotlib/numpy** 的底座
组件。它把 ``compute`` 阶段产出的确定性结构化结果（``ComputeResult.summary`` /
``figures``）映射为有序的图表 / 表格方案（:class:`VisualizationPlan`），交由既有
``ChartService`` 渲染、``ReportService`` / ``docx_export`` 注入。
合规底线（与 spec requirements 一致）：
- **数值只来自 compute**：推荐引擎只在既有数值之上做"形态判别 → 选图 → 排序 → 预算"，
  绝不选数 / 造数 / 改数；参考线只用 compute 解析出的限度 / AV / 货架期数值。
- **可降级、可测试**：不依赖绘图库即可完成形态分类 / 选图 / 排序，可被纯逻辑单测覆盖。
- **确定性 / 幂等 / 可序列化**：同输入产出等价方案；重复排序结果不变；``to_dict`` /
  ``from_dict`` 往返一致。
本文件实现任务 1 的数据模型与序列化；分类器 / 适配器 / 推荐编排在后续任务补齐。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional


# ---------------------------------------------------------------------------
# 枚举
# ---------------------------------------------------------------------------

class ShapeSignature(str, Enum):
    """数据形态签名（离散标签）。"""

    TIME_SERIES = "time_series"
    GROUPED_SINGLE_METRIC = "grouped_single_metric"
    REPEATED_UNIT_DISTRIBUTION = "repeated_unit_distribution"
    CROSS_ATTRIBUTE_COMPLIANCE = "cross_attribute_compliance"
    BIVARIATE = "bivariate"
    SINGLE_VALUE_VS_LIMIT = "single_value_vs_limit"
    UNKNOWN = "unknown"


class ChartType(str, Enum):
    """图型。"""

    TREND_BAND = "trend_band"
    GROUPED_BAR = "grouped_bar"
    GROUPED_DOT = "grouped_dot"
    DISTRIBUTION_DOT = "distribution_dot"
    BOX_PLOT = "box_plot"
    STATUS_MATRIX = "status_matrix"
    SCATTER = "scatter"
    BULLET = "bullet"
    TABLE = "table"  # 降级表格


def _coerce_enum(enum_cls, value, default):
    if isinstance(value, enum_cls):
        return value
    try:
        return enum_cls(str(value))
    except (ValueError, TypeError):
        return default


# ---------------------------------------------------------------------------
# DataBlock：各 Skill summary 归一后的中间块（分类器只读它）
# ---------------------------------------------------------------------------

@dataclass
class DataBlock:
    """SummaryAdapter 把各 Skill 的 summary 归一成的中间块。
    其中所有数值字段（``times`` / ``series`` / ``groups`` / ``units`` / 限度）均为对
    ``ComputeResult`` 既有数值的**逐字拷贝**，不新增、不取整、不插值（需求 17.3）。
    """

    block_id: str
    title: str = ""
    source_table: str = ""
    times: Optional[list] = None              # time_series：时间点
    series: Optional[list] = None             # [{label, values, ...}]
    groups: Optional[list] = None             # [{label, value, within_spec, risk, limit}]
    units: Optional[list] = None              # repeated_unit_distribution：逐单位值
    limit_low: Optional[float] = None
    limit_high: Optional[float] = None
    limit_text: str = ""
    acceptance_value: Optional[dict] = None   # 含量均匀度 AV 信息
    attribute: str = ""
    intent_hint: str = ""

    def to_dict(self) -> dict:
        return {
            "block_id": self.block_id,
            "title": self.title,
            "source_table": self.source_table,
            "times": list(self.times) if self.times is not None else None,
            "series": [dict(s) for s in self.series] if self.series is not None else None,
            "groups": [dict(g) for g in self.groups] if self.groups is not None else None,
            "units": list(self.units) if self.units is not None else None,
            "limit_low": self.limit_low,
            "limit_high": self.limit_high,
            "limit_text": self.limit_text,
            "acceptance_value": dict(self.acceptance_value) if self.acceptance_value else None,
            "attribute": self.attribute,
            "intent_hint": self.intent_hint,
        }

    @staticmethod
    def from_dict(d: dict) -> "DataBlock":
        d = d or {}
        return DataBlock(
            block_id=str(d.get("block_id", "")),
            title=str(d.get("title", "")),
            source_table=str(d.get("source_table", "")),
            times=list(d["times"]) if d.get("times") is not None else None,
            series=[dict(s) for s in d["series"]] if d.get("series") is not None else None,
            groups=[dict(g) for g in d["groups"]] if d.get("groups") is not None else None,
            units=list(d["units"]) if d.get("units") is not None else None,
            limit_low=d.get("limit_low"),
            limit_high=d.get("limit_high"),
            limit_text=str(d.get("limit_text", "")),
            acceptance_value=dict(d["acceptance_value"]) if d.get("acceptance_value") else None,
            attribute=str(d.get("attribute", "")),
            intent_hint=str(d.get("intent_hint", "")),
        )


# ---------------------------------------------------------------------------
# ChartSpec：待渲染图表的确定性规格（不含 LLM 生成的数值）
# ---------------------------------------------------------------------------

@dataclass
class ChartSpec:
    chart_type: ChartType
    block_id: str                                  # 引用 DataBlock / figures key
    title: str = ""
    caption_key: str = ""                          # 图注（文字，渲染时填充）
    reference_lines: list = field(default_factory=list)   # [{value,label,kind}]，仅来自 compute
    risk_colored: bool = False
    redundant_marks: bool = True                   # 无障碍：状态附符号/文字（需求 15）
    companion_table_block_id: str = ""             # 配套源数据表（可审计，需求 14）
    aggregated_from: list = field(default_factory=list)   # 聚合自哪些分组（需求 16）
    degraded_from: Optional[ChartType] = None
    degrade_reason: str = ""

    def to_dict(self) -> dict:
        return {
            "chart_type": self.chart_type.value,
            "block_id": self.block_id,
            "title": self.title,
            "caption_key": self.caption_key,
            "reference_lines": [dict(r) for r in self.reference_lines],
            "risk_colored": bool(self.risk_colored),
            "redundant_marks": bool(self.redundant_marks),
            "companion_table_block_id": self.companion_table_block_id,
            "aggregated_from": list(self.aggregated_from),
            "degraded_from": self.degraded_from.value if self.degraded_from else None,
            "degrade_reason": self.degrade_reason,
        }

    @staticmethod
    def from_dict(d: dict) -> "ChartSpec":
        d = d or {}
        df = d.get("degraded_from")
        return ChartSpec(
            chart_type=_coerce_enum(ChartType, d.get("chart_type"), ChartType.TABLE),
            block_id=str(d.get("block_id", "")),
            title=str(d.get("title", "")),
            caption_key=str(d.get("caption_key", "")),
            reference_lines=[dict(r) for r in (d.get("reference_lines") or [])],
            risk_colored=bool(d.get("risk_colored", False)),
            redundant_marks=bool(d.get("redundant_marks", True)),
            companion_table_block_id=str(d.get("companion_table_block_id", "")),
            aggregated_from=list(d.get("aggregated_from") or []),
            degraded_from=_coerce_enum(ChartType, df, None) if df else None,
            degrade_reason=str(d.get("degrade_reason", "")),
        )


# ---------------------------------------------------------------------------
# VisualizationPlan：有序 ChartSpec 列表 + 降级/忽略记录
# ---------------------------------------------------------------------------

@dataclass
class VisualizationPlan:
    specs: list = field(default_factory=list)   # list[ChartSpec]，有序
    notes: list = field(default_factory=list)   # 降级/忽略原因（可审计）

    def to_dict(self) -> dict:
        return {
            "specs": [s.to_dict() for s in self.specs],
            "notes": list(self.notes),
        }

    @staticmethod
    def from_dict(d: dict) -> "VisualizationPlan":
        d = d or {}
        return VisualizationPlan(
            specs=[ChartSpec.from_dict(s) for s in (d.get("specs") or [])],
            notes=[str(n) for n in (d.get("notes") or [])],
        )


__all__ = [
    "ShapeSignature",
    "ChartType",
    "DataBlock",
    "ChartSpec",
    "VisualizationPlan",
    "build_data_blocks",
    "classify_block",
    "map_block_to_spec",
    "MIN_TREND_POINTS",
    "MIN_GROUPS",
    "MIN_UNITS",
    "VisualizationRecommender",
    "figures_from_plan",
]


# ===========================================================================
# 任务 2：SummaryAdapter —— summary → list[DataBlock]
# ===========================================================================

def _is_number(v: Any) -> bool:
    return isinstance(v, (int, float)) and not isinstance(v, bool)


def _status_of(within_spec: Optional[bool]) -> str:
    """合规布尔 → 三态标签（用于状态矩阵的冗余编码）。"""
    if within_spec is True:
        return "pass"
    if within_spec is False:
        return "fail"
    return "na"


def _descriptive_blocks(summary: dict) -> list["DataBlock"]:
    """从 descriptive_summary 的 ``groups`` 产出 DataBlock（逐字拷贝数值）。
    产出三类：
    - 含量均匀度等"重复单位分布"：单组 values≥3 且带 acceptance_value。
    - "跨属性合规"：汇总所有 within_spec 非空的分组为一个状态矩阵块。
    - "分组×单指标"：同一属性跨多分组（规格/批次）的均值聚合为一张多分组图（需求 16.2）。
    """
    groups = summary.get("groups") or []
    blocks: list[DataBlock] = []

    # 1) 重复单位分布（逐单位值）。
    for g in groups:
        values = g.get("values") or []
        av = g.get("acceptance_value")
        numeric = [v for v in values if _is_number(v)]
        if av and len(numeric) >= 3:
            strength = str(g.get("strength", "") or "")
            attr = str(g.get("attribute", "") or "")
            blocks.append(DataBlock(
                block_id=f"dist::{attr}::{strength}::{g.get('table','')}",
                title=f"{attr} {strength}".strip(),
                source_table=str(g.get("table", "") or ""),
                units=list(numeric),                       # 逐字拷贝
                acceptance_value=dict(av),
                attribute=attr,
            ))

    # 2) 跨属性合规状态矩阵（汇总所有有判定的分组）。
    compliance_rows = []
    for g in groups:
        ws = g.get("within_spec")
        if ws is None:
            continue
        label = f"{g.get('attribute','')} {g.get('strength','')}".strip()
        compliance_rows.append({
            "label": label,
            "attribute": str(g.get("attribute", "") or ""),
            "strength": str(g.get("strength", "") or ""),
            "within_spec": ws,
            "status": _status_of(ws),
            "spec_limit": g.get("spec_limit"),
        })
    if compliance_rows:
        blocks.append(DataBlock(
            block_id="compliance::all",
            title="限度符合性",
            groups=compliance_rows,
        ))

    # 3) 分组×单指标：同属性跨多分组的均值聚合（聚合优先，需求 16.2）。
    by_attr: dict[str, list[dict]] = {}
    for g in groups:
        if g.get("acceptance_value"):
            continue  # CU 已单独成图
        mean = g.get("mean")
        if not _is_number(mean):
            continue
        attr = str(g.get("attribute", "") or "")
        if not attr:
            continue
        by_attr.setdefault(attr, []).append({
            "label": str(g.get("strength", "") or g.get("batch", "") or attr),
            "value": mean,                                 # 逐字拷贝
            "within_spec": g.get("within_spec"),
            "spec_limit": g.get("spec_limit"),
        })
    for attr, rows in by_attr.items():
        if len(rows) < 2:
            continue  # 单分组无可比性，交由门槛/降级处理
        blocks.append(DataBlock(
            block_id=f"grouped::{attr}",
            title=attr,
            attribute=attr,
            groups=rows,
        ))

    return blocks


def _generic_timeseries_blocks(summary: dict) -> list["DataBlock"]:
    """从通用 ``data_overview.rows``（稳定性等）产出时序 DataBlock。
    每行形如 ``{batch, condition, cqa, timepoints:[...], values:[...]}``。
    """
    overview = summary.get("data_overview") or {}
    rows = overview.get("rows") or []
    blocks: list[DataBlock] = []
    for i, r in enumerate(rows):
        tps = r.get("timepoints") or []
        vals = r.get("values") or []
        if len(tps) < 2 or len(vals) != len(tps):
            continue
        label = f"{r.get('batch','')} {r.get('condition','')} {r.get('cqa','')}".strip()
        blocks.append(DataBlock(
            block_id=f"ts::{i}::{label}",
            title=label,
            times=list(tps),                               # 逐字拷贝
            series=[{"label": str(r.get("cqa", "") or label), "values": list(vals)}],
            attribute=str(r.get("cqa", "") or ""),
        ))
    return blocks


def build_data_blocks(summary: dict, *, skill_id: str = "", skill: Any = None) -> list["DataBlock"]:
    """把任意 Skill 的 ``summary`` 归一为 DataBlock 列表（SummaryAdapter 入口）。
    优先级：
    1. Skill 自定义钩子（鸭子类型）：``skill.viz_data_blocks(summary)`` → list[DataBlock]
       （需求 10.1/10.2）。无钩子则用底座默认适配。
    2. 底座默认：descriptive_summary 的 ``groups`` + 通用 ``data_overview`` 时序。
    所有数值为对 summary 的逐字拷贝，绝不取整/插值/换算（需求 17.3）。
    """
    summary = summary or {}
    hook = getattr(skill, "viz_data_blocks", None)
    if callable(hook):
        try:
            out = hook(summary)
            if isinstance(out, list) and all(isinstance(b, DataBlock) for b in out):
                return out
        except Exception:  # noqa: BLE001 - 钩子异常回退底座默认，不崩溃
            pass

    blocks: list[DataBlock] = []
    if summary.get("groups"):
        blocks.extend(_descriptive_blocks(summary))
    blocks.extend(_generic_timeseries_blocks(summary))
    return blocks


# ===========================================================================
# 任务 3：Data_Shape_Classifier —— DataBlock → ShapeSignature
# ===========================================================================

def classify_block(block: "DataBlock") -> "ShapeSignature":
    """按字段存在性与基数判定数据形态签名（确定性，见设计 §3 判据表）。"""
    # 1) 时序：times 非空且某 series 数值与 times 等长、点数≥2。
    times = block.times or []
    if len(times) >= 2 and block.series:
        for s in block.series:
            vals = s.get("values") or []
            if len(vals) == len(times):
                return ShapeSignature.TIME_SERIES

    # 2) 重复单位分布：units 非空、单位数≥3。
    if block.units and len([v for v in block.units if _is_number(v)]) >= 3:
        return ShapeSignature.REPEATED_UNIT_DISTRIBUTION

    groups = block.groups or []
    if groups:
        # 3) 跨属性合规：每项带 status/within_spec、且≥2 项（状态语义优先于数值）。
        has_status = all(("status" in g or "within_spec" in g) for g in groups)
        has_value = all(_is_number(g.get("value")) for g in groups)
        if has_status and not has_value and len(groups) >= 1:
            return ShapeSignature.CROSS_ATTRIBUTE_COMPLIANCE
        # 4) 分组×单指标：每项含数值 value、分组数≥2。
        if has_value and len(groups) >= 2:
            return ShapeSignature.GROUPED_SINGLE_METRIC

    # 5) 双变量：两 series 成对 (x, y)。
    if block.series and len(block.series) == 2:
        xs = block.series[0].get("values") or []
        ys = block.series[1].get("values") or []
        if xs and len(xs) == len(ys):
            return ShapeSignature.BIVARIATE

    # 6) 单值对限度：单一数值 + 限度。
    has_limit = (block.limit_low is not None or block.limit_high is not None
                 or bool(block.limit_text))
    single_val = (block.units and len([v for v in block.units if _is_number(v)]) == 1) or (
        len(groups) == 1 and _is_number(groups[0].get("value")))
    if has_limit and single_val:
        return ShapeSignature.SINGLE_VALUE_VS_LIMIT

    return ShapeSignature.UNKNOWN


# ===========================================================================
# 任务 4：形态→图型映射 + Eligibility_Gate（门槛与降级）
# 任务 5：参考线来源约束（仅来自 compute 解析值）
# ===========================================================================

#: 出图门槛（下限）。
MIN_TREND_POINTS = 3
MIN_GROUPS = 2
MIN_UNITS = 3

#: 首期支持的图型（需求 13.1）；范围外一律降级 TABLE。
_FIRST_PHASE_CHARTS = {
    ChartType.TREND_BAND, ChartType.GROUPED_BAR, ChartType.GROUPED_DOT,
    ChartType.DISTRIBUTION_DOT, ChartType.BOX_PLOT, ChartType.STATUS_MATRIX,
}

#: 形态 → 首选图型（确定性映射）。
_SHAPE_TO_CHART = {
    ShapeSignature.TIME_SERIES: ChartType.TREND_BAND,
    ShapeSignature.GROUPED_SINGLE_METRIC: ChartType.GROUPED_BAR,
    ShapeSignature.REPEATED_UNIT_DISTRIBUTION: ChartType.DISTRIBUTION_DOT,
    ShapeSignature.CROSS_ATTRIBUTE_COMPLIANCE: ChartType.STATUS_MATRIX,
    ShapeSignature.BIVARIATE: ChartType.SCATTER,
    ShapeSignature.SINGLE_VALUE_VS_LIMIT: ChartType.BULLET,
    ShapeSignature.UNKNOWN: ChartType.TABLE,
}


def _reference_lines_for(block: "DataBlock") -> list:
    """仅用 block 中来自 compute 的限度 / AV 数值构造参考线（需求 5/17）。
    block 的 ``limit_low``/``limit_high``/``acceptance_value`` 均为 compute 逐字拷贝；
    本函数不推导、不外推、不臆造任何数值。
    """
    lines: list = []
    if block.limit_low is not None:
        lines.append({"value": block.limit_low, "label": f"下限 {block.limit_low:g}", "kind": "lower"})
    if block.limit_high is not None:
        lines.append({"value": block.limit_high, "label": f"上限 {block.limit_high:g}", "kind": "upper"})
    av = block.acceptance_value or {}
    if _is_number(av.get("limit")):
        lines.append({"value": av["limit"], "label": f"AV≤{av['limit']:g}", "kind": "av"})
    return lines


def _degrade(block: "DataBlock", intended: "ChartType", reason: str) -> "ChartSpec":
    return ChartSpec(
        chart_type=ChartType.TABLE,
        block_id=block.block_id,
        title=block.title,
        companion_table_block_id=block.block_id,
        degraded_from=intended,
        degrade_reason=reason,
    )


def map_block_to_spec(
    block: "DataBlock",
    sig: "ShapeSignature",
    *,
    first_phase: bool = True,
) -> "ChartSpec":
    """把 (block, 形态) 映射为 ChartSpec；不达门槛或超出首期范围则降级 TABLE。
    返回的 ChartSpec 的 ``reference_lines`` 仅来自 compute 数值（需求 5/17）；
    ``companion_table_block_id`` 指向自身，保证图表-源表配对（需求 14）。
    """
    intended = _SHAPE_TO_CHART.get(sig, ChartType.TABLE)

    # 出图门槛（Eligibility_Gate）。
    if sig is ShapeSignature.TIME_SERIES:
        if len(block.times or []) < MIN_TREND_POINTS:
            return _degrade(block, intended, f"时间点不足{MIN_TREND_POINTS}")
    elif sig is ShapeSignature.GROUPED_SINGLE_METRIC:
        if len(block.groups or []) < MIN_GROUPS:
            return _degrade(block, intended, f"分组不足{MIN_GROUPS}")
    elif sig is ShapeSignature.REPEATED_UNIT_DISTRIBUTION:
        if len([v for v in (block.units or []) if _is_number(v)]) < MIN_UNITS:
            return _degrade(block, intended, f"单位不足{MIN_UNITS}")
    elif sig is ShapeSignature.CROSS_ATTRIBUTE_COMPLIANCE:
        levels = [g for g in (block.groups or []) if g.get("status") or g.get("within_spec") is not None]
        if not levels:
            return _degrade(block, intended, "无任何合规/风险等级")
    elif sig is ShapeSignature.SINGLE_VALUE_VS_LIMIT:
        if block.limit_low is None and block.limit_high is None and not block.limit_text:
            return _degrade(block, intended, "缺少限度，无法绘制子弹图")

    # 首期范围门禁（需求 13.2）。
    if first_phase and intended not in _FIRST_PHASE_CHARTS:
        return _degrade(block, intended, "首期范围外图型")

    return ChartSpec(
        chart_type=intended,
        block_id=block.block_id,
        title=block.title,
        reference_lines=_reference_lines_for(block),
        risk_colored=(intended is ChartType.STATUS_MATRIX),
        redundant_marks=True,
        companion_table_block_id=block.block_id,
    )


# ===========================================================================
# 任务 6：Intent_Weighting；任务 7：图表预算与聚合；任务 8：recommend 编排
# ===========================================================================

#: 各意图下图型的优先级权重（值越大越靠前）。默认序兜底 unknown/缺失。
_INTENT_PRIORITY = {
    "descriptive_summary": {
        ChartType.STATUS_MATRIX: 100, ChartType.DISTRIBUTION_DOT: 90,
        ChartType.GROUPED_BAR: 80, ChartType.GROUPED_DOT: 80,
        ChartType.BOX_PLOT: 70, ChartType.TREND_BAND: 40,
    },
    "shelf_life_extrapolation": {
        ChartType.TREND_BAND: 100, ChartType.GROUPED_BAR: 70,
        ChartType.GROUPED_DOT: 70, ChartType.STATUS_MATRIX: 60,
        ChartType.DISTRIBUTION_DOT: 50,
    },
    "compatibility": {
        ChartType.STATUS_MATRIX: 100, ChartType.GROUPED_BAR: 70,
        ChartType.DISTRIBUTION_DOT: 60, ChartType.TREND_BAND: 50,
    },
}
#: 默认（unknown/缺失）排序：时序 > 分布 > 分组 > 状态。
_DEFAULT_PRIORITY = {
    ChartType.TREND_BAND: 100, ChartType.DISTRIBUTION_DOT: 90,
    ChartType.BOX_PLOT: 85, ChartType.GROUPED_BAR: 80, ChartType.GROUPED_DOT: 80,
    ChartType.STATUS_MATRIX: 70,
}


def _priority_of(chart_type: "ChartType", intent: str) -> int:
    table = _INTENT_PRIORITY.get(str(intent or ""), _DEFAULT_PRIORITY)
    if chart_type is ChartType.TABLE:
        return -1  # 表格始终排在图表之后
    return table.get(chart_type, _DEFAULT_PRIORITY.get(chart_type, 10))


class VisualizationRecommender:
    """图表推荐引擎：summary(+intent) → VisualizationPlan。
    **纯 Python、确定性、不接收 svc/LLM 句柄、不 import matplotlib**。
    """

    def __init__(self, max_figures: int = 6) -> None:
        self.max_figures = int(max_figures)

    def recommend(
        self,
        summary: dict,
        *,
        skill_id: str = "",
        intent: str = "",
        skill: Any = None,
        existing_figures: Optional[dict] = None,  # 预留：稳定性既有 figures（时序）
    ) -> "VisualizationPlan":
        notes: list[str] = []
        blocks = build_data_blocks(summary, skill_id=skill_id, skill=skill)

        specs: list[ChartSpec] = []
        for block in blocks:
            sig = classify_block(block)
            if sig is ShapeSignature.UNKNOWN:
                notes.append(f"{block.block_id}: 无法归类数据形态，降级表格")
            spec = map_block_to_spec(block, sig)
            # 聚合记录（需求 16.2）：分组块记录其被合并的分组标签。
            if block.groups and spec.chart_type in (ChartType.GROUPED_BAR, ChartType.GROUPED_DOT):
                spec.aggregated_from = [str(g.get("label", "")) for g in block.groups]
            if spec.degrade_reason:
                notes.append(f"{block.block_id}: 降级表格（{spec.degrade_reason}）")
            specs.append(spec)

        # 意图加权排序（稳定排序 → 幂等；只改顺序不改数据）。
        order = list(enumerate(specs))
        order.sort(key=lambda t: (-_priority_of(t[1].chart_type, intent), t[0]))
        specs = [s for _, s in order]

        # 图表预算（需求 16.1/16.4）：超额的非表格图降级为表格。
        kept = 0
        for spec in specs:
            if spec.chart_type is ChartType.TABLE:
                continue
            if kept < self.max_figures:
                kept += 1
            else:
                notes.append(f"{spec.block_id}: 超出图表预算({self.max_figures})，降级表格")
                spec.degraded_from = spec.chart_type
                spec.degrade_reason = f"超出图表预算({self.max_figures})"
                spec.chart_type = ChartType.TABLE

        return VisualizationPlan(specs=specs, notes=notes)


# ===========================================================================
# 任务 10 协同：把 VisualizationPlan 转为可渲染的 figures 数据
# ===========================================================================

def figures_from_plan(
    summary: dict,
    plan: "VisualizationPlan",
    *,
    skill: Any = None,
) -> dict:
    """把 plan 中的非 TABLE 图表转为 ChartService 可渲染的 figures 数据条目。
    数值来自重建的 DataBlock（与 recommend 时同一确定性来源，逐字一致）。返回
    ``{block_id: {kind, ...data..., reference_lines, title, caption}}``；TABLE 类
    spec 不产出 figure（由报告层以表格呈现）。
    """
    blocks = {b.block_id: b for b in build_data_blocks(summary, skill=skill)}
    figures: dict = {}
    for spec in plan.specs:
        if spec.chart_type is ChartType.TABLE:
            continue
        block = blocks.get(spec.block_id)
        if block is None:
            continue  # 引用完整性兜底
        ct = spec.chart_type
        common = {
            "title": spec.title or block.title,
            "caption": spec.caption_key or block.title,
            "reference_lines": [dict(r) for r in spec.reference_lines],
        }
        if ct is ChartType.DISTRIBUTION_DOT or ct is ChartType.BOX_PLOT:
            av = block.acceptance_value or {}
            figures[spec.block_id] = {
                "kind": "distribution_dot",
                "box": ct is ChartType.BOX_PLOT,
                "units": list(block.units or []),
                "mean": av.get("mean"),
                "value_label": block.attribute,
                **common,
            }
        elif ct in (ChartType.GROUPED_BAR, ChartType.GROUPED_DOT):
            labels = [str(g.get("label", "")) for g in (block.groups or [])]
            values = [g.get("value") for g in (block.groups or [])]
            figures[spec.block_id] = {
                "kind": "grouped_bar",
                "dot": ct is ChartType.GROUPED_DOT,
                "labels": labels,
                "values": values,
                "value_label": block.attribute,
                **common,
            }
        elif ct is ChartType.STATUS_MATRIX:
            rows = [{"label": str(g.get("label", "")), "status": str(g.get("status", "na"))}
                    for g in (block.groups or [])]
            figures[spec.block_id] = {"kind": "status_matrix", "rows": rows, **common}
        elif ct is ChartType.TREND_BAND:
            series = block.series or []
            traces = [{"times": list(block.times or []),
                       "values": list(s.get("values") or []),
                       "label": str(s.get("label", ""))} for s in series]
            figures[spec.block_id] = {
                "kind": "observed_trends",
                "traces": traces,
                "label": block.attribute,
                **common,
            }
    return figures