core/services/chart_service.py

"""底座 ``ChartService``：真实置信区间（CI）带与 QbD 风险色块渲染（任务 8）。

对应 design.md「10. ChartService」与需求 4.1 / 4.2 / 4.4 / 11.5 / 14.3：

- **真实 CI 带，零硬编码系数**（需求 4.1 / 4.2）：``prediction_band()`` 接收
  ``compute`` 阶段在连续时间网格上算出的**真实** CI 数据（``times / point /
  lower / upper``）并**直接渲染**喇叭形阴影。本模块**绝不**自行从点预测推导带宽
  （不存在 ``se_scale=0.02`` 之类的硬编码比例系数）——上下界完全来自传入数据，
  渲染层仅作几何绘制。
- **一级动力学非对称带**（需求 4.4）：带的非对称性由 ``compute`` 传入的非对称
  上下界（``upper-point ≠ point-lower``）自然呈现，渲染层原样保留，不强制对称化。
- **多 CQA 叠加图并标注短板 CQA**（需求 11.5）：``multi_cqa_overlay()`` 将多个 CQA
  以对齐的时间轴面板（或单轴叠加）呈现，并以风险色高亮标注决定货架期的**短板
  （木桶）CQA**。
- **统一 QbD 风险色板**（需求 14.3）：红 / 黄 / 绿（高 / 中 / 低危）色板集中定义，
  供报告与图表复用；``risk_color()`` 同时识别中英文与高/中/低三档别名。
- **内嵌中文字体与降级**（需求 14.1 关联）：加载 ``fonts/NotoSansSC-Regular.otf``
  用于中文标签；**字体缺失则降级为英文标签**，不报错、不中断。

可降级性（呼应 design Property 7）：matplotlib / numpy 等重依赖一律 **try-import**；
缺失时各渲染方法返回带提示的 :class:`ChartResult`（``ok=False`` + ``warning``），
**绝不抛出导致应用崩溃**。

输出约定：渲染成功时 ``ChartResult.image_base64`` 为
``"data:image/png;base64,..."`` 形式的内联 PNG（与既有 ``utils/chart_executor.py``
约定一致），可直接嵌入 HTML 报告。
"""

from __future__ import annotations

import base64
import io
import logging
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Sequence

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# QbD 风险色板（红 / 黄 / 绿）— 需求 14.3
# ---------------------------------------------------------------------------

#: 统一 QbD 风险矩阵色板：高危(红) / 中危(黄) / 低危(绿)。
#: 与 ``config/chart_styles.py`` 的 compliant/marginal/non_compliant 取色保持一致。
QBD_RISK_COLORS: dict[str, str] = {
    "high": "#dc3545",    # 红 —— 高危 / 不合规 / 超规格
    "medium": "#ffc107",  # 黄 —— 中危 / 临界 / 接近规格
    "low": "#28a745",     # 绿 —— 低危 / 合规 / 满足规格
}

#: 风险等级别名 → 标准档位（识别中英文与多种表述）。
_RISK_ALIASES: dict[str, str] = {
    # 高危
    "high": "high", "高": "high", "高危": "high", "高风险": "high",
    "danger": "high", "fail": "high", "failed": "high", "critical": "high",
    "non_compliant": "high", "noncompliant": "high", "不合规": "high",
    "不合格": "high", "超规格": "high", "超标": "high", "red": "high",
    # 中危
    "medium": "medium", "mid": "medium", "中": "medium", "中危": "medium",
    "中风险": "medium", "warning": "medium", "warn": "medium",
    "marginal": "medium", "临界": "medium", "注意": "medium", "yellow": "medium",
    "amber": "medium",
    # 低危
    "low": "low", "低": "low", "低危": "low", "低风险": "low",
    "safe": "low", "pass": "low", "passed": "low", "ok": "low",
    "compliant": "low", "合规": "low", "合格": "low", "满足": "low",
    "green": "low",
}

#: 默认档位（无法识别风险等级时使用中危黄，保守提示）。
_DEFAULT_RISK_LEVEL = "medium"


# ---------------------------------------------------------------------------
# 双语标签目录（中文字体可用→中文；缺失→英文）— 需求 14.1 关联
# ---------------------------------------------------------------------------

_LABELS: dict[str, dict[str, str]] = {
    "time_axis": {"zh": "时间 (月)", "en": "Time (months)"},
    "value_axis": {"zh": "数值", "en": "Value"},
    "point_pred": {"zh": "点预测", "en": "Point prediction"},
    "ci_band": {"zh": "置信区间", "en": "Confidence interval"},
    "observed": {"zh": "实测值", "en": "Observed"},
    "spec_limit": {"zh": "规格限", "en": "Spec limit"},
    "shelf_life": {"zh": "货架期", "en": "Shelf life"},
    "limiting_cqa": {"zh": "短板 CQA（木桶原理）", "en": "Limiting CQA (weakest-link)"},
    "prediction_title": {"zh": "预测趋势与置信带", "en": "Prediction trend with CI band"},
    "multi_cqa_title": {"zh": "多 CQA 联合评估", "en": "Multi-CQA joint assessment"},
    "observed_title": {"zh": "实测数据趋势", "en": "Observed data trends"},
    "target_point": {"zh": "预测时间点", "en": "Prediction timepoint"},
    "months": {"zh": "个月", "en": "months"},
}

