데이터 계보

# 데이터 계보 추적 시스템 구현 예제
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Optional
from enum import Enum
import json
import hashlib

class TransformationType(Enum):
    EXTRACTION = "extraction"      # 원본 추출
    CLEANING = "cleaning"          # 정제
    TRANSFORMATION = "transformation"  # 변환
    AGGREGATION = "aggregation"    # 집계
    ANONYMIZATION = "anonymization"  # 익명화/가명화
    AUGMENTATION = "augmentation"  # 데이터 증강
    SPLIT = "split"               # 학습/검증 분할

@dataclass
class LineageNode:
    """계보 그래프의 노드 (데이터셋 또는 변환)"""
    node_id: str
    node_type: str  # "dataset" or "transform"
    name: str
    metadata: Dict = field(default_factory=dict)

@dataclass
class LineageEdge:
    """계보 그래프의 엣지 (데이터 흐름)"""
    source_id: str
    target_id: str
    transform_type: TransformationType
    transform_params: Dict
    timestamp: datetime
    executor: str  # 실행자 또는 시스템

class DataLineageTracker:
    """AI 학습 데이터 계보 추적기"""

    def __init__(self, project_name: str):
        self.project_name = project_name
        self.nodes: Dict[str, LineageNode] = {}
        self.edges: List[LineageEdge] = []

    def register_dataset(self, dataset_id: str, name: str,
                         source: str, record_count: int,
                         schema: Dict = None) -> LineageNode:
        """데이터셋 등록"""
        node = LineageNode(
            node_id=dataset_id,
            node_type="dataset",
            name=name,
            metadata={
                "source": source,
                "record_count": record_count,
                "schema": schema or {},
                "registered_at": datetime.now().isoformat()
            }
        )
        self.nodes[dataset_id] = node
        return node

    def record_transformation(self,
                              source_ids: List[str],
                              target_id: str,
                              transform_type: TransformationType,
                              transform_params: Dict,
                              executor: str = "system") -> str:
        """변환 이력 기록"""
        transform_id = f"transform_{hashlib.md5(f'{source_ids}{target_id}{datetime.now()}'.encode()).hexdigest()[:8]}"

        # 변환 노드 생성
        transform_node = LineageNode(
            node_id=transform_id,
            node_type="transform",
            name=f"{transform_type.value}",
            metadata={"params": transform_params}
        )
        self.nodes[transform_id] = transform_node

        # 엣지 생성 (source -> transform -> target)
        for source_id in source_ids:
            self.edges.append(LineageEdge(
                source_id=source_id,
                target_id=transform_id,
                transform_type=transform_type,
                transform_params=transform_params,
                timestamp=datetime.now(),
                executor=executor
            ))

        self.edges.append(LineageEdge(
            source_id=transform_id,
            target_id=target_id,
            transform_type=transform_type,
            transform_params=transform_params,
            timestamp=datetime.now(),
            executor=executor
        ))

        return transform_id

    def trace_upstream(self, dataset_id: str) -> List[str]:
        """특정 데이터셋의 상위 계보 추적 (출처 찾기)"""
        upstream = []
        visited = set()

        def dfs(node_id):
            if node_id in visited:
                return
            visited.add(node_id)

            for edge in self.edges:
                if edge.target_id == node_id:
                    upstream.append(edge.source_id)
                    dfs(edge.source_id)

        dfs(dataset_id)
        return upstream

    def get_gdpr_impact(self, source_dataset_id: str) -> List[str]:
        """GDPR 삭제권 영향 분석: 특정 원본에서 파생된 모든 데이터셋"""
        downstream = []
        visited = set()

        def dfs(node_id):
            if node_id in visited:
                return
            visited.add(node_id)

            for edge in self.edges:
                if edge.source_id == node_id:
                    if self.nodes[edge.target_id].node_type == "dataset":
                        downstream.append(edge.target_id)
                    dfs(edge.target_id)

        dfs(source_dataset_id)
        return downstream

    def export_for_audit(self) -> str:
        """EU AI Act 감사용 계보 문서 생성"""
        return json.dumps({
            "project": self.project_name,
            "export_timestamp": datetime.now().isoformat(),
            "datasets": {k: {"name": v.name, **v.metadata}
                         for k, v in self.nodes.items()
                         if v.node_type == "dataset"},
            "transformations": [
                {
                    "source": e.source_id,
                    "target": e.target_id,
                    "type": e.transform_type.value,
                    "params": e.transform_params,
                    "timestamp": e.timestamp.isoformat(),
                    "executor": e.executor
                } for e in self.edges
            ]
        }, indent=2, ensure_ascii=False)

# 사용 예시
tracker = DataLineageTracker("고객이탈예측모델")

# 원본 데이터 등록
tracker.register_dataset("raw_customer", "원본 고객 데이터",
                         source="CRM 시스템", record_count=100000)

# 변환 이력 기록
tracker.record_transformation(
    source_ids=["raw_customer"],
    target_id="anonymized_customer",
    transform_type=TransformationType.ANONYMIZATION,
    transform_params={"method": "k-anonymity", "k": 5}
)

# 상위 계보 추적
print(f"학습데이터 출처: {tracker.trace_upstream('anonymized_customer')}")