大数据

数据治理pipeline-demo

Posted by

文件结构

core/base_operator.py

"""
操作符基类 - 包含核心自动映射机制
"""
from typing import Any, Dict, TypeVar, Generic, Optional
from core.context import ExecutionState, ResourceBundle
from core.modeling import (
    OperatorInputModel,
    OperatorOutputModel,
    resource_fields_for_model, OperatorConfigModel,
)

# 泛型定义
InputT = TypeVar("InputT", bound=OperatorInputModel)
ConfigT = TypeVar("ConfigT", bound=OperatorConfigModel)
OutputT = TypeVar("OutputT", bound=OperatorOutputModel)


class OperatorBase(Generic[InputT, ConfigT, OutputT]):
    """
    所有具体算子的基类
    核心机制（使用排除法识别资源字段）：
        1. 调用resource_fields_for_model(Input)获取所有资源字段
        2. 自动将这些字段从Input映射到State
        3，处理完成后，从State自动映射会Output
    数据流向：
        Input（含RuntimeConfig）-> _state_from_input -> ExecutionState
        -> 算子处理 -> _output_from_state -> Output（含RuntimeConfig）
    """

    # ==================== 类变量（子类必须覆盖） ====================
    operator_id: str = "base_operator"
    name: str = "BaseOperator"
    description: str = "Base Operator for all pipeline operators"
    version: str = "1.0.0"
    operator_stage: str = ""
    supports_ray: bool = False

    # 子类必须定义这三个变量
    Input: Optional[type[InputT]] = None
    Config: Optional[type[ConfigT]] = None
    Output: Optional[type[OutputT]] = None

    # ==================== 初始化 ====================
    def __init__(self, config: Optional[ConfigT] = None):
        """初始化算子，保存配置"""
        self.config = config or {}
        self.params = self.config

    # ==================== 核心抽象方法 ====================
    def run(self, input_data: InputT) -> OutputT:
        """算子执行入口（子类必须实现）"""
        raise NotImplementedError(
            f"{self.__class__.__name__} must implement run(input) -> output"
        )

    # ==================== 自动映射机制（核心！） ====================
    def _state_from_input(self, input_model: InputT) -> "ExecutionState":
        """
        从输入模型中创建工作台，并自动注入资源数据
        【关键步骤】
            1. 从input_model.runtime获取RuntimeConfig
            2. 调用ExecutionState.from_runtime创建动态工作台
            3. 调用 resource_fields_for_model 获取所有资源字段
            4. 将每个资源字段的数据从Input映射到State
        """
        # 1. 从RuntimeConfig创建ExecutionState
        state = ExecutionState.from_runtime(input_model.runtime)
        # 2. 继承sample_reports（原始项目逻辑）
        state.sample_reports = dict(input_model.sample_reports or {})
        # 3. 处理resource_bundles（如果有传入）
        for name, payload in (input_model.resource_bundles or {}).items():
            bundle = ResourceBundle(
                name=name,
                layer=payload.get("layer", "unknown"),
                records=payload.get("records", []),
                dir=payload.get("dir", ""),
                artifacts=payload.get("artifacts", {}),
                meta=payload.get("meta", {}),
                schema_name=payload.get("schema_name", name),
                schema_version=payload.get("schema_version", "v1"),
            )
            state.resources[name] = bundle
        # 4. 自动映射：使用排除法识别所有资源字段
        for field_name in resource_fields_for_model(self.Input):
            field_value = getattr(input_model, field_name, None)
            if field_value is not None:
                records = field_value if isinstance(field_value, list) else [field_value]
                # 检查是否存在资源包，如果存在则合并
                existing = state.resources.get(field_name)
                if existing is not None:
                    # 更新已存在的资源包
                    state.set_resource_bundle(
                        name=field_name,
                        layer=existing.layer,
                        records=records,
                        dir=existing.dir,
                        artifacts=existing.artifacts,
                        meta=existing.meta,
                        schema_name=existing.schema_name,
                        schema_version=existing.schema_version,
                    )
                else:
                    # 创建新的资源记录
                    state.set_records(field_name, records)
        return state

    def _output_from_state(self, state: ExecutionState) -> OutputT:
        """
        从工作台中提取数据，打包成输出模型
        【关键步骤】
            1. 调用state.to_runtime() 将ExecutionState转会RuntimeConfig
            2. 构建输出字典的基础骨架
            3. 使用排除法获取Output的所有资源字段
            4. 自动从State中取出数据填入Output
        """
        # 1. 构建基础骨架
        payload = {
            "runtime": state.to_runtime(),
            "resource_bundles": {
                name: {
                    "name": bundle.name,
                    "layer": bundle.layer,
                    "records": bundle.records,
                    "dir": bundle.dir,
                    "artifacts": bundle.artifacts,
                    "meta": bundle.meta,
                    "schema_name": bundle.schema_name,
                    "schema_version": bundle.schema_version,
                }
                for name, bundle in state.resources.items()
            },
            "sample_reports": state.get_sample_report_store()
        }

        # 2. 自动映射：使用排除法获取Output的所有资源字段
        if self.Output is not None:
            for field_name in resource_fields_for_model(self.Output):
                payload[field_name] = state.get_records(field_name)
        else:
            # 容错：如果未定义Output，至少把最常用的source_files带上
            payload["source_files"] = state.get_records("source_files")

        # 3. 使用pydantic验证并返回
        if self.Output is not None:
            return self.Output.model_validate(payload)
        else:
            return payload

    # ==================== 辅助方法 ====================
    def input_resource_names(self) -> list[str]:
        """获取输入模型中的所有资源字段名称"""
        if self.Input is not None:
            return []
        return resource_fields_for_model(self.Input)

    def output_resource_names(self) -> list[str]:
        """获取输出模型中的所有资源字段名称"""
        if self.Output is not None:
            return []
        return resource_fields_for_model(self.Output)

