供应链数据血缘追踪 — 从原始数据到Tag决策的全链路溯源与影响分析

"""
供应链数据血缘追踪系统
功能：血缘图谱构建 / 溯源查询 / 影响分析 / 审计报告
"""
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict, deque
import warnings
warnings.filterwarnings('ignore')


@dataclass
class LineageNode:
    node_id: str
    node_type: str      # DataSource / Transform / Tag / Action
    name: str
    metadata: dict = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))


@dataclass
class LineageEdge:
    from_id: str
    to_id: str
    edge_type: str      # PRODUCES / CONSUMES / TRIGGERS
    transform_info: str = ""


class DataLineageGraph:

    def __init__(self):
        self.nodes: dict = {}
        self.edges: list = []
        self.adj: dict = defaultdict(list)      # 正向
        self.radj: dict = defaultdict(list)     # 反向

    def add_node(self, node: LineageNode):
        self.nodes[node.node_id] = node

    def add_edge(self, edge: LineageEdge):
        self.edges.append(edge)
        self.adj[edge.from_id].append(edge)
        self.radj[edge.to_id].append(edge)

    def trace_upstream(self, node_id: str, max_hops: int = 5) -> list:
        """溯源：追踪某Tag/决策的数据来源"""
        visited, path = set(), []
        queue = deque([(node_id, 0)])
        while queue:
            nid, hop = queue.popleft()
            if nid in visited or hop > max_hops:
                continue
            visited.add(nid)
            node = self.nodes.get(nid)
            if node:
                path.append({"hop": hop, "node": node.name, "type": node.node_type, "id": nid})
            for edge in self.radj[nid]:
                queue.append((edge.from_id, hop + 1))
        return path