core/kernel/task_sheet.py

"""《分析任务单》数据模型（理解层产物，intent-understanding-layer 任务 1）。

对应 design.md「Components and Interfaces / 1. Data model」。本模块定义理解层
（``UnderstandingLayer``）在正式分析（``compute``）之前产出的**强结构化、可审计、
可序列化**的中间契约——《分析任务单》（:class:`AnalysisTaskSheet`），以及其内嵌的
文档画像、提取项与澄清决策类型。

设计取向（与需求 / 设计一致）：

- **不含任何分析数值计算**：任务单只承载「文档画像 + 意图 + 来源可定位的提取项 +
  缺失/歧义 + 澄清决策」，所有分析数值由后续 ``compute`` 阶段（纯 Python）产出
  （需求 1.4、Property 4）。
- **可序列化且往返一致**：:meth:`AnalysisTaskSheet.to_dict` /
  :meth:`AnalysisTaskSheet.from_dict` 互为逆操作（``from_dict(to_dict(x)) == x``，
  需求 9.3、Property 9），便于会话状态保存、审计与无 UI / 无 LLM 的单元测试。
- **仅依赖标准库**：纯 dataclass + enum，不触网、无重依赖，便于稳定测试。

枚举值采用稳定的字符串字面量，作为对外序列化契约的一部分（不可随意更名）。
"""

from __future__ import annotations

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


# ---------------------------------------------------------------------------
# 枚举：意图、表类型、澄清原因码
# ---------------------------------------------------------------------------

class Intent(str, Enum):
    """用户请求意图分类（需求 3）。

    - ``DESCRIPTIVE_SUMMARY``：描述性质量属性梳理（分组/统计/限度符合性，不外推）。
    - ``SHELF_LIFE``：货架期外推 / 动力学建模。
    - ``UNKNOWN``：无法自信分类 → 进入澄清模式（需求 3.4 / 6.4）。
    """

    DESCRIPTIVE_SUMMARY = "descriptive_summary"
    SHELF_LIFE = "shelf_life_extrapolation"
    COMPATIBILITY = "compatibility"
    UNKNOWN = "unknown"


class SlotState(str, Enum):
    """意图槽位的**来源态**（意图保真核心元信息）。

    意图是潜变量——系统只能从「用户话语 + 文档上下文」观测它。为确保**用户真实意图
    被采纳**，每个槽位必须显式携带其取值来源，杜绝「系统凭空具象化用户从未表达的意图」：

    - ``STATED``：用户在指令中**明示**——最高优先级，经确认后锁定，永不被重推断覆盖。
    - ``INFERRED``：由文档画像 / 输入信号**语用推断**——可用，但必须向用户**回显来源**
      并允许一键改正（推断可以激进，但绝不隐瞒「这是推断」）。
    - ``MISSING``：既无明示也无可靠推断——触发**选择性澄清**，绝不填默认值。

    注意：**刻意不设 ``ASSUMED`` 态**。静默默认值（如规格 0.5 / 主 CQA「总杂质」/
    目标时间点 [24,36]）是篡改用户意图的暗门，在本保真模型中被禁止——缺失即 ``MISSING``。
    """

    STATED = "stated"
    INFERRED = "inferred"
    MISSING = "missing"


class TableType(str, Enum):
    """文档中检测到的表类型（需求 2.2）。"""

    RELATED_SUBSTANCES = "related_substances"
    ASSAY = "assay"
    DISSOLUTION = "dissolution"
    CONTENT_UNIFORMITY = "content_uniformity"
    PHYSICAL_PROPERTIES = "physical_properties"
    STABILITY_TIME_SERIES = "stability_time_series"
    UNKNOWN = "unknown"


#: 澄清原因码（需求 6）：无 LLM / 意图未知 / 数据与意图不匹配。
CLARIFY_NO_LLM = "no_llm"
CLARIFY_INTENT_UNKNOWN = "intent_unknown"
CLARIFY_DATA_INTENT_MISMATCH = "data_intent_mismatch"


def _coerce_enum(enum_cls, value, default):
    """把字符串 / 枚举安全归一为 ``enum_cls`` 成员；无法识别回退 ``default``。"""
    if isinstance(value, enum_cls):
        return value
    try:
        return enum_cls(str(value))
    except ValueError:
        return default


# ---------------------------------------------------------------------------
# 文档画像
# ---------------------------------------------------------------------------

@dataclass
class DetectedTable:
    """文档中检测到的单张表（需求 2.1 / 2.2 / 2.3）。"""

    table_type: TableType
    title: str = ""
    source_document: str = ""          # 多文档聚合：记录来源文件名（需求 2.2）
    is_stability_time_series: bool = False
    columns: list[str] = field(default_factory=list)
    sample_rows: list[list[str]] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "table_type": self.table_type.value,
            "title": self.title,
            "source_document": self.source_document,
            "is_stability_time_series": bool(self.is_stability_time_series),
            "columns": list(self.columns),
            "sample_rows": [list(r) for r in self.sample_rows],
        }

    @staticmethod
    def from_dict(d: dict) -> "DetectedTable":
        d = d or {}
        return DetectedTable(
            table_type=_coerce_enum(TableType, d.get("table_type"), TableType.UNKNOWN),
            title=str(d.get("title", "")),
            source_document=str(d.get("source_document", "")),
            is_stability_time_series=bool(d.get("is_stability_time_series", False)),
            columns=[str(c) for c in (d.get("columns") or [])],
            sample_rows=[[str(c) for c in row] for row in (d.get("sample_rows") or [])],
        )