#: 仓库内内嵌中文字体的默认路径（相对本文件定位 ``<repo>/fonts``）。
#: ``platform/services/chart_service.py`` → 上溯三级到仓库根。
_DEFAULT_FONT_PATH = (
    Path(__file__).resolve().parents[2] / "fonts" / "NotoSansSC-Regular.otf"
)


# ---------------------------------------------------------------------------
# 数据载体
# ---------------------------------------------------------------------------

@dataclass
class CIBand:
    """连续时间网格上的**真实** CI 带数据（由 ``compute`` 阶段产出）。

    渲染层据此直接绘制，不做任何带宽推导：

    - ``times``：时间网格（与 point/lower/upper 等长）。
    - ``point``：点预测序列。
    - ``lower`` / ``upper``：CI 下界 / 上界序列（一级动力学下天然非对称）。
    - ``label``：CQA 名称（用于图例 / 子图标题）。
    - ``spec_type``：``"upper"``（上限型，如杂质）或 ``"lower"``（下限型，如含量）。
    - ``spec_limit``：规格限（可选，绘制为参考线）。
    - ``shelf_life``：该 CQA 满足规格的最长时间（可选，绘制为竖线标记）。
    - ``risk_level``：该 CQA 的风险档位（可选，影响标注色，见 :data:`_RISK_ALIASES`）。
    - ``observed_t`` / ``observed_y``：实测散点（可选，叠加显示）。
    """

    times: Sequence[float]
    point: Sequence[float]
    lower: Sequence[float]
    upper: Sequence[float]
    label: str = ""
    spec_type: str = "upper"
    spec_limit: Optional[float] = None
    shelf_life: Optional[float] = None
    risk_level: Optional[str] = None
    observed_t: Optional[Sequence[float]] = None
    observed_y: Optional[Sequence[float]] = None
    target_timepoints: Optional[Sequence[float]] = None


@dataclass
class ObservedTrace:
    """一条**实测**时间序列（用于数据梳理趋势图，不含任何模型外推）。

    - ``times`` / ``values``：实测时间点（月）与对应测定值，等长且非空。
    - ``label``：序列图例名（通常为「批次@条件」）。
    - ``spec_limit``：可选规格限，仅当**确为用户/数据提供**时绘制参考线（避免
      在无规格时凭空画线，呼应「无规格不得判定合规」的科学约束）。
    """

    times: Sequence[float]
    values: Sequence[float]
    label: str = ""
    spec_limit: Optional[float] = None


@dataclass
class ChartResult:
    """统一的图表渲染结果。

    - ``ok``：是否成功生成图像。
    - ``image_base64``：成功时为 ``"data:image/png;base64,..."`` 内联 PNG。
    - ``warning``：非致命提示（如重依赖缺失已降级、未生成图像）。
    - ``error``：失败原因（数据不合法、渲染异常等）。
    - ``used_chinese_font``：本次渲染是否使用了中文字体（否则为英文标签降级）。
    """

    ok: bool
    image_base64: str = ""
    warning: str = ""
    error: str = ""
    used_chinese_font: bool = False

    def __bool__(self) -> bool:  # 便于 ``if result:`` 直接判定成功与否
        return self.ok


# ---------------------------------------------------------------------------
# ChartService
# ---------------------------------------------------------------------------

