基于GEO优化的智能内容分发与效果追踪系统

┌─────────────────────────────────────────────────────────┐
│                   内容分发与追踪系统                       │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────┐  │
│  │ 内容管理  │───│ 分发调度器    │───│ 渠道适配器    │  │
│  │ (CMS)    │    │ (Scheduler)  │    │ (Adapters)   │  │
│  └──────────┘    └──────┬───────┘    └──────┬───────┘  │
│                         │                    │          │
│                   ┌─────▼───────┐    ┌──────▼───────┐  │
│                   │ UTM生成器    │    │ 短链接服务    │  │
│                   │ (UTM Gen)   │    │ (Short URL)  │  │
│                   └─────────────┘    └──────┬───────┘  │
│                                             │          │
│                                    ┌────────▼────────┐ │
│                                    │   用户点击       │ │
│                                    │   (Click)       │ │
│                                    └────────┬────────┘ │
│                                             │          │
│                          ┌──────────────────┼───────┐  │
│                          ▼                  ▼       ▼  │
│                   ┌───────────┐  ┌──────────┐ ┌──────┐│
│                   │ 点击日志   │  │ 会话追踪  │ │转化  ││
│                   │ (Logs)    │  │(Session) │ │跟踪  ││
│                   └─────┬─────┘  └────┬─────┘ └──┬───┘│
│                         │             │           │    │
│                         └──────┬──────┴───────────┘    │
│                                ▼                        │
│                       ┌────────────────┐                │
│                       │   归因引擎      │                │
│                       │ (Attribution)  │                │
│                       └───────┬────────┘                │
│                               │                         │
│                               ▼                         │
│                       ┌────────────────┐                │
│                       │   数据看板      │                │
│                       │  (Dashboard)   │                │
│                       └────────────────┘                │
└─────────────────────────────────────────────────────────┘

"""
geo_distribution_tracker.py - GEO智能内容分发与效果追踪系统
技术栈: Python / dataclasses / enum / typing / hashlib / datetime / json / uuid
场景: GEO内容多渠道分发、UTM追踪、点击日志、转化归因、数据看板
"""
import os
import json
import time
import uuid
import hashlib
import logging
import re
from typing import List, Dict, Optional, Any, Tuple, Set
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
from collections import defaultdict
from urllib.parse import urlencode, urlparse, parse_qs

# ==================== 模块1: 数据模型定义 ====================

class ChannelType(Enum):
    """分发渠道类型"""
    WECHAT = "wechat"
    WEIBO = "weibo"
    ZHIHU = "zhihu"
    DOUYIN = "douyin"
    BILIBILI = "bilibili"
    EMAIL = "email"
    SMS = "sms"
    SEO = "seo"
    SEM = "sem"
    AFFILIATE = "affiliate"
    DIRECT = "direct"


class ContentStatus(Enum): 5030.baike.tongsou.com 
    """内容状态"""
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"


class AttributionModel(Enum):
    """归因模型"""
    LAST_CLICK = "last_click"
    FIRST_CLICK = "first_click"
    LINEAR = "linear"
    TIME_DECAY = "time_decay"
    POSITION_BASED = "position_based"


@dataclass
class Content: 5034.baike.tongsou.com 
    """内容模型"""
    content_id: str
    title: str
    body: str
    content_type: str  # article / video / infographic
    tags: List[str]
    status: ContentStatus
    created_at: str
    updated_at: str
    author: str
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class DistributionChannel: 5038.baike.tongsou.com 
    """分发渠道"""
    channel_id: str
    name: str
    channel_type: ChannelType
    config: Dict[str, Any]  # 渠道特定配置（API密钥等）
    is_active: bool = True
    priority: int = 0


@dataclass
class UTMCampaign: 5052.baike.tongsou.com 
    """UTM活动配置"""
    campaign_id: str
    utm_source: str
    utm_medium: str
    utm_campaign: str
    utm_term: str = ""
    utm_content: str = ""


@dataclass
class TrackingLink:
    """追踪链接"""
    link_id: str
    original_url: str
    short_url: str
    content_id: str
    channel_id: str
    utm_params: Dict[str, str]
    created_at: str
    expires_at: Optional[str] = None
    click_count: int = 0
    conversion_count: int = 0


@dataclass
class ClickEvent:
    """点击事件"""
    event_id: str
    link_id: str
    content_id: str
    channel_id: str
    user_id: str
    session_id: str
    ip_address: str
    user_agent: str
    referrer: str
    timestamp: str
    geo_location: Dict[str, str] = field(default_factory=dict)


