core/kernel/pipeline.py

"""固定执行流水线 Pipeline（需求 2.2、2.3）。

底座对任意选定的 Skill 执行**固定且确定**的流水线，从架构层面锁死三层职责与
执行顺序（对应 design.md「4. 流水线 Pipeline」）：

``extract → compute → record_trace →（拒绝分支）→ explain → prompt_guard.clean
→ report.assemble``

关键合规保证：

- **Layer2 ``compute`` 不接收 ``svc``**（需求 2.1）——流水线调用 ``skill.compute(data)``
  时不传任何服务 / LLM 句柄，从源头阻断「大模型幻觉计算」。
- **顺序不可乱**（需求 2.2）：extract（LLM 提取 JSON）→ compute（纯 Python 计算）→
  explain（LLM 仅写文字），且 explain 接收的是**只读**的 ``ComputeResult``。
- **拒绝分支**（需求 8.3）：当 ``compute`` 判定数据不足（``can_proceed=False``）时，
  流水线在 record_trace 之后、explain 之前短路返回拒绝说明，**不再调用 LLM**。

服务健壮性：本阶段 ``Services`` 各字段可能为占位 ``None``（真实服务由任务 4–10
实现）。流水线对 ``audit`` / ``prompt_guard`` / ``report`` 的调用均做存在性判断，
使骨架可在服务未就绪时独立运行与测试；服务就绪后行为与 design 完全一致。
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from .skill_base import ComputeResult, PharmaSkill, RawInput, ReportSections

if TYPE_CHECKING:  # pragma: no cover - 仅类型检查
    from .services import Services

logger = logging.getLogger(__name__)


class Pipeline:
    """固定流水线执行器。

    用法：``Pipeline().run(skill, raw, svc)``。每个阶段产出只读 dataclass，
    作为下一阶段的唯一输入媒介。
    """

    def run(self, skill: PharmaSkill, raw: RawInput, svc: "Services") -> ReportSections:
        """执行固定流水线并返回最终的 :class:`ReportSections`。

        顺序严格为：extract → compute → record_trace →（拒绝分支）→ explain →
        prompt_guard.clean → report.assemble。
        """
        skill_id = getattr(skill.meta, "id", skill.__class__.__name__)

        # Layer1：LLM 可用，仅产出结构化数据。
        data = skill.extract(raw, svc)

        # Layer2：纯 Python 计算。注意——**不传 svc**（需求 2.1 硬保证）。
        result = skill.compute(data)

        # 审计：由底座统一把计算追踪落库（需求 2 / 7.4 / 9.5）。
        self._record_trace(svc, skill_id, result)

        # 拒绝分支：数据不足等情形优雅拒绝，短路返回，不触达 LLM（需求 8.3）。
        if not result.can_proceed:
            return self._render_refusal(svc, result)

        # Layer3：LLM 仅将只读计算结果转写为文字，禁止改数值（需求 2.3）。
        sections = skill.explain(result, svc)

        # 自适应可视化：按数据形态 + 意图推荐图表方案，并把方案转为可渲染的 figures
        # 数据（纯 Python、不接收 svc/LLM、数值仅来自 compute）。失败静默跳过（向后兼容）。
        self._build_visualization_plan(skill, raw, result)

        # 图表渲染：把 compute 产出的图表数据（CI 带等）经 ChartService 渲染为内联 PNG，
        # 写回 result.figures，供报告组装注入（修复「图表已算却未渲染」）。
        self._render_figures(svc, result)

        # 输出强约束：移除 Markdown 围栏 / LLM-SVG / 占位符 / forbidden_phrases（需求 13）。
        sections = self._clean(svc, sections)

        # 报告组装：品牌包装 + 段落拼装 + 图表注入（需求 7.3 / 14）。
        sections.html = self._assemble(svc, skill, result, sections)

        return sections

    # ------------------------------------------------------------------
    # 各阶段的服务调用（对占位 None 服务做存在性判断）
    # ------------------------------------------------------------------
    @staticmethod
    def _record_trace(svc: "Services", skill_id: str, result: ComputeResult) -> None:
        audit = getattr(svc, "audit", None)
        recorder = getattr(audit, "record_trace", None)
        if callable(recorder):
            try:
                recorder(skill_id, result.trace)
            except Exception:  # noqa: BLE001 - 审计失败不得中断主流程
                logger.warning("记录计算追踪失败（skill=%s）。", skill_id, exc_info=True)

    @staticmethod
    def _build_visualization_plan(skill: PharmaSkill, raw: RawInput, result: ComputeResult) -> None:
        """按数据形态 + 意图推荐图表方案，并合入 ``result.figures``（向后兼容、不崩溃）。

        - 推荐引擎为纯 Python、不接收 svc/LLM；数值仅来自 ``result.summary``。
        - 把非 TABLE 图表转为 figures 数据条目（供 ``_render_figures`` 渲染）；
          完整方案存入 ``result.summary["_viz_plan"]`` 供报告层按序注入与降级表呈现。
        - 任一异常或模块缺失即跳过，既有行为不变。
        """
        try:
            from services.visualization import (
                VisualizationRecommender,
                figures_from_plan,
            )
        except Exception:  # noqa: BLE001 - 模块不可用则跳过
            return
        try:
            summary = getattr(result, "summary", None)
            if not isinstance(summary, dict) or not summary:
                return
            intent = ((getattr(raw, "extra", None) or {}).get("task_sheet") or {}).get("intent", "")
            skill_id = getattr(getattr(skill, "meta", None), "id", "") or ""
            plan = VisualizationRecommender().recommend(
                summary, skill_id=skill_id, intent=intent, skill=skill,
            )
            if not plan.specs:
                return
            viz_figs = figures_from_plan(summary, plan, skill=skill)
            figures = getattr(result, "figures", None)
            if not isinstance(figures, dict):
                figures = {}
            # 不覆盖既有同名 figure（如稳定性已产出的图）。
            for key, val in viz_figs.items():
                figures.setdefault(key, val)
            result.figures = figures
            summary["_viz_plan"] = plan.to_dict()
        except Exception:  # noqa: BLE001 - 推荐/转换失败不影响主流程
            logger.warning("构建可视化方案失败（已跳过）。", exc_info=True)

    @staticmethod
    def _render_refusal(svc: "Services", result: ComputeResult) -> ReportSections:
        report = getattr(svc, "report", None)
        renderer = getattr(report, "render_refusal", None)
        if callable(renderer):
            rendered = _call_with_optional_lang(
                renderer, result.refusal, svc
            )
            # report 服务可能直接返回 ReportSections，或返回 HTML 字符串。
            if isinstance(rendered, ReportSections):
                return rendered
            return ReportSections(sections={"refusal": result.refusal or {}}, html=str(rendered))
        # 报告服务未就绪：返回结构化拒绝说明，html 留空。
        return ReportSections(sections={"refusal": result.refusal or {}})

    @staticmethod
    def _render_figures(svc: "Services", result: ComputeResult) -> None:
        """把 ``result.figures`` 中的图表数据经 ChartService 渲染为内联 PNG。

        约定：``compute`` 阶段产出的图表为**纯数据**（如 CI 带的 times/point/lower/
        upper），渲染交由底座 ChartService 统一完成（保持「计算/渲染」分离）。
        渲染成功后把对应条目替换为 ``{"image_base64", "caption"}``，供报告注入。
        ChartService 不可用或渲染失败时静默跳过该图（不影响其它内容）。
        """
        chart = getattr(svc, "chart", None)
        if chart is None:
            return
        figures = getattr(result, "figures", None)
        if not isinstance(figures, dict) or not figures:
            return

        try:
            from services.chart_service import CIBand
        except Exception:  # noqa: BLE001 - ChartService 依赖不可用则跳过
            return

        try:
            from services.chart_service import ObservedTrace
        except Exception:  # noqa: BLE001 - 旧版 ChartService 无 ObservedTrace
            ObservedTrace = None  # type: ignore[assignment]

        rendered: dict = {}
        for key, fig in figures.items():
            # 已是图像（含 image_base64）则原样保留。
            if isinstance(fig, dict) and fig.get("image_base64"):
                rendered[key] = fig
                continue
            # 自适应可视化新增图型：分组条形/点图、分布点图、状态矩阵。
            if isinstance(fig, dict) and fig.get("kind") in (
                "grouped_bar", "distribution_dot", "status_matrix"
            ):
                out = Pipeline._render_viz_chart(chart, fig)
                if out is not None:
                    rendered[key] = out
                    continue
            # 实测趋势图数据 → 渲染为折线散点趋势图（数据梳理可视化）。
            if (
                isinstance(fig, dict)
                and fig.get("kind") == "observed_trends"
                and ObservedTrace is not None
            ):
                drawer = getattr(chart, "observed_trends", None)
                if callable(drawer):
                    try:
                        traces = [
                            ObservedTrace(
                                times=tr.get("times", []),
                                values=tr.get("values", []),
                                label=tr.get("label", ""),
                            )
                            for tr in fig.get("traces", [])
                        ]
                        res = drawer(
                            traces,
                            value_axis_label=fig.get("label") or None,
                            spec_limit=fig.get("spec_limit"),
                        )
                        if getattr(res, "ok", False):
                            rendered[key] = {
                                "image_base64": res.image_base64,
                                "caption": fig.get("caption", fig.get("label", "")),
                            }
                            continue
                    except Exception:  # noqa: BLE001 - 单图渲染失败不影响其它
                        logger.warning("渲染实测趋势图 %s 失败（已跳过）。", key, exc_info=True)
            # CI 带数据 → 渲染为预测带图。
            if isinstance(fig, dict) and {"times", "point", "lower", "upper"} <= set(fig):
                try:
                    band = CIBand(
                        times=fig["times"], point=fig["point"],
                        lower=fig["lower"], upper=fig["upper"],
                        label=fig.get("label", ""),
                        spec_type=fig.get("spec_type", "upper"),
                        spec_limit=fig.get("spec_limit"),
                        shelf_life=fig.get("shelf_life"),
                        observed_t=fig.get("observed_t"),
                        observed_y=fig.get("observed_y"),
                        target_timepoints=fig.get("target_timepoints"),
                    )
                    res = chart.prediction_band(band)
                    if getattr(res, "ok", False):
                        rendered[key] = {
                            "image_base64": res.image_base64,
                            "caption": fig.get("caption", fig.get("label", "")),
                        }
                        continue
                except Exception:  # noqa: BLE001 - 单图渲染失败不影响其它
                    logger.warning("渲染图表 %s 失败（已跳过）。", key, exc_info=True)
            # 其它形态：原样保留（report 层会尽力处理或忽略）。
            rendered[key] = fig
        try:
            result.figures = rendered
        except Exception:  # noqa: BLE001 - result 不可写时忽略
            pass

    @staticmethod
    def _render_viz_chart(chart, fig: dict):
        """渲染自适应可视化新增图型为 ``{image_base64, caption}``；失败返回 None。"""
        kind = fig.get("kind")
        caption = fig.get("caption", fig.get("title", ""))
        try:
            if kind == "grouped_bar":
                drawer = getattr(chart, "grouped_bar", None)
                if not callable(drawer):
                    return None
                res = drawer(
                    fig.get("labels", []), fig.get("values", []),
                    title=fig.get("title", ""), value_label=fig.get("value_label", ""),
                    reference_lines=fig.get("reference_lines"), dot=bool(fig.get("dot")),
                )
            elif kind == "distribution_dot":
                drawer = getattr(chart, "distribution_dot", None)
                if not callable(drawer):
                    return None
                res = drawer(
                    fig.get("units", []), title=fig.get("title", ""),
                    value_label=fig.get("value_label", ""), mean=fig.get("mean"),
                    reference_lines=fig.get("reference_lines"), box=bool(fig.get("box")),
                )
            elif kind == "status_matrix":
                drawer = getattr(chart, "status_matrix", None)
                if not callable(drawer):
                    return None
                res = drawer(fig.get("rows", []), title=fig.get("title", ""))
            else:
                return None
            if getattr(res, "ok", False):
                # 保留原始数据（kind/units/labels/rows...）以便报告层渲染配套源数据表。
                return {**fig, "image_base64": res.image_base64, "caption": caption}
        except Exception:  # noqa: BLE001 - 单图渲染失败不影响其它
            logger.warning("渲染可视化图表（kind=%s）失败（已跳过）。", kind, exc_info=True)
        return None

    @staticmethod
    def _clean(svc: "Services", sections: ReportSections) -> ReportSections:
        guard = getattr(svc, "prompt_guard", None)
        cleaner = getattr(guard, "clean", None)
        if callable(cleaner):
            cleaned = cleaner(sections)
            # clean 约定返回处理后的 ReportSections；若返回 None 则视为原地修改。
            if isinstance(cleaned, ReportSections):
                return cleaned
        return sections

    @staticmethod
    def _assemble(
        svc: "Services",
        skill: PharmaSkill,
        result: ComputeResult,
        sections: ReportSections,
    ) -> str:
        report = getattr(svc, "report", None)
        assembler = getattr(report, "assemble", None)
        if callable(assembler):
            lang = getattr(svc, "lang", None)
            i18n = getattr(svc, "i18n", None)
            # 报告**叙述 / 文案语言**随会话语言（svc.lang）；数值由 compute 决定、与语言
            # 无关（需求 17.3）。对不支持 lang/i18n 关键字的实现做向后兼容回退。
            try:
                return assembler(skill.meta, result, sections, lang=lang, i18n=i18n)
            except TypeError:
                return assembler(skill.meta, result, sections)
        # 报告服务未就绪：保留 explain 阶段已有的 html（可能为空）。
        return sections.html


def _call_with_optional_lang(func, refusal, svc: "Services"):
    """调用 ``render_refusal``，向后兼容是否支持 ``lang`` / ``i18n`` 关键字。"""
    lang = getattr(svc, "lang", None)
    i18n = getattr(svc, "i18n", None)
    try:
        return func(refusal, lang=lang, i18n=i18n)
    except TypeError:
        return func(refusal)


__all__ = ["Pipeline"]