SKU跨平台实体统一标识与标签同步 — ASIN/SKU/ERP三码合一的实体对齐与标签一致性保障

"""
SKU跨平台实体统一标识与标签同步
功能：多平台ID映射 / 实体对齐 / 标签冲突解决 / 统一标签视图
输入：各平台SKU数据 + 映射规则
输出：Golden Record / 统一ID映射 / 跨平台一致性标签
"""
import numpy as np
import pandas as pd
import hashlib
import re
from dataclasses import dataclass, field
from typing import Optional
import warnings
warnings.filterwarnings('ignore')


@dataclass
class PlatformID:
    platform: str
    external_id: str
    confidence: float = 1.0
    match_method: str = "manual"  # manual/rule/embedding


@dataclass
class GoldenRecord:
    """黄金记录：跨平台统一实体"""
    internal_id: str
    canonical_name: str
    product_family: str
    brand: str
    platform_ids: list = field(default_factory=list)  # [PlatformID]
    unified_tags: dict = field(default_factory=dict)   # tag_id → {value, source, confidence}
    tag_conflicts: list = field(default_factory=list)  # 冲突记录

    def add_platform_id(self, platform: str, external_id: str,
                         confidence: float = 1.0, method: str = "manual"):
        self.platform_ids.append(PlatformID(platform, external_id, confidence, method))

    def merge_tag(self, tag_id: str, value, source: str, confidence: float,
                   priority_order: list = None):
        """合并标签（处理冲突）"""
        priority_order = priority_order or ["erp", "amazon", "shopify", "tiktok", "manual"]

        existing = self.unified_tags.get(tag_id)
        if existing is None:
            self.unified_tags[tag_id] = {
                "value": value, "source": source,
                "confidence": confidence, "all_sources": {source: value}
            }
        else:
            # 记录所有来源的值
            existing["all_sources"][source] = value

            # 检测冲突（值不同）
            if existing["value"] != value:
                # 按优先级选择
                existing_priority = priority_order.index(existing["source"]) \
                    if existing["source"] in priority_order else 999
                new_priority = priority_order.index(source) \