@dataclass
class ConversionEvent: 5055.baike.tongsou.com 
    """转化事件"""
    conversion_id: str
    event_type: str  # signup / purchase / download / lead
    user_id: str
    session_id: str
    value: float
    currency: str = "CNY"
    timestamp: str = ""
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class AttributionResult: 5057.baike.tongsou.com 
    """归因结果"""
    session_id: str
    touchpoints: List[Dict[str, Any]]
    attributed_channel: str
    attributed_value: float
    model: str


# ==================== 模块2: 渠道管理 ====================

class ChannelManager: 6004.baike.tongsou.com 
    """
    渠道管理器
    管理所有分发渠道的配置与状态
    """

    def __init__(self): 6007.baike.tongsou.com 
        self.channels: Dict[str, DistributionChannel] = {}

    def add_channel(self, name: str, channel_type: ChannelType,
                    config: Dict[str, Any] = None, priority: int = 0) -> str:
        """添加渠道"""
        channel_id = hashlib.md5(f"{name}_{channel_type.value}".encode()).hexdigest()[:12]
        channel = DistributionChannel(
            channel_id=channel_id,
            name=name,
            channel_type=channel_type,
            config=config or {},
            is_active=True,
            priority=priority,
        )
        self.channels[channel_id] = channel
        return channel_id

    def remove_channel(self, channel_id: str) -> bool: 6008.baike.tongsou.com 
        """移除渠道"""
        if channel_id in self.channels: 14009.baike.tongsou.com 
            del self.channels[channel_id]
            return True
        return False

    def activate_channel(self, channel_id: str) -> bool:
        """激活渠道"""
        if channel_id in self.channels:
            self.channels[channel_id].is_active = True
            return True
        return False

    def deactivate_channel(self, channel_id: str) -> bool: 14068.baike.tongsou.com 
        """停用渠道"""
        if channel_id in self.channels: 14013.baike.tongsou.com 
            self.channels[channel_id].is_active = False
            return True
        return False

    def get_active_channels(self) -> List[DistributionChannel]:
        """获取所有活跃渠道，按优先级排序"""
        active = [ch for ch in self.channels.values(14070.baike.tongsou.com) if ch.is_active]
        return sorted(active, key=lambda x: x.priority, reverse=True)

    def get_channel(self, channel_id: str) -> Optional[DistributionChannel]:
        """获取单个渠道"""
        return self.channels.get(channel_id)

    def list_all(self) -> List[Dict[str, Any]]: 14071.baike.tongsou.com 
        """列出所有渠道"""
        return [
            {
                "channel_id": ch.channel_id,
                "name": ch.name,
                "type": ch.channel_type.value,
                "is_active": ch.is_active,
                "priority": ch.priority,
            }
            for ch in self.channels.values(14080.baike.tongsou.com)
        ]


# ==================== 模块3: UTM参数生成器 ====================

class UTMGenerator:
    """
    UTM参数生成器
    为每个分发链接生成标准UTM追踪参数
    """

    def __init__(self): 14082.baike.tongsou.com 
        self.campaigns: Dict[str, UTMCampaign] = {}

    def create_campaign(self, source: str, medium: str, campaign: str,
                        term: str = "", content: str = "") -> str:
        """创建UTM活动"""
        campaign_id = hashlib.md5(
            f"{source}_{medium}_{campaign}".encode(14083.baike.tongsou.com)
        ).hexdigest()[:12]
        utm = UTMCampaign(
            campaign_id=campaign_id,
            utm_source=source,
            utm_medium=medium,
            utm_campaign=campaign,
            utm_term=term,
            utm_content=content,
        )
        self.campaigns[campaign_id] = utm
        return campaign_id

    def get_utm_params(self, campaign_id: str) -> Dict[str, str]: 14091.baike.tongsou.com 
        """获取UTM参数字典"""
        utm = self.campaigns.get(campaign_id)
        if not utm:
            return {}
        params = {
            "utm_source": utm.utm_source,
            "utm_medium": utm.utm_medium,
            "utm_campaign": utm.utm_campaign,
        }
        if utm.utm_term: 14098.baike.tongsou.com 
            params["utm_term"] = utm.utm_term
        if utm.utm_content:
            params["utm_content"] = utm.utm_content
        return params

    def build_tracking_url(self, base_url: str, campaign_id: str) -> str:
        """构建带UTM参数的完整追踪URL"""
        params = self.get_utm_params(campaign_id)
        if not params: 14100.baike.tongsou.com 
            return base_url

        separator = "&" if "?" in base_url else "?"
        return f"{base_url}{separator}{urlencode(params)}"

    def parse_utm_from_url(self, url: str) -> Dict[str, str]:
        """从URL中解析UTM参数"""
        parsed = urlparse(url)
        params = parse_qs(parsed.query)
        utm_keys = ["utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content"]
        return {k: v[0] for k, v in params.items() if k in utm_keys}


