供应链知识图谱标签传播算法 — LPA/层级继承/关系链传播的Tag扩散引擎

"""
供应链知识图谱标签传播引擎
功能：层级继承传播 / LPA迭代传播 / 关系链传播 / 置信度衰减管理
输入：实体关系图谱 + 种子标签集合
输出：传播后完整标签集 + 传播路径追踪 + 覆盖率提升报告
"""
import numpy as np
import pandas as pd
from collections import defaultdict, deque
from typing import Optional
import warnings
warnings.filterwarnings('ignore')


class SupplyChainTagGraph:
    """供应链实体关系图谱（用于标签传播）"""

    def __init__(self):
        self.entities = {}   # entity_id → {type, name, tags: {tag_id: {value, confidence, source}}}
        self.edges = []      # (src, dst, relation_type, weight)
        self.adj = defaultdict(list)   # src → [(dst, relation, weight)]
        self.radj = defaultdict(list)  # dst → [(src, relation, weight)] 反向

    def add_entity(self, entity_id: str, entity_type: str, name: str, tags: dict = None):
        self.entities[entity_id] = {
            "type": entity_type, "name": name,
            "tags": {k: {"value": v, "confidence": 1.0, "source": "manual"}
                     for k, v in (tags or {}).items()}
        }

    def add_relation(self, src: str, dst: str, relation: str, weight: float = 1.0):
        self.edges.append((src, dst, relation, weight))
        self.adj[src].append((dst, relation, weight))
        self.radj[dst].append((src, relation, weight))

    def get_tag(self, entity_id: str, tag_id: str) -> Optional[dict]:
        e = self.entities.get(entity_id)
        if e and tag_id in e["tags"]:
            return e["tags"][tag_id]
        return None

    def set_tag(self, entity_id: str, tag_id: str, value, confidence: float, source: str):
        if entity_id in self.entities:
            self.entities[entity_id]["tags"][tag_id] = {
                "value": value, "confidence": confidence, "source": source
            }


class TagPropagationEngine:
    """三种传播模式引擎"""

    def __init__(self, graph: SupplyChainTagGraph, decay_alpha: float = 0.85, conf_threshold: float = 0.5):
        self.graph = graph
        self.decay_alpha = decay_alpha
        self.conf_threshold = conf_threshold
        self.propagation_log = []

    def hierarchical_propagation(self, tag_id: str, allowed_relations: list,
                                  max_hops: int = 3) -> int:
        """