BIDI_EXTRACTION_AI/BiddingKG/dl/pipeline/context.py

# -*- coding: utf-8 -*-
"""PipelineContext 数据结构骨架。

按 ARCHITECTURE.md §6.1 定义，所有 Stage 共享一个上下文对象，
替代当前 extract.py::predict() 中函数间传递的大量散列变量。

Phase 2 状态：
    本文件仅提供 dataclass 骨架，不接入 extract.py。
    Phase 4/5/6 拆分 extract.py 时，各 Stage 会读写此 context。

约束（ARCHITECTURE.md §5.4）：
    - AI 规则代码不得修改 PipelineContext 中非本规则声明的字段。
    - 核心代码通过 Pipeline 编排顺序传递结果，Predictor 之间不互相 import。

Python 2/3 兼容说明：
    本项目代码风格显示仍兼容 Python 2（见 dl/__init__.py 中的
    sys.path 处理和部分旧文件编码声明）。但 dataclass 是 Python 3.7+ 特性。
    现有代码中已有 3.7+ 特性（如 f-string），因此本文件使用 dataclass。
    若需在 Python 2 下运行，可改为普通类 + __init__。
"""
from __future__ import absolute_import

try:
    from dataclasses import dataclass, field
    _HAS_DATACLASS = True
except ImportError:  # pragma: no cover - Python < 3.7
    _HAS_DATACLASS = False

__all__ = ["PipelineContext"]


if _HAS_DATACLASS:

    @dataclass
    class PipelineContext(object):
        """Pipeline 各 Stage 之间传递的共享上下文。

        字段对应 extract.py::predict() 当前散落的全局/局部变量：
            - 输入字段：doc_id / html / title / page_time / web_source_no 等
            - 预处理产物：articles / sentences / entities / outlines
            - 模型产物：model_results（按 predictor name 分桶）
            - 规则产物：rule_results（按 rule name 分桶）
            - 装配产物：assembled（包-角色-金额-联系人结构）
            - 最终结果：result（输出 JSON）
            - 性能数据：cost_time（各 stage 耗时）

        Stage 接口约定（Phase 4+ 落地）::

            class Stage:
                name: str
                def execute(self, ctx: PipelineContext) -> None: ...
        """

        # --- 输入字段（来自 predict() 入参） ---
        doc_id: str = ""
        html: str = ""
        title: str = ""
        page_time: str = ""
        web_source_no: str = ""
        web_source_name: str = ""
        original_docchannel: str = ""
        page_attachments: str = "[]"

        # --- 预处理产物 ---
        articles: list = field(default_factory=list)
        sentences: list = field(default_factory=list)
        entities: list = field(default_factory=list)
        outlines: list = field(default_factory=list)

        # --- 各层产物 ---
        model_results: dict = field(default_factory=dict)
        rule_results: dict = field(default_factory=dict)
        assembled: dict = field(default_factory=dict)
        result: dict = field(default_factory=dict)
        cost_time: dict = field(default_factory=dict)

else:  # pragma: no cover - Python 2 回退

    class PipelineContext(object):
        """PipelineContext（Python 2 兼容版，字段同上）。"""

        def __init__(self):
            self.doc_id = ""
            self.html = ""
            self.title = ""
            self.page_time = ""
            self.web_source_no = ""
            self.web_source_name = ""
            self.original_docchannel = ""
            self.page_attachments = "[]"
            self.articles = []
            self.sentences = []
            self.entities = []
            self.outlines = []
            self.model_results = {}
            self.rule_results = {}
            self.assembled = {}
            self.result = {}
            self.cost_time = {}