# ==================== 模块4: 短链接服务 ====================

class ShortURLService: 14101.baike.tongsou.com 
    """
    短链接服务
    生成短链接并维护映射关系
    """

    def __init__(self, domain: str = "geo.link"):
        self.domain = domain
        self.links: Dict[str, TrackingLink] = {}
        self.short_to_link_id: Dict[str, str] = {}

    def create_short_link(self, original_url: str, content_id: str,
                          channel_id: str, utm_params: Dict[str, str],
                          expires_days: int = 365) -> TrackingLink:
        """创建短链接"""
        link_id = uuid.uuid4(14102.baike.tongsou.com).hex[:10]
        short_code = self._generate_short_code(link_id)
        short_url = f"https://{self.domain}/{short_code}"

        now = datetime.now().isoformat()
        expires_at = None
        if expires_days > 0: 14125.baike.tongsou.com 
            expires_at = (datetime.now() + timedelta(days=expires_days)).isoformat()

        link = TrackingLink(
            link_id=link_id,
            original_url=original_url,
            short_url=short_url,
            content_id=content_id,
            channel_id=channel_id,
            utm_params=utm_params,
            created_at=now,
            expires_at=expires_at,
            click_count=0,
            conversion_count=0,
        )
        self.links[link_id] = link
        self.short_to_link_id[short_code] = link_id
        return link

    def resolve(self, short_code: str) -> Optional[TrackingLink]:
        """解析短链接，返回原始追踪链接"""
        link_id = self.short_to_link_id.get(short_code)
        if not link_id: 14133.baike.tongsou.com 
            return None
        link = self.links.get(link_id)
        if not link: 14384.baike.tongsou.com 
            return None
        # 检查是否过期
        if link.expires_at: 14145.baike.tongsou.com 
            if datetime.fromisoformat(link.expires_at) < datetime.now():
                return None
        return link

    def increment_click(self, link_id: str):
        """增加点击计数"""
        if link_id in self.links: 14383.baike.tongsou.com 
            self.links[link_id].click_count += 1

    def increment_conversion(self, link_id: str):
        """增加转化计数"""
        if link_id in self.links:
            self.links[link_id].conversion_count += 1

    def _generate_short_code(self, link_id: str) -> str:
        """生成短代码"""
        chars = "abcdefghijklmnopqrstuvwxyz0123456789"
        hash_val = int(hashlib.md5(link_id.encode()).hexdigest(), 16)
        code = ""
        for _ in range(6): 15002.baike.tongsou.com 
            code += chars[hash_val % len(chars)]
            hash_val //= len(chars)
        return code

    def get_link_stats(self, link_id: str) -> Dict[str, Any]:
        """获取链接统计"""
        link = self.links.get(link_id)
        if not link: 15015.baike.tongsou.com 
            return {}
        return {
            "link_id": link.link_id,
            "short_url": link.short_url,
            "original_url": link.original_url,
            "click_count": link.click_count,
            "conversion_count": link.conversion_count,
            "ctr": link.conversion_count / link.click_count if link.click_count > 0 else 0,
            "created_at": link.created_at,
        }


# ==================== 模块5: 点击追踪与日志 ====================