@dataclass
class DocumentProfile:
    """文档画像：检测到的表集合 + 是否含稳定性时序（需求 2）。"""

    tables: list[DetectedTable] = field(default_factory=list)
    has_stability_time_series: bool = False
    notes: list[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "tables": [t.to_dict() for t in self.tables],
            "has_stability_time_series": bool(self.has_stability_time_series),
            "notes": list(self.notes),
        }

    @staticmethod
    def from_dict(d: dict) -> "DocumentProfile":
        d = d or {}
        return DocumentProfile(
            tables=[DetectedTable.from_dict(t) for t in (d.get("tables") or [])],
            has_stability_time_series=bool(d.get("has_stability_time_series", False)),
            notes=[str(n) for n in (d.get("notes") or [])],
        )


# ---------------------------------------------------------------------------
# 提取项与澄清决策
# ---------------------------------------------------------------------------

@dataclass
class ExtractedItem:
    """单个提取项，必须可回溯到原文（需求 4）。

    - ``source_ref``：可在原文定位的片段 / 单元格内容（回填校验依据，需求 4.1）。
    - ``located``：是否在原文成功定位；``False`` 表示被标记缺失、不得带入捏造值
      （需求 4.2 / Property 1）。
    - ``group``：分组维度（如 ``{"strength": "20μg", "batch": "B1", "attribute": "总杂"}``），
      供描述性梳理按规格/批次/属性组织（需求 7.2）。
    """

    field: str
    value: str = ""
    source_ref: str = ""
    located: bool = True
    group: dict = field(default_factory=dict)
    source: str = "llm"   # "llm"（模型提取）| "user_edited"（用户手动修改/新增）

    def to_dict(self) -> dict:
        return {
            "field": self.field,
            "value": self.value,
            "source_ref": self.source_ref,
            "located": bool(self.located),
            "group": dict(self.group),
            "source": self.source,
        }

    @staticmethod
    def from_dict(d: dict) -> "ExtractedItem":
        d = d or {}
        return ExtractedItem(
            field=str(d.get("field", "")),
            value=str(d.get("value", "")),
            source_ref=str(d.get("source_ref", "")),
            located=bool(d.get("located", True)),
            group=dict(d.get("group") or {}),
            source=str(d.get("source", "llm")),
        )


@dataclass
class IntentSlot:
    """单个**意图槽位**及其保真元信息（意图保真核心载体）。

    把「意图」从一次性扁平分类，升级为「带来源、带置信、可锁定、缺失即问」的可确认
    结构。每个槽位回答三个保真关键问题：这是用户**说的**还是系统**猜的**（``state`` /
    ``evidence``）？系统有**多确定**（``confidence``）？用户**改得动**吗（``locked``）？

    - ``name``：槽位名（如 ``primary_cqa`` / ``spec_limit`` / ``target_timepoints``）。
    - ``value``：取值；``MISSING`` 态下应为 ``None``（绝不填默认）。
    - ``state``：来源态（:class:`SlotState`）。
    - ``confidence``：0..1 置信度，驱动选择性澄清与确认策略（STATED≈1.0 / INFERRED 居中 / MISSING=0）。
    - ``evidence``：明示 / 推断的原文依据片段（对话接地溯源，供回显）。
    - ``locked``：用户确认 / 编辑后置 ``True``——**任何重推断都不得覆盖**（采纳真实意图的硬保证）。
    - ``affects_result``：是否影响计算结论，驱动「信息价值」澄清决策（关键槽缺失才必问）。
    """

    name: str
    value: Any = None
    state: SlotState = SlotState.MISSING
    confidence: float = 0.0
    evidence: str = ""
    locked: bool = False
    affects_result: bool = True

    def needs_clarification(self) -> bool:
        """该槽位是否应触发澄清：缺失、且影响结论、且未被用户锁定。"""
        return (
            self.state is SlotState.MISSING
            and self.affects_result
            and not self.locked
        )

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "value": self.value,
            "state": self.state.value,
            "confidence": float(self.confidence),
            "evidence": self.evidence,
            "locked": bool(self.locked),
            "affects_result": bool(self.affects_result),
        }

    @staticmethod
    def from_dict(d: dict) -> "IntentSlot":
        d = d or {}
        return IntentSlot(
            name=str(d.get("name", "")),
            value=d.get("value"),
            state=_coerce_enum(SlotState, d.get("state"), SlotState.MISSING),
            confidence=float(d.get("confidence", 0.0) or 0.0),
            evidence=str(d.get("evidence", "")),
            locked=bool(d.get("locked", False)),
            affects_result=bool(d.get("affects_result", True)),
        )