class ChartService:
    """真实 CI 带与 QbD 风险色块渲染服务。

    构造参数：
    - ``font_path``：内嵌中文字体路径。默认指向 ``<repo>/fonts/NotoSansSC-Regular.otf``。
    - ``enable_chinese``：是否启用中文标签（即便字体存在也可显式关闭以测试英文降级）。
    - ``dpi``：导出 PNG 的分辨率。
    """

    def __init__(
        self,
        font_path: Optional[str] = None,
        *,
        enable_chinese: bool = True,
        dpi: int = 150,
    ) -> None:
        self.dpi = dpi
        self._font_path = (
            Path(font_path) if font_path is not None else _DEFAULT_FONT_PATH
        )
        # 是否真正具备中文渲染能力：启用开关 + 字体文件存在。
        self._enable_chinese = bool(enable_chinese)
        self._font_available = self._enable_chinese and self._font_path.is_file()
        self._font_prop = None  # 延迟到首次渲染时构造（依赖 matplotlib）

    # ------------------------------------------------------------------
    # 语言 / 字体
    # ------------------------------------------------------------------
    @property
    def chinese_available(self) -> bool:
        """当前是否以中文标签渲染（字体可用且未被显式关闭）。"""
        return self._font_available

    @property
    def lang(self) -> str:
        """当前标签语言：``"zh"``（中文字体可用）或 ``"en"``（降级英文）。"""
        return "zh" if self._font_available else "en"

    def label(self, key: str) -> str:
        """按当前语言取标签文案；缺 key 时回退 key 本身。"""
        entry = _LABELS.get(key)
        if not entry:
            return key
        return entry.get(self.lang) or entry.get("en") or key

    # ------------------------------------------------------------------
    # QbD 风险色板（需求 14.3）
    # ------------------------------------------------------------------
    @staticmethod
    def risk_palette() -> dict[str, str]:
        """返回统一 QbD 风险色板（high/medium/low → 红/黄/绿）的副本。"""
        return dict(QBD_RISK_COLORS)

    @staticmethod
    def normalize_risk_level(level: Optional[str]) -> str:
        """把任意风险表述归一化为 ``"high"`` / ``"medium"`` / ``"low"`` 三档。

        识别中英文及多种别名（见 :data:`_RISK_ALIASES`）；无法识别时返回
        默认档位（中危）。
        """
        if level is None:
            return _DEFAULT_RISK_LEVEL
        key = str(level).strip().lower()
        if key in _RISK_ALIASES:
            return _RISK_ALIASES[key]
        # 容错：子串匹配（如「中等风险」「high risk」）。
        for alias, normalized in _RISK_ALIASES.items():
            if alias and alias in key:
                return normalized
        return _DEFAULT_RISK_LEVEL

    @classmethod
    def risk_color(cls, level: Optional[str]) -> str:
        """按风险等级返回 QbD 红 / 黄 / 绿色值（识别中英文与高/中/低）。"""
        return QBD_RISK_COLORS[cls.normalize_risk_level(level)]

    # ------------------------------------------------------------------
    # 带数据校验（纯函数，无副作用，便于测试）
    # ------------------------------------------------------------------
    @staticmethod
    def validate_band(band: "CIBand"):
        """校验并归一化一条 CI 带为等长 numpy 数组，返回 ``(times, point, lower, upper)``。

        - 四个序列必须等长且非空。
        - **不**对带宽做任何缩放 / 推导——上下界原样保留（需求 4.2 的硬保证）。
        - 任一界越界（``lower > upper``）将抛出 ``ValueError``，以便尽早暴露上游错误。

        依赖 numpy；numpy 不可用时抛出 ``RuntimeError``（由调用方降级处理）。
        """
        try:
            import numpy as np
        except ImportError as exc:  # pragma: no cover - 环境缺 numpy 时降级
            raise RuntimeError("numpy 不可用，无法处理 CI 带数据。") from exc

        times = np.asarray(band.times, dtype=float)
        point = np.asarray(band.point, dtype=float)
        lower = np.asarray(band.lower, dtype=float)
        upper = np.asarray(band.upper, dtype=float)

        n = times.size
        if n == 0:
            raise ValueError("CI 带时间网格为空。")
        if not (point.size == lower.size == upper.size == n):
            raise ValueError(
                "CI 带各序列长度不一致："
                f"times={times.size}, point={point.size}, "
                f"lower={lower.size}, upper={upper.size}。"
            )
        if np.any(lower > upper + 1e-9):
            raise ValueError("CI 带存在 lower > upper 的非法区间。")
        return times, point, lower, upper

    @classmethod
    def band_halfwidths(cls, band: "CIBand"):
        """返回 ``(upper-point, point-lower)`` 两个半宽序列（numpy 数组）。

        用于**证明**渲染层直接消费真实上下界：当 ``upper-point != point-lower`` 时，
        带为非对称（一级动力学，需求 4.4）。此方法不做任何缩放，仅作差。
        """
        _, point, lower, upper = cls.validate_band(band)
        return upper - point, point - lower

    # ------------------------------------------------------------------
    # 重依赖准备（matplotlib / numpy）— try-import 优雅降级
    # ------------------------------------------------------------------
    def _ensure_backend(self):
        """准备渲染后端，返回 ``(plt, np)``；任一缺失返回 ``None``（调用方降级）。"""
        try:
            import matplotlib
            matplotlib.use("Agg")  # 非交互后端，适合服务端 / 测试
            import matplotlib.pyplot as plt
            import numpy as np
        except ImportError as exc:
            logger.warning("matplotlib / numpy 不可用，图表渲染降级：%s", exc)
            return None

        # 字体配置：中文字体可用则注册并设为首选；缺失则保持默认（英文标签）。
        font_name = None
        if self._font_available and self._font_prop is None:
            try:
                from matplotlib import font_manager

                font_manager.fontManager.addfont(str(self._font_path))
                self._font_prop = font_manager.FontProperties(fname=str(self._font_path))
            except Exception as exc:  # noqa: BLE001 - 字体加载失败即降级英文
                logger.warning("加载中文字体失败，降级英文标签：%s", exc)
                self._font_available = False
                self._font_prop = None
        if self._font_prop is not None:
            try:
                font_name = self._font_prop.get_name()
            except Exception:  # noqa: BLE001
                font_name = None

        # 出版级（SCI）排版基线：统一字体、去顶/右边框、细线宽、外向刻度、
        # 一致字号与高分辨率导出。所有图表共用，无需逐方法重复设置。
        self._apply_publication_style(plt, font_name)
        return plt, np

    def _apply_publication_style(self, plt, font_name: Optional[str]) -> None:
        """设置 matplotlib 出版级 rcParams（SCI 论文取向）。

        - **简洁坐标轴**：去除上 / 右边框、外向短刻度、细线宽——符合期刊制图规范。
        - **一致字号与高 DPI**：正文 9.5、标题略大、加粗标题；高分辨率导出。

        说明：**不**全局改写 ``font.family`` / ``font.sans-serif``——既有渲染通过
        逐文本 ``fontproperties`` 应用内嵌字体（见 ``_font_kwargs``），全局改写字体族会
        与字体缓存 / ``tight_layout`` 交互而引入跨渲染不确定性。这里仅设置确定性、
        与文本字体解析无关的外观参数。
        """
        rc = {
            "axes.titleweight": "bold",
            "figure.titleweight": "bold",
            # 简洁坐标轴（去顶/右框、细线宽、外向刻度）
            "axes.spines.top": False,
            "axes.spines.right": False,
            "axes.linewidth": 0.8,
            "axes.edgecolor": "#333333",
            "axes.labelcolor": "#222222",
            "axes.titlecolor": "#1a1a1a",
            "text.color": "#222222",
            "xtick.color": "#333333",
            "ytick.color": "#333333",
            "xtick.direction": "out",
            "ytick.direction": "out",
            "xtick.major.width": 0.8,
            "ytick.major.width": 0.8,
            "xtick.major.size": 3.5,
            "ytick.major.size": 3.5,
            # 网格（默认关闭，由各图按需开启；统一为浅虚线风格）
            "grid.color": "#dfe3e6",
            "grid.linewidth": 0.6,
            "grid.linestyle": "--",
            "legend.frameon": False,
            # 负号正常显示（自定义字体下尤为重要）
            "axes.unicode_minus": False,
            # 高质量导出（注意：不设 savefig.bbox="tight" —— 它会按文本范围裁剪，
            # 使画布尺寸随标签文字长度变化、并引入渲染不确定性，破坏可复现比较。
            # 统一裁剪由各图的 tight_layout + 固定 figsize 保证）。
            "figure.dpi": self.dpi,
            "savefig.dpi": self.dpi,
        }
        try:
            plt.rcParams.update(rc)
        except Exception as exc:  # noqa: BLE001 - 个别键不被某版本支持时不致命
            logger.info("应用出版级图表样式时部分参数被忽略：%s", exc)
            plt.rcParams["axes.unicode_minus"] = False

    def _font_kwargs(self) -> dict:
        """返回供 matplotlib 文本接口使用的字体参数（中文时附 FontProperties）。"""
        if self._font_available and self._font_prop is not None:
            return {"fontproperties": self._font_prop}
        return {}

    def _fig_to_result(self, plt, fig) -> ChartResult:
        """把图形导出为内联 PNG 的 :class:`ChartResult`，并释放资源。"""
        try:
            buf = io.BytesIO()
            fig.savefig(
                buf,
                format="png",
                dpi=self.dpi,
                bbox_inches="tight",
                facecolor="white",
                edgecolor="none",
            )
            buf.seek(0)
            encoded = base64.b64encode(buf.getvalue()).decode("ascii")
            return ChartResult(
                ok=True,
                image_base64=f"data:image/png;base64,{encoded}",
                used_chinese_font=self._font_available,
            )
        finally:
            plt.close(fig)

    # ------------------------------------------------------------------
    # 单 CQA 预测带（需求 4.1 / 4.2 / 4.4）
    # ------------------------------------------------------------------
    def prediction_band(
        self,
        band: "CIBand",
        *,
        title: Optional[str] = None,
        value_axis_label: Optional[str] = None,
    ) -> ChartResult:
        """渲染单个 CQA 的真实预测带（喇叭形 / 非对称阴影直接来自传入上下界）。

        **不含任何硬编码比例系数**：阴影由 ``band.lower`` / ``band.upper`` 直接绘制；
        一级动力学的非对称形态原样保留（需求 4.2 / 4.4）。
        """
        backend = self._ensure_backend()
        if backend is None:
            return ChartResult(
                ok=False,
                warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。",
            )
        plt, np = backend

        try:
            times, point, lower, upper = self.validate_band(band)
        except (ValueError, RuntimeError) as exc:
            return ChartResult(ok=False, error=f"CI 带数据不合法：{exc}")

        try:
            fig, ax = plt.subplots(figsize=(7.2, 4.5))
            self._draw_band(ax, plt, np, times, point, lower, upper, band)

            ax.set_xlabel(self.label("time_axis"), **self._font_kwargs())
            ax.set_ylabel(
                value_axis_label or band.label or self.label("value_axis"),
                **self._font_kwargs(),
            )
            ax.set_title(
                title or self.label("prediction_title"),
                **self._font_kwargs(),
                fontsize=13,
            )
            self._apply_legend(ax)
            ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6)
            fig.tight_layout()
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001 - 渲染异常不得中断主流程
            logger.warning("渲染预测带失败：%s", exc, exc_info=True)
            plt.close("all")
            return ChartResult(ok=False, error=f"渲染预测带失败：{exc}")

    # ------------------------------------------------------------------
    # 实测数据趋势图（数据梳理可视化）—— 仅画实测点，不含外推
    # ------------------------------------------------------------------
    def observed_trends(
        self,
        traces: Sequence["ObservedTrace"],
        *,
        title: Optional[str] = None,
        value_axis_label: Optional[str] = None,
        spec_limit: Optional[float] = None,
    ) -> ChartResult:
        """把多条**实测**序列绘成折线散点趋势图（数据梳理可视化）。

        与 :meth:`prediction_band` 严格区分：本图**只呈现实测数据本身**（按批次×条件
        分系列的折线 + 散点），不绘制任何模型外推或置信带，用于「先把原始数据梳理
        清楚、再谈建模」。规格限仅在 ``spec_limit`` 显式提供（即确有规格）时绘制参考
        线——无规格时不画线，避免误导性的「合规」暗示。

        可降级：matplotlib/numpy 缺失或无有效序列时返回 ``ok=False`` 的
        :class:`ChartResult`，绝不抛出。
        """
        backend = self._ensure_backend()
        if backend is None:
            return ChartResult(
                ok=False,
                warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。",
            )
        plt, np = backend

        # 过滤出至少含 1 个点的有效序列。
        valid: list[tuple[ObservedTrace, Any, Any]] = []
        for tr in traces or []:
            try:
                t = np.asarray(tr.times, dtype=float)
                y = np.asarray(tr.values, dtype=float)
            except Exception:  # noqa: BLE001 - 单序列异常跳过
                continue
            if t.size and t.size == y.size:
                valid.append((tr, t, y))
        if not valid:
            return ChartResult(ok=False, error="无有效实测序列可绘制。")

        try:
            palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#9467bd",
                       "#8c564b", "#e377c2", "#17becf", "#bcbd22"]
            fig, ax = plt.subplots(figsize=(7.6, 4.6))
            for idx, (tr, t, y) in enumerate(valid):
                color = palette[idx % len(palette)]
                order = np.argsort(t)
                ax.plot(
                    t[order], y[order], color=color, linewidth=1.6,
                    marker="o", markersize=5, markeredgecolor="white",
                    markeredgewidth=0.6,
                    label=(tr.label or f"{self.label('observed')} {idx + 1}"),
                )

            # 规格限参考线：仅在显式提供时绘制（无规格不画，避免误导）。
            eff_spec = spec_limit
            if eff_spec is None:
                for tr, _, _ in valid:
                    if tr.spec_limit is not None:
                        eff_spec = tr.spec_limit
                        break
            if eff_spec is not None:
                ax.axhline(
                    eff_spec, color=QBD_RISK_COLORS["high"],
                    linewidth=1.2, linestyle="--",
                    label=f"{self.label('spec_limit')}={eff_spec:g}",
                )

            ax.set_xlabel(self.label("time_axis"), **self._font_kwargs())
            ax.set_ylabel(
                value_axis_label or self.label("value_axis"),
                **self._font_kwargs(),
            )
            ax.set_title(
                title or self.label("observed_title"),
                **self._font_kwargs(), fontsize=13,
            )
            ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6)
            self._apply_legend(ax)
            fig.tight_layout()
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001 - 渲染异常不得中断主流程
            logger.warning("渲染实测趋势图失败：%s", exc, exc_info=True)
            plt.close("all")
            return ChartResult(ok=False, error=f"渲染实测趋势图失败：{exc}")

    # ------------------------------------------------------------------
    # 多 CQA 叠加 / 分图，并标注短板 CQA（需求 11.5）
    # ------------------------------------------------------------------
    def multi_cqa_overlay(
        self,
        bands: Sequence["CIBand"],
        *,
        limiting_cqa: Optional[str] = None,
        title: Optional[str] = None,
        mode: str = "panels",
    ) -> ChartResult:
        """渲染多个 CQA 并标注决定货架期的短板（木桶）CQA（需求 11.5）。

        - ``mode="panels"``（默认）：每个 CQA 一个对齐时间轴的子图面板，避免不同
          量纲（含量 ~100% vs 杂质 ~0.3%）混轴失真；短板 CQA 的标题以高危红高亮。
        - ``mode="overlay"``：所有 CQA 叠加于同一坐标轴（适用于同量纲对比）。

        ``limiting_cqa`` 指定短板 CQA 的 ``label``；缺省时若各 band 提供
        ``shelf_life``，自动取最短者为短板（木桶原理）。
        """
        backend = self._ensure_backend()
        if backend is None:
            return ChartResult(
                ok=False,
                warning="matplotlib / numpy 不可用，已跳过图表渲染（核心分析不受影响）。",
            )
        plt, np = backend

        if not bands:
            return ChartResult(ok=False, error="未提供任何 CQA 带数据。")

        # 预校验全部 band，任一不合法即整体报错（尽早暴露上游问题）。
        validated = []
        try:
            for band in bands:
                validated.append((band, *self.validate_band(band)))
        except (ValueError, RuntimeError) as exc:
            return ChartResult(ok=False, error=f"CI 带数据不合法：{exc}")

        limiting = self._resolve_limiting(bands, limiting_cqa)

        try:
            if mode == "overlay":
                fig = self._render_overlay(plt, np, validated, limiting, title)
            else:
                fig = self._render_panels(plt, np, validated, limiting, title)
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001 - 渲染异常不得中断主流程
            logger.warning("渲染多 CQA 图失败：%s", exc, exc_info=True)
            plt.close("all")
            return ChartResult(ok=False, error=f"渲染多 CQA 图失败：{exc}")

    # ------------------------------------------------------------------
    # 短板 CQA 判定（木桶原理）
    # ------------------------------------------------------------------
    @staticmethod
    def _resolve_limiting(
        bands: Sequence["CIBand"], limiting_cqa: Optional[str]
    ) -> Optional[str]:
        """确定短板 CQA 的 label：优先显式指定，否则取 shelf_life 最短者。"""
        if limiting_cqa:
            return limiting_cqa
        candidates = [
            b for b in bands if b.shelf_life is not None and b.label
        ]
        if not candidates:
            return None
        shortest = min(candidates, key=lambda b: b.shelf_life)
        return shortest.label

    # ------------------------------------------------------------------
    # 内部绘制原语
    # ------------------------------------------------------------------
    def _draw_band(self, ax, plt, np, times, point, lower, upper, band, *, color=None):
        """在给定坐标轴上绘制一条带（阴影 + 点预测 + 规格线 + 货架期 + 实测点）。

        阴影直接由 ``lower`` / ``upper`` 填充——这是「真实 CI、零硬编码系数」的核心：
        渲染层不重算带宽，非对称形态原样呈现（需求 4.2 / 4.4）。
        """
        line_color = color or "#003366"
        # 真实 CI 阴影（喇叭形 / 非对称均由数据决定）。
        ax.fill_between(
            times, lower, upper,
            color=line_color, alpha=0.18, linewidth=0,
            label=self.label("ci_band"),
        )
        # 上下包络线（虚线），强调带边界形态。
        ax.plot(times, upper, color=line_color, alpha=0.5, linewidth=0.8, linestyle=":")
        ax.plot(times, lower, color=line_color, alpha=0.5, linewidth=0.8, linestyle=":")
        # 点预测线。
        ax.plot(times, point, color=line_color, linewidth=1.8, label=self.label("point_pred"))

        # 规格限参考线（按上限/下限型着色）。
        if band.spec_limit is not None:
            ax.axhline(
                band.spec_limit, color=QBD_RISK_COLORS["high"],
                linewidth=1.2, linestyle="--",
                label=f"{self.label('spec_limit')}={band.spec_limit:g}",
            )
        # 货架期竖线标记。
        if band.shelf_life is not None:
            ax.axvline(
                band.shelf_life, color=QBD_RISK_COLORS["medium"],
                linewidth=1.2, linestyle="-.",
                label=f"{self.label('shelf_life')}={band.shelf_life:g}",
            )
        # 实测散点。
        if band.observed_t is not None and band.observed_y is not None:
            ot = np.asarray(band.observed_t, dtype=float)
            oy = np.asarray(band.observed_y, dtype=float)
            if ot.size and ot.size == oy.size:
                ax.scatter(
                    ot, oy, color="#0066cc", s=28, zorder=5,
                    edgecolors="white", linewidths=0.6,
                    label=self.label("observed"),
                )

        # 目标预测时间点高亮标记（不同颜色 + 竖线 + 数值标注），便于用户直观感知。
        self._mark_target_timepoints(ax, np, times, point, band)

    def _mark_target_timepoints(self, ax, np, times, point, band) -> None:
        """在预测曲线上以醒目颜色标注用户的目标预测时间点（竖线 + 星形点 + 数值）。

        预测值由真实点预测序列在该时间点插值得到（不另算）；时间点超出网格则跳过。
        """
        tps = getattr(band, "target_timepoints", None)
        if not tps:
            return
        try:
            t_arr = np.asarray(times, dtype=float)
            y_arr = np.asarray(point, dtype=float)
        except Exception:  # noqa: BLE001
            return
        if t_arr.size == 0:
            return
        marker_color = "#d6336c"  # 醒目品红，与 QbD 红/黄/绿区分，专指"预测时间点"
        labeled = False
        for tp in tps:
            try:
                tp_f = float(tp)
            except (TypeError, ValueError):
                continue
            if tp_f < float(t_arr.min()) or tp_f > float(t_arr.max()):
                continue
            y_at = float(np.interp(tp_f, t_arr, y_arr))
            ax.axvline(
                tp_f, color=marker_color, linewidth=1.0, linestyle="--", alpha=0.7,
            )
            ax.scatter(
                [tp_f], [y_at], color=marker_color, s=90, marker="*", zorder=6,
                edgecolors="white", linewidths=0.8,
                label=(self.label("target_point") if not labeled else None),
            )
            labeled = True
            ax.annotate(
                f"{tp_f:g}{self.label('months')}: {y_at:.3g}",
                xy=(tp_f, y_at),
                xytext=(4, 8), textcoords="offset points",
                fontsize=8, color=marker_color,
                **self._font_kwargs(),
            )

    def _render_panels(self, plt, np, validated, limiting, title):
        """每个 CQA 一个子图面板，短板 CQA 以高危红高亮标题。"""
        n = len(validated)
        fig, axes = plt.subplots(
            n, 1, figsize=(7.2, 2.8 * n + 0.4), squeeze=False, sharex=True
        )
        for idx, (band, times, point, lower, upper) in enumerate(validated):
            ax = axes[idx][0]
            self._draw_band(ax, plt, np, times, point, lower, upper, band)
            is_limiting = bool(limiting) and band.label == limiting
            label = band.label or f"CQA {idx + 1}"
            if is_limiting:
                marker = self.label("limiting_cqa")
                ax.set_title(
                    f"★ {label} — {marker}",
                    color=QBD_RISK_COLORS["high"],
                    **self._font_kwargs(),
                    fontsize=12,
                )
                # 用高危红描边强调短板面板。
                for spine in ax.spines.values():
                    spine.set_edgecolor(QBD_RISK_COLORS["high"])
                    spine.set_linewidth(1.6)
            else:
                ax.set_title(label, **self._font_kwargs(), fontsize=12)
            ax.set_ylabel(label, **self._font_kwargs())
            ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6)
            self._apply_legend(ax, fontsize=8)

        axes[-1][0].set_xlabel(self.label("time_axis"), **self._font_kwargs())
        fig.suptitle(
            title or self.label("multi_cqa_title"),
            **self._font_kwargs(),
            fontsize=14,
        )
        fig.tight_layout(rect=(0, 0, 1, 0.97))
        return fig

    def _render_overlay(self, plt, np, validated, limiting, title):
        """所有 CQA 叠加于同一坐标轴；短板 CQA 加粗并以高危红高亮。"""
        palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#9467bd", "#8c564b", "#e377c2"]
        fig, ax = plt.subplots(figsize=(8.0, 5.0))
        for idx, (band, times, point, lower, upper) in enumerate(validated):
            is_limiting = bool(limiting) and band.label == limiting
            color = QBD_RISK_COLORS["high"] if is_limiting else palette[idx % len(palette)]
            label = band.label or f"CQA {idx + 1}"
            ax.fill_between(times, lower, upper, color=color, alpha=0.12, linewidth=0)
            ax.plot(
                times, point, color=color,
                linewidth=2.6 if is_limiting else 1.6,
                label=(f"★ {label}" if is_limiting else label),
            )
            if band.spec_limit is not None:
                ax.axhline(band.spec_limit, color=color, linewidth=0.9, linestyle="--", alpha=0.7)
            if band.observed_t is not None and band.observed_y is not None:
                ot = np.asarray(band.observed_t, dtype=float)
                oy = np.asarray(band.observed_y, dtype=float)
                if ot.size and ot.size == oy.size:
                    ax.scatter(ot, oy, color=color, s=24, zorder=5,
                               edgecolors="white", linewidths=0.5)

        ax.set_xlabel(self.label("time_axis"), **self._font_kwargs())
        ax.set_ylabel(self.label("value_axis"), **self._font_kwargs())
        subtitle = title or self.label("multi_cqa_title")
        if limiting:
            subtitle = f"{subtitle}（{self.label('limiting_cqa')}: {limiting}）" \
                if self._font_available else f"{subtitle} ({self.label('limiting_cqa')}: {limiting})"
        ax.set_title(subtitle, **self._font_kwargs(), fontsize=13)
        ax.grid(True, color="#e0e0e0", linestyle="--", linewidth=0.6)
        self._apply_legend(ax)
        fig.tight_layout()
        return fig

    def _apply_legend(self, ax, *, fontsize: int = 9) -> None:
        """添加图例；中文字体可用时为图例文本设定字体属性。"""
        handles, labels = ax.get_legend_handles_labels()
        if not handles:
            return
        legend = ax.legend(loc="best", fontsize=fontsize, framealpha=0.9)
        if self._font_available and self._font_prop is not None:
            for text in legend.get_texts():
                text.set_fontproperties(self._font_prop)

    # ------------------------------------------------------------------
    # 自适应可视化新增图型（adaptive-report-visualization 任务 9）
    # 全部：matplotlib 缺失→ok=False；中文字体缺失→英文标签仍出图；不抛异常。
    # ------------------------------------------------------------------
    def _unavailable(self) -> ChartResult:
        return ChartResult(ok=False, warning="matplotlib/numpy 不可用，图表已降级。")

    def grouped_bar(self, labels, values, *, title="", value_label="",
                    reference_lines=None, dot=False) -> ChartResult:
        """分组条形图（``dot=True`` 时为点图）：每个分组一个数值。

        ``reference_lines``：``[{value,label,kind}]``，仅绘制由调用方（来自 compute）
        提供的限度参考线，不推导任何值。
        """
        backend = self._ensure_backend()
        if backend is None:
            return self._unavailable()
        plt, np = backend
        try:
            fig, ax = plt.subplots(figsize=(7.2, 4.0))
            x = list(range(len(labels)))
            fk = self._font_kwargs()
            if dot:
                ax.scatter(x, values, s=70, color="#1f6f78", zorder=3)
            else:
                ax.bar(x, values, color="#1f6f78", width=0.6, zorder=3)
            for xi, v in zip(x, values):
                ax.annotate(f"{v:g}", (xi, v), textcoords="offset points",
                            xytext=(0, 5), ha="center", fontsize=8, **fk)
            for rl in (reference_lines or []):
                if not isinstance(rl.get("value"), (int, float)):
                    continue
                ax.axhline(rl["value"], color="#d08a1d", linestyle="--", linewidth=1.2)
                ax.annotate(str(rl.get("label", "")), (x[-1] if x else 0, rl["value"]),
                            fontsize=8, color="#8a5a12", **fk)
            ax.set_xticks(x)
            ax.set_xticklabels([str(l) for l in labels], **fk)
            if value_label:
                ax.set_ylabel(value_label, **fk)
            if title:
                ax.set_title(title, **fk)
            ax.grid(axis="y", linestyle=":", alpha=0.5)
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001
            logger.warning("grouped_bar 渲染失败：%s", exc)
            return ChartResult(ok=False, error=str(exc))

    def distribution_dot(self, units, *, title="", value_label="", mean=None,
                         reference_lines=None, box=False) -> ChartResult:
        """重复单位分布点图（``box=True`` 叠加箱线）：逐单位散点 + 均值线 + 参考线。"""
        backend = self._ensure_backend()
        if backend is None:
            return self._unavailable()
        plt, np = backend
        try:
            nums = [float(v) for v in units if isinstance(v, (int, float))]
            fig, ax = plt.subplots(figsize=(6.4, 4.0))
            fk = self._font_kwargs()
            if box and len(nums) >= 2:
                ax.boxplot(nums, vert=True, widths=0.4, positions=[1],
                           patch_artist=True,
                           boxprops=dict(facecolor="#eef6f7", color="#1f6f78"))
            jitter = (np.random.default_rng(0).uniform(-0.06, 0.06, size=len(nums))
                      if nums else [])
            ax.scatter([1 + j for j in jitter], nums, s=60, color="#1f6f78",
                       zorder=3, alpha=0.85)
            if mean is None and nums:
                mean = float(np.mean(nums))
            if isinstance(mean, (int, float)):
                ax.axhline(mean, color="#1f6f78", linewidth=1.5,
                           label=(self.label("mean") if self.label("mean") != "mean" else "mean"))
                ax.annotate(f"{mean:g}", (1.15, mean), fontsize=8, color="#155057", **fk)
            for rl in (reference_lines or []):
                if not isinstance(rl.get("value"), (int, float)):
                    continue
                ax.axhline(rl["value"], color="#d08a1d", linestyle="--", linewidth=1.2)
                ax.annotate(str(rl.get("label", "")), (0.6, rl["value"]),
                            fontsize=8, color="#8a5a12", **fk)
            ax.set_xticks([])
            if value_label:
                ax.set_ylabel(value_label, **fk)
            if title:
                ax.set_title(title, **fk)
            ax.grid(axis="y", linestyle=":", alpha=0.5)
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001
            logger.warning("distribution_dot 渲染失败：%s", exc)
            return ChartResult(ok=False, error=str(exc))

    def status_matrix(self, rows, *, title="") -> ChartResult:
        """跨属性合规状态矩阵：颜色 + 符号/文字双通道（无障碍，需求 15）。

        ``rows``：``[{label, status in {pass,fail,na}}]``。
        """
        backend = self._ensure_backend()
        if backend is None:
            return self._unavailable()
        plt, np = backend
        try:
            fk = self._font_kwargs()
            color_map = {"pass": QBD_RISK_COLORS["low"], "fail": QBD_RISK_COLORS["high"],
                         "na": "#c7ccd1"}
            mark_map = {"pass": "√", "fail": "×", "na": "—"}
            n = len(rows)
            fig, ax = plt.subplots(figsize=(6.6, max(1.2, 0.42 * n + 0.6)))
            for i, r in enumerate(rows):
                y = n - 1 - i
                status = str(r.get("status", "na"))
                ax.add_patch(plt.Rectangle((0, y), 1, 0.9,
                             facecolor=color_map.get(status, "#c7ccd1"),
                             edgecolor="white"))
                # 冗余编码：色块内放符号；右侧放文字标签。
                ax.text(0.5, y + 0.45, mark_map.get(status, "—"),
                        ha="center", va="center", fontsize=13, color="white", **fk)
                ax.text(1.1, y + 0.45, str(r.get("label", "")),
                        ha="left", va="center", fontsize=9, **fk)
            ax.set_xlim(0, 4)
            ax.set_ylim(0, n)
            ax.axis("off")
            if title:
                ax.set_title(title, **fk)
            return self._fig_to_result(plt, fig)
        except Exception as exc:  # noqa: BLE001
            logger.warning("status_matrix 渲染失败：%s", exc)
            return ChartResult(ok=False, error=str(exc))


__all__ = [
    "ChartService",
    "CIBand",
    "ObservedTrace",
    "ChartResult",
    "QBD_RISK_COLORS",
]