class ClickTracker:
    """
    点击追踪器
    记录所有点击事件，支持会话追踪
    """

    def __init__(self): 15025.baike.tongsou.com 
        self.clicks: List[ClickEvent] = []
        self.sessions: Dict[str, List[str]] = {}  # session_id -> [event_ids]
        self.user_sessions: Dict[str, List[str]] = {}  # user_id -> [session_ids]

    def record_click(self, link_id: str, content_id: str, channel_id: str,
                     user_id: str, ip_address: str, user_agent: str,
                     referrer: str = "", geo: Dict[str, str] = None) -> ClickEvent:
        """记录点击事件"""
        event_id = uuid.uuid4().hex[:16]
        session_id = self._get_or_create_session(user_id)
        now = datetime.now().isoformat(15027.baike.tongsou.com)

        event = ClickEvent(
            event_id=event_id,
            link_id=link_id,
            content_id=content_id,
            channel_id=channel_id,
            user_id=user_id,
            session_id=session_id,
            ip_address=ip_address,
            user_agent=user_agent,
            referrer=referrer,
            timestamp=now,
            geo_location=geo or {},
        )
        self.clicks.append(event)

        # 更新会话
        if session_id not in self.sessions: 15029.baike.tongsou.com 
            self.sessions[session_id] = []
        self.sessions[session_id].append(event_id)

        # 更新用户会话映射
        if user_id not in self.user_sessions: 15030.baike.tongsou.com 
            self.user_sessions[user_id] = []
        if session_id not in self.user_sessions[user_id]:
            self.user_sessions[user_id].append(session_id)

        return event

    def _get_or_create_session(self, user_id: str) -> str:
        """获取或创建会话ID"""
        # 简化：每次点击创建新会话
        # 实际应基于时间窗口（如30分钟无活动则新建会话）
        return uuid.uuid4().hex[:12]

    def get_user_journey(self, user_id: str) -> List[ClickEvent]:
        """获取用户完整点击旅程"""
        session_ids = self.user_sessions.get(user_id, [])
        journey = []
        for sid in session_ids:
            event_ids = self.sessions.get(sid, [])
            for eid in event_ids: 15031.baike.tongsou.com 
                for click in self.clicks:
                    if click.event_id == eid:
                        journey.append(click)
                        break
        # 按时间排序
        journey.sort(key=lambda x: x.timestamp)
        return journey

    def get_channel_clicks(self, channel_id: str) -> int: 15033.baike.tongsou.com 
        """获取渠道总点击数"""
        return sum(1 for c in self.clicks if c.channel_id == channel_id)

    def get_content_clicks(self, content_id: str) -> int: 15032.baike.tongsou.com 
        """获取内容总点击数"""
        return sum(1 for c in self.clicks if c.content_id == content_id)

    def get_daily_clicks(self, days: int = 7) -> Dict[str, int]:
        """获取最近N天的每日点击数"""
        daily = defaultdict(int)
        cutoff = datetime.now() - timedelta(days=days)
        for click in self.clicks: 15034.baike.tongsou.com 
            ts = datetime.fromisoformat(click.timestamp)
            if ts >= cutoff:
                date_key = ts.strftime("%Y-%m-%d")
                daily[date_key] += 1
        return dict(daily)


# ==================== 模块6: 转化追踪 ====================

class ConversionTracker: 15035.baike.tongsou.com 
    """
    转化追踪器
    记录转化事件并与点击会话关联
    """

    def __init__(self): 15037.baike.tongsou.com 
        self.conversions: List[ConversionEvent] = []
        self.session_conversions: Dict[str, List[str]] = {}  # session_id -> [conversion_ids]

    def record_conversion(self, event_type: str, user_id: str,
                          session_id: str, value: float,
                          currency: str = "CNY",
                          metadata: Dict[str, Any] = None) -> ConversionEvent:
        """记录转化事件"""
        conversion_id = uuid.uuid4().hex[:16]
        now = datetime.now().isoformat()

        conv = ConversionEvent(
            conversion_id=conversion_id,
            event_type=event_type,
            user_id=user_id,
            session_id=session_id,
            value=value,
            currency=currency,
            timestamp=now,
            metadata=metadata or {},
        )
        self.conversions.append(conv)

        if session_id not in self.session_conversions:
            self.session_conversions[session_id] = []
        self.session_conversions[session_id].append(conversion_id)

        return conv

    def get_session_conversions(self, session_id: str) -> List[ConversionEvent]:
        """获取会话的所有转化"""
        conv_ids = self.session_conversions.get(session_id, [])
        return [c for c in self.conversions if c.conversion_id in conv_ids]

    def get_total_value(self, event_type: str = None) -> float:
        """获取总转化价值"""
        total = 0.0
        for conv in self.conversions: 16003.baike.tongsou.com 
            if event_type is None or conv.event_type == event_type:
                total += conv.value
        return total

    def get_conversion_rate(self, total_clicks: int) -> float:
        """计算转化率"""
        if total_clicks == 0: 16007.baike.tongsou.com 
            return 0.0
        return len(self.conversions) / total_clicks


# ==================== 模块7: 归因引擎 ====================

class AttributionEngine: 17020.baike.tongsou.com 
    """
    归因引擎
    支持多种归因模型，将转化价值分配给各个触点的渠道
    """

    def __init__(self, click_tracker: ClickTracker,
                 conversion_tracker: ConversionTracker):
        self.click_tracker = click_tracker
        self.conversion_tracker = conversion_tracker

    def attribute(self, session_id: str,
                  model: AttributionModel = AttributionModel.LAST_CLICK) -> AttributionResult:
        """
        对单个会话进行归因
        """
        # 获取会话中的所有点击
        event_ids = self.click_tracker.sessions.get(session_id, [])
        touchpoints = []
        for eid in event_ids:
            for click in self.click_tracker.clicks:
                if