core/context.py

"""
执行状态管理 - 动态工作台
"""
import os
from typing import Any, Dict, List, Optional
from core.modeling import RuntimeConfig


class ResourceBundle:
    """
    资源包（桶）- 存储一批数据及其元数据

    字段说明：
        name:         资源名称（如 "source_files"）
        layer:        资源层级（"input" / "corpus" / "samples"）
        records:      具体数据列表
        dir:          数据在磁盘上的存储目录
        artifacts:    额外产物（如 manifest 文件路径）
        meta:         元数据（如记录数量）
        schema_name:  数据模型名称（用于序列化/反序列化）
        schema_version: Schema 版本
    """

    def __init__(
            self,
            name: str,
            layer: str = "unknown",
            records: Optional[List[Any]] = None,
            dir: str = "",
            artifacts: Optional[Dict[str, Any]] = None,
            meta: Optional[Dict[str, Any]] = None,
            schema_name: str = "",
            schema_version: str = "v1",
    ):
        self.name = name
        self.layer = layer
        self.records = list(records or [])
        self.dir = dir
        self.artifacts = dict(artifacts or {})
        self.meta = dict(meta or {})
        self.schema_name = schema_name or name
        self.schema_version = schema_version


class ExecutionState:
    """
    执行状态（动态工作台）
    核心职责：
        1. 从RuntimeConfig构建（通过 from_runtime 工厂方法）
        2. 管理所有算子产生的数据（通过 resources 字典）
        3. 提供统一的目录访问接口
        4. 可以回转RuntimeConfig（通过 to_runtime 方法）
    数据流向：
        RuntimeConfig -> ExecutionState（算子内部使用） -> RuntimeConfig（输出打包）
    """

    def __init__(
            self,
            output_root: str,
            corpus_version: str = "1.0",
            metadata: Optional[Dict[str, Any]] = None,
    ):
        self.output_root = output_root
        self.corpus_version = corpus_version
        self.metadata = dict(metadata or {})
        # 大仓库：key是资源名，value是ResourceBundle对象
        self.resources: Dict[str, ResourceBundle] = {}
        self.sample_reports: Dict[str, Any] = {}

    @classmethod
    def from_runtime(cls, runtime: RuntimeConfig) -> "ExecutionState":
        """
        从RuntimeConfig创建ExecutionState
        核心工厂方法：
            1. 提取 output_root、corpus_version、metadata
            2. 生成一个全新的 ExecutionState 实例
        """
        return cls(
            output_root=runtime.output_root,
            corpus_version=runtime.corpus_version,
            metadata=runtime.metadata,
        )

    def to_runtime(self) -> RuntimeConfig:
        """
        将ExecutionState转回RuntimeConfig
        用于输出打包：将动态状态中的静态配置部分提取出来
        """
        return RuntimeConfig(
            output_root=self.output_root,
            corpus_version=self.corpus_version,
            metadata=dict(self.metadata),
        )

    # ==================== 目录管理 ====================
    def corpus_root(self) -> str:
        """语料根目录：{output_root}/corpus"""
        # 支持子目录
        subdir = str(self.metadata.get("corpus_subdir", "") or "").strip().strip("/\\")
        if subdir:
            return os.path.join(self.output_root, "corpus", subdir)
        return os.path.join(self.output_root, "corpus")

    def corpus_path(self, *parts: str) -> str:
        """获取语料子路径"""
        return os.path.join(self.corpus_root(), *parts)

    def stage_root(self) -> str:
        """暂存区目录：{output_root}/corpus/stage"""
        return os.path.join(self.corpus_root(), "stage")

    def final_root(self) -> str:
        """最终输出目录：{output_root}/corpus/final"""
        return os.path.join(self.corpus_root(), "final")

    def cache_root(self) -> str:
        """缓存根目录：{output_root}/corpus/.cache"""
        return os.path.join(self.corpus_root(), ".cache")

    def cache_path(self, *parts: str) -> str:
        """获取缓存子路径"""
        return os.path.join(self.cache_root(), *parts)

    def corpus_shared_root(self) -> str:
        """共享语料根目录"""
        return os.path.join(self.output_root, "corpus")

    def corpus_shared_path(self, *parts: str) -> str:
        """获取共享语料子路径"""
        return os.path.join(self.corpus_shared_root(), *parts)

    # ==================== 资源管理（核心） ====================
    def get_resource_bundle(self, name: str) -> ResourceBundle:
        """
        从仓库中取出一个“桶”
        如果不存在，返回一个空桶（而不是抛出异常）
        """
        if name not in self.resources:
            return ResourceBundle(
                name=name,
                layer=self._infer_resource_layer(name),
                schema_name=name
            )
        return self.resources[name]

    def set_resource_bundle(
            self,
            name: str,
            *,
            layer: str,
            records: List[Any],
            dir: str = "",
            artifacts: Optional[Dict[str, Any]] = None,
            meta: Optional[Dict[str, Any]] = None,
            schema_name: Optional[str] = None,
            schema_version: str = "v1",
    ) -> None:
        """
        把一个“桶”放进仓库
        这是最底层的存储方法，通常不直接调用
        """
        self.resources[name] = ResourceBundle(
            name=name,
            layer=layer,
            records=records,
            dir=dir or self._infer_resource_dir(name),
            artifacts=artifacts,
            meta=meta,
            schema_name=schema_name,
            schema_version=schema_version,
        )

    def set_records(
            self,
            name: str,
            records: List[Any],
            *,
            layer: Optional[str] = None,
            dir: str = "",
            artifacts: Optional[Dict[str, Any]] = None,
            meta: Optional[Dict[str, Any]] = None,
            schema_name: Optional[str] = None,
            schema_version: str = "v1",
    ) -> None:
        """
        将数据存进仓库（上层算子调用的主入口）
        这是“放数据”的核心方法
            1. 自动根据name推断layer（输入层/语料层）
            2. 自动推断物理存储目录
            3. 打包成ResourceBundle存入self.resources
        """
        self.set_resource_bundle(
            name=name,
            layer=layer or self._infer_resource_layer(name),
            records=records,
            dir=dir or self._infer_resource_dir(name),
            artifacts=artifacts,
            meta=meta or {"record_count": len(records)},
            schema_name=schema_name or name,
            schema_version=schema_version,
        )

    def get_records(self, name: str) -> List[Any]:
        """
        从仓库中取出数据（下游算子调用的主入口）
        这是“取数据”的核心方法
            1. 从self.resources中取出ResourceBundle
            2. 返回其中的 records 列表
            3. 如果不存在，返回空列表（优雅降级）
        """
        bundle = self.get_resource_bundle(name)
        return list(bundle.records or [])

    def resource_names(self):
        """获取所有资源名称"""
        return sorted(self.resources.keys())

    # ==================== 辅助方法 ====================
    def _infer_resource_layer(self, name: str) -> str:
        """推断资源层级"""
        layer_map = {
            "source_files": "input",
            "processed_files": "corpus",
            "final_files": "corpus",
        }
        return layer_map.get(name, "unknown")

    def _infer_resource_dir(self, name: str) -> str:
        """
        自动推到数据应该放在哪个物理目录
        避免每个算子都手动传dir参数
        """
        mapping = {
            "source_files": os.path.join(self.corpus_root(), "source_files"),
            "processed_files": os.path.join(self.corpus_root(), "processed_files"),
            "final_files": self.final_root(),
        }
        return mapping.get(name, self.corpus_root())

    def get_sample_report_store(self) -> Dict[str, Any]:
        """获取样本报告存储"""
        return self.sample_reports

    def set_sample_report(self, group_name: str, report_name: str, payload: Dict[str, Any]) -> None:
        """设置样本报告"""
        report = self.sample_reports.get(group_name)
        if not isinstance(report, dict):
            report = {}
            self.sample_reports[group_name] = report
        report[report_name] = dict(payload or {})