@dataclass
class ClarificationDecision:
    """澄清决策（需求 6）：是否需要向用户澄清及其原因。"""

    needed: bool = False
    reason_code: str = ""              # CLARIFY_* 之一
    message_key: str = ""             # i18n key（界面提示文案）
    options: list[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "needed": bool(self.needed),
            "reason_code": self.reason_code,
            "message_key": self.message_key,
            "options": list(self.options),
        }

    @staticmethod
    def from_dict(d: dict) -> "ClarificationDecision":
        d = d or {}
        return ClarificationDecision(
            needed=bool(d.get("needed", False)),
            reason_code=str(d.get("reason_code", "")),
            message_key=str(d.get("message_key", "")),
            options=[str(o) for o in (d.get("options") or [])],
        )


# ---------------------------------------------------------------------------
# 分析任务单
# ---------------------------------------------------------------------------

@dataclass
class AnalysisTaskSheet:
    """《分析任务单》：理解层的强结构化产物，经用户确认后驱动分析（需求 5）。

    字段：
    - ``intent``：分类意图（需求 3.5）。
    - ``document_profile``：文档画像（需求 2）。
    - ``extracted_items``：来源可定位的提取项（需求 4）。
    - ``proposed_skill_id``：建议路由的 Skill id（用户可编辑，需求 5.3 / 8 路由）。
    - ``missing_items`` / ``ambiguous_items``：缺失与歧义清单（需求 4.5）。
    - ``clarification``：澄清决策；``None`` 表示无需澄清（需求 6）。
    - ``source_method``：``"llm"`` | ``"clarification"``，标注任务单来源。
    """

    intent: Intent = Intent.UNKNOWN
    document_profile: DocumentProfile = field(default_factory=DocumentProfile)
    extracted_items: list[ExtractedItem] = field(default_factory=list)
    proposed_skill_id: str = ""
    missing_items: list[str] = field(default_factory=list)
    ambiguous_items: list[str] = field(default_factory=list)
    clarification: Optional[ClarificationDecision] = None
    source_method: str = "llm"
    #: 意图槽位（保真核心）：带来源态 / 置信度 / 锁定的可确认意图维度（需求 3 / 5）。
    #: 向后兼容默认空列表——旧任务单与未启用槽位的技能不受影响。
    intent_slots: list[IntentSlot] = field(default_factory=list)

    def needs_clarification(self) -> bool:
        """是否处于澄清状态（澄清决策存在且 needed=True）。"""
        return self.clarification is not None and bool(self.clarification.needed)

    def slots_needing_clarification(self) -> list[IntentSlot]:
        """返回应触发澄清的关键意图槽位（缺失、影响结论、未锁定）。

        供选择性澄清（信息价值驱动）使用：只对真正影响结果且不确定的槽位发问，
        避免过度澄清。返回空列表表示无需就意图槽位追问。
        """
        return [s for s in self.intent_slots if s.needs_clarification()]

    def get_slot(self, name: str) -> Optional[IntentSlot]:
        """按名取意图槽位；不存在返回 ``None``。"""
        for s in self.intent_slots:
            if s.name == name:
                return s
        return None

    def to_dict(self) -> dict:
        return {
            "intent": self.intent.value,
            "document_profile": self.document_profile.to_dict(),
            "extracted_items": [it.to_dict() for it in self.extracted_items],
            "proposed_skill_id": self.proposed_skill_id,
            "missing_items": list(self.missing_items),
            "ambiguous_items": list(self.ambiguous_items),
            "clarification": self.clarification.to_dict() if self.clarification else None,
            "source_method": self.source_method,
            "intent_slots": [s.to_dict() for s in self.intent_slots],
        }

    @staticmethod
    def from_dict(d: dict) -> "AnalysisTaskSheet":
        d = d or {}
        clar = d.get("clarification")
        return AnalysisTaskSheet(
            intent=_coerce_enum(Intent, d.get("intent"), Intent.UNKNOWN),
            document_profile=DocumentProfile.from_dict(d.get("document_profile") or {}),
            extracted_items=[ExtractedItem.from_dict(it) for it in (d.get("extracted_items") or [])],
            proposed_skill_id=str(d.get("proposed_skill_id", "")),
            missing_items=[str(m) for m in (d.get("missing_items") or [])],
            ambiguous_items=[str(a) for a in (d.get("ambiguous_items") or [])],
            clarification=ClarificationDecision.from_dict(clar) if clar else None,
            source_method=str(d.get("source_method", "llm")),
            intent_slots=[IntentSlot.from_dict(s) for s in (d.get("intent_slots") or [])],
        )


__all__ = [
    "Intent",
    "SlotState",
    "TableType",
    "CLARIFY_NO_LLM",
    "CLARIFY_INTENT_UNKNOWN",
    "CLARIFY_DATA_INTENT_MISMATCH",
    "DetectedTable",
    "DocumentProfile",
    "ExtractedItem",
    "IntentSlot",
    "ClarificationDecision",
    "AnalysisTaskSheet",
]