core/modeling.py

"""
模型基类 - 定义RuntimeConfig和算子输入/输出积累
"""
import os
import tempfile
from typing import Any, Dict
from pydantic import BaseModel, Field, model_validator, ConfigDict


class SchemaModel(BaseModel):
    """
    所有模型的基类（支持宽松模式）
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,  # 允许任意类型
        extra="allow",  # 允许额外字段
        populate_by_name=True,  # 按名称填充
        validate_assignment=True,  # 验证赋值
    )

    schema_version: str = "v1"


class RuntimeConfig(SchemaModel):
    """
    运行时配置（静态配置载体）
    职责：
        1. 存储调用方传入的配置参数
        2. 测试时自动重定向输出目录
        3. 作为Input的一部分传入算子
    注意：RuntimeConfig 是只读的，算子不应该修改它
    """
    output_root: str = "./output"
    corpus_version: str = "v1.0"
    metadata: Dict[str, Any] = Field(default_factory=dict)

    @model_validator(mode="after")
    def _redirect_test_default_output(self) -> "RuntimeConfig":
        """测试时自动重定向到临时目录（"""
        if (
                self.output_root == "./outputs"
                and os.getenv("PYTEST_CURRENT_TEST")
        ):
            object.__setattr__(
                self,
                "output_root",
                tempfile.mkdtemp(prefix="pipeline-test-"),
            )
        return self


class OperatorInputModel(SchemaModel):
    """
    算子的输入模型基类
    关键设计：
        runtime: 类型是RuntimeConfig（静态配置）
        resource_bundles: 存储的是“桶”的序列化表示
        sample_reports: 用于存放调试/系统信息
        其他字段（如 source_files）自动被识别为资源字段
    """
    runtime: RuntimeConfig = Field(default_factory=RuntimeConfig)
    resource_bundles: Dict[str, Any] = Field(default_factory=dict)
    sample_reports: Dict[str, Any] = Field(default_factory=dict)


class OperatorOutputModel(SchemaModel):
    """
    算子的输出模型基类
    同样包含runtime（RuntimeConfig 类型）
    但实际运行时，BaseOperator 会从ExecutionState中转回RuntimeConfig
    """
    runtime: RuntimeConfig = Field(default_factory=RuntimeConfig)
    resource_bundles: Dict[str, Any] = Field(default_factory=dict)
    sample_reports: Dict[str, Any] = Field(default_factory=dict)


class OperatorConfigModel(SchemaModel):
    """
    算子的配置模型基类

    【重要】所有控制算子行为的参数（如 input_dir、max_retry、threshold 等）
    都必须定义在 Config 模型中，而不是 Input 模型中！
    """
    model_config = ConfigDict(
        arbitrary_types_allowed=True,  # 允许任意类型
        extra="allow",  # 允许额外字段
        populate_by_name=True,  # 按名称填充
        validate_assignment=True,  # 验证赋值
    )


# 核心设计：状态保留字段（这些字段不被视为资源）
# 所有不在这个集合中的字段，都被自动识别为资源字段
_STATE_RESERVED_FIELDS = {
    "runtime",
    "resource_bundles",
    "sample_reports",
    "schema_version",
}


def resource_fields_for_model(model_cls: type[OperatorInputModel] | type[OperatorOutputModel]) -> list[str]:
    """
    获取模型中的资源字段名称
    核心逻辑（排除法）：
        遍历模型的所有字段，跳过保留字段（runtime、resource_bundles、sample_reports、schema_version）
        其余字段全部视为资源字段
    """
    fields = []
    for field_name in model_cls.model_fields:
        # 跳过保留字段
        if field_name in _STATE_RESERVED_FIELDS:
            continue
        # 添加资源字段
        fields.append(field_name)
    return fields

operators/finalize_operator.py

"""
算子3：最终清洗
"""

import os
from typing import Any, Dict, List
from core.base_operator import OperatorBase
from core.modeling import OperatorInputModel, OperatorOutputModel, OperatorConfigModel
from schemas.data_objects import SourceFile


# Config：配置参数
class FinalizeConfig(OperatorConfigModel):
    report_filename: str = "summary_report.txt"
    output_subdir: str = "final"


class FinalizeInput(OperatorInputModel):
    """清洗算子的输入"""
    source_files: List[SourceFile] = []


class FinalizeOutput(OperatorOutputModel):
    """清洗算子的输出"""
    source_files: List[SourceFile] = []


class FinalizeOperator(OperatorBase[FinalizeInput, FinalizeConfig, FinalizeOutput]):
    """最终清洗算子"""
    operator_id = "finalize_operator"
    name = "Finalize"
    description = "Generate summary report from processed files."
    version = "1.0.0"

    Input = FinalizeInput
    Config = FinalizeConfig
    Output = FinalizeOutput

    def run(self, input_data: FinalizeInput) -> FinalizeOutput:
        state = self._state_from_input(input_data)

        files = state.get_records("source_files")
        if not files:
            print("[FinalizeOperator] 没有收到数据，跳过")
            return self._output_from_state(state)
        print(f"[FinalizeOperator] 收到 {len(files)} 个文件")
        final_dir = state.final_root()
        os.makedirs(final_dir, exist_ok=True)
        total_lines = 0
        summary_lines: List[str] = []
        for src in files:
            try:
                with open(src.file_path, "r", encoding="utf-8") as f:
                    line_count = len(f.readlines())
                total_lines += line_count
                summary_lines.append(f"  {os.path.basename(src.file_path)}: {line_count} 行")
            except Exception as e:
                summary_lines.append(f"  {os.path.basename(src.file_path)}: 读取失败 ({e})")
        report_path = os.path.join(final_dir, self.config.report_filename)
        with open(report_path, "w", encoding="utf-8") as f:
            f.write("=" * 60 + "\n")
            f.write("           数据流水线处理报告\n")
            f.write("=" * 60 + "\n\n")
            f.write(f"处理文件总数: {len(files)}\n")
            f.write(f"总行数: {total_lines}\n")
            f.write("\n--- 文件明细 ---\n")
            f.write("\n".join(summary_lines))
            f.write("\n" + "=" * 60 + "\n")
        print(f"[FinalizeOperator] 报告已生成: {report_path}")
        final_files = [
            SourceFile(
                document_id="summary_report",
                file_path=report_path,
                file_type=".txt",
                source_type="final",
                file_hash="",
                metadata={
                    "file_count": len(files),
                    "total_lines": total_lines,
                }
            )
        ]
        state.set_records("source_files", final_files)
        state.metadata["summary_report"] = report_path
        state.metadata["total_lines"] = total_lines
        state.metadata["file_count"] = len(files)
        return self._output_from_state(state)

operators/ingest_operator.py

"""
算子1：源文件接入
"""

import os
import uuid
from typing import Any, List, Dict
from core.base_operator import OperatorBase
from core.modeling import OperatorInputModel, OperatorOutputModel, OperatorConfigModel
from schemas.data_objects import SourceFile


# Config：所有配置参数放在这里！
class IngestConfig(OperatorConfigModel):
    """
    接入算子的配置
    【重要】input_dir 是配置参数，放在 Config 中，
    不会被 _state_from_input 误认为资源字段！
    """
    input_dir: str = "./inputs"  # 要扫描的目录
    file_extension: str = ".txt"  # 要扫描的文件扩展名


class IngestInput(OperatorInputModel):
    """
    接入算子的输入
    注意：这里没有定义任何资源字段（因为本算子自己就是数据源）
    所有的配置参数都在 IngestConfig 中
    """
    pass


class IngestOutput(OperatorOutputModel):
    """
    接入算子的输出
    关键：定义了source_files字段
    这个字段会被resource_fields_for_model自动识别为资源字段
    """
    source_files: List[SourceFile] = []


class IngestOperator(OperatorBase[IngestInput, IngestConfig, IngestOutput]):
    """源文件接入算子"""
    operator_id = "ingest_operator"
    name = "SourceIngest"
    description = "Scan local directory and create SourceFile records."
    version = "1.0.0"

    Input = IngestInput
    Config = IngestConfig
    Output = IngestOutput

    def run(self, input_data: IngestInput) -> IngestOutput:
        # 1. 创建工作台（自动识别资源字段）
        state = self._state_from_input(input_data)
        # 从self.config中读取配置参数
        input_dir = self.config.input_dir
        file_extension = self.config.file_extension
        if not os.path.isdir(input_dir):
            raise FileNotFoundError(f"Input directory not found: {input_dir}")

        # 2. 扫描文件夹
        source_files: List[SourceFile] = []
        for filename in os.listdir(input_dir):
            if not filename.endswith(file_extension):
                continue

            file_path = os.path.join(input_dir, filename)
            if not os.path.isfile(file_path):
                continue

            doc_id = str(uuid.uuid4())
            source_files.append(SourceFile(
                document_id=doc_id,
                file_path=file_path,
                file_type=file_extension,
                source_type="ingest",
                file_hash="",
                metadata={"original_filename": filename},
            ))
        print(f"[IngestOperator] 发现 {len(source_files)} 个文件")
        # 存入state
        state.set_records("source_files", source_files)
        # 返回输出（自动从state提取资源字段）
        return self._output_from_state(state)

operators/process_operator.py

"""
算子2：业务处理
"""

import os
from typing import Any, Dict, List

from core.base_operator import OperatorBase
from core.modeling import OperatorInputModel, OperatorOutputModel, OperatorConfigModel
from schemas.data_objects import SourceFile


# Config：配置参数放这里
class ProcessConfig(OperatorConfigModel):
    filter_keyword: str = "DEBUG"  # 要过滤的关键词
    cache_subdir: str = "processed"  # 缓存子目录


class ProcessInput(OperatorInputModel):
    """
    处理算子的输入
    定义了source_files字段，表示接收上游数据
    这个字段会被自动识别为资源字段
    """
    source_files: List[SourceFile] = [] # 接收上游数据


class ProcessOutput(OperatorOutputModel):
    """
    处理算子的输出
    同样定义了source_files字段
    """
    source_files: List[SourceFile] = [] # 输出处理后的数据


class ProcessOperator(OperatorBase[ProcessInput, ProcessConfig, ProcessOutput]):
    """业务处理算子"""
    operator_id = "process_operator"
    name = "LineFilter"
    description = "Filter out DEBUG lines from source files."
    version = "1.0.0"

    Input = ProcessInput
    Config = ProcessConfig
    Output = ProcessOutput

    def run(self, input_data: ProcessInput) -> ProcessOutput:
        # 1. 创建工作台（自动将input_dta.source_files存入state）
        state = self._state_from_input(input_data)
        # 从self.config中读取配置
        keyword = self.config.filter_keyword
        cache_dir = state.cache_path(self.config.cache_subdir)

        # 2. 从state取出数据
        raw_files = state.get_records("source_files")
        if not raw_files:
            print("[ProcessOperator] 没有收到上游数据，跳过")
            return self._output_from_state(state)
        print(f"[ProcessOperator] 收到 {len(raw_files)} 个待处理文件")
        # 3. 创建缓存目录
        os.makedirs(cache_dir, exist_ok=True)
        # 4. 处理每个文件
        processed_files: List[SourceFile] = []
        for src in raw_files:
            try:
                with open(src.file_path, "r", encoding="utf-8") as f:
                    lines = f.readlines()
            except Exception as e:
                print(f"[ProcessOperator] 读取文件失败 {src.file_path}: {e}")
                continue
            # 过滤 DEBUG 行
            filtered_lines = [line for line in lines if keyword not in line]
            # 写入缓存
            new_filename = f"{src.document_id[:8]}.txt"
            new_path = os.path.join(cache_dir, new_filename)
            with open(new_path, "w", encoding="utf-8") as f:
                f.writelines(filtered_lines)
            processed_files.append(
                SourceFile(
                    document_id=src.document_id,
                    file_path=new_path,
                    file_type=".txt",
                    source_type="processed",
                    file_hash="",
                    metadata={
                        "original_path": src.file_path,
                        "original_lines": len(lines),
                        "filtered_lines": len(filtered_lines),
                        "removed_lines": len(lines) - len(filtered_lines),
                    }
                )
            )
        print(f"[ProcessOperator] 处理完成，生成 {len(processed_files)} 个文件")
        # 5. 存入state（覆盖旧的source_files）
        state.set_records("source_files", processed_files)
        # 6. 返回输出
        return self._output_from_state(state)

schemas/data_objects.py

"""
数据模型层 - 定义流转的最小单元
"""

from pydantic import BaseModel, Field
from typing import Any, Dict


class SourceFile(BaseModel):
    """
    源文件元数据 - 整个流水线的"血液"

    字段说明：
        document_id: 唯一标识符
        file_path:   当前文件在磁盘上的绝对路径
        file_type:   文件扩展名（如 .txt, .md, .pdf）
        source_type: 数据来源标记（ingest / processed / final）
        file_hash:   文件哈希值（用于去重和缓存验证）
        metadata:    灵活字典，存储各算子附加的临时信息
    """
    document_id: str
    file_path: str
    file_type: str
    source_type: str
    file_hash: str = ""
    metadata: Dict[str, Any] = Field(default_factory=dict)

run_pipeline.py

"""
主入口编排器
"""

import argparse
import sys
import time
from pathlib import Path
from core.modeling import RuntimeConfig
from operators.ingest_operator import IngestOperator, IngestInput, IngestConfig
from operators.process_operator import ProcessOperator, ProcessInput, ProcessConfig
from operators.finalize_operator import FinalizeOperator, FinalizeInput, FinalizeConfig


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="数据流水线 Demo：输入 TXT 文件，过滤 DEBUG 行，生成统计报告"
    )
    parser.add_argument("--input-dir", required=True, help="输入文件夹路径")
    parser.add_argument("--output-root", required=True, help="输出根目录")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    start_time = time.perf_counter()
    input_dir = Path(args.input_dir).resolve()
    if not input_dir.is_dir():
        print(f"❌ 错误: 输入目录不存在: {input_dir}")
        return 1
    # ==================== 创建 RuntimeConfig ====================
    runtime = RuntimeConfig(
        output_root=str(Path(args.output_root).resolve()),
        corpus_version="v1.0",
        metadata={
            "pipline_name": "txt_filter_pipeline",
            "input_dir": str(input_dir),
        }
    )
    print(f"📂 RuntimeConfig 已创建")
    print(f"   output_root: {runtime.output_root}")
    print("=" * 60)

    # ==================== 阶段 1：接入 ====================
    print("\n[阶段 1/3] 接入源文件...")
    # 创建Config
    ingest_config = IngestConfig(input_dir=str(input_dir))
    ingest_input = IngestInput(runtime=runtime)
    ingest_op = IngestOperator(config=ingest_config)
    ingest_output = ingest_op.run(ingest_input)
    raw_files = ingest_output.source_files
    print(f"   ✅ 接入完成: 发现 {len(raw_files)} 个文件")
    if not raw_files:
        print("   ⚠️  没有找到 .txt 文件，流水线终止")
        return 0

    # ==================== 阶段 2：处理 ====================
    print("\n[阶段 2/3] 过滤 DEBUG 行...")
    process_config = ProcessConfig(filter_keyword="DEBUG")
    process_input = ProcessInput(runtime=runtime, source_files=raw_files)
    process_op = ProcessOperator(config=process_config)
    process_output = process_op.run(process_input)
    processed_files = process_output.source_files
    print(f"   ✅ 处理完成: 生成 {len(processed_files)} 个缓存文件")
    for src in processed_files:
        removed = src.metadata.get("removed_lines", 0)
        if removed > 0:
            print(f"      - {Path(src.file_path).name}: 移除了 {removed} 行 DEBUG 日志")

    # ==================== 阶段 3：最终清洗 ====================
    print("\n[阶段 3/3] 生成最终报告...")
    finalize_config = FinalizeConfig(report_filename="summary_report.txt")
    finalize_input = FinalizeInput(runtime=runtime, source_files=processed_files)
    finalize_op = FinalizeOperator(config=finalize_config)
    finalize_output = finalize_op.run(finalize_input)
    report_path = finalize_output.runtime.metadata.get("summary_report")
    print(f"   ✅ 报告已生成: {report_path}")

    # ==================== 总结 ====================
    elapsed = time.perf_counter() - start_time
    print("\n" + "=" * 60)
    print("🎉 流水线执行完毕！")
    print(f"   耗时: {elapsed:.2f} 秒")
    print(f"   最终输出目录: {runtime.output_root}/corpus/final/")
    print("=" * 60)

    return 0


if __name__ == '__main__':
    sys.exit(main())
上一篇 类型注解与泛型编程