短链接服务与MarTech生态集成：CRM、CDP、广告归因与营销自动化

{
  "event": "link.click",
  "timestamp": "2025-01-15T14:30:00Z",
  "link_id": "l_abc123",
  "short_url": "https://brand.link/sale25",
  "original_url": "https://brand.com/sale",
  "contact": {
    "email": "user@example.com",
    "crm_id": "hubspot_123456",
    "segment": "warm_lead"
  },
  "context": {
    "referrer": "https://newsletter.brand.com",
    "utm_source": "newsletter",
    "utm_medium": "email",
    "utm_campaign": "jan_sale",
    "device": "iPhone 15",
    "os": "iOS 17",
    "city": "Shanghai"
  }
}

import requests

class HubSpotIntegration:
    def __init__(self, access_token):
        self.token = access_token
        self.base = "https://api.hubapi.com"
    
    def track_link_click(self, contact_email, link_event):
        # 1. 查找联系人
        contact = self.find_contact_by_email(contact_email)
        
        # 2. 记录活动
        self.create_engagement(
            contact_id=contact['vid'],
            engagement_type="INTERACTION",
            metadata={
                "link_url": link_event['short_url'],
                "utm_campaign": link_event['context']['utm_campaign'],
                "device": link_event['context']['device']
            }
        )
        
        # 3. 更新线索得分
        current_score = contact['properties'].get('hubspotscore', 0)
        new_score = min(current_score + 5, 100)  # 点击+5分
        self.update_contact(contact['vid'], {'hubspotscore': new_score})

// 短链点击时触发 Segment 事件
analytics.track('Short Link Clicked', {
  link_id: 'l_abc123',
  short_url: 'brand.link/sale25',
  campaign: 'jan_sale',
  channel: 'email',
  device_type: 'mobile',
  os: 'iOS',
  city: 'Shanghai'
});

type CDPEvent struct {
    EventID     string                 `json:"event_id"`
    EventType   string                 `json:"event_type"`  // "link.click"
    Timestamp   time.Time              `json:"timestamp"`
    UserID      string                 `json:"user_id"`     // 统一ID
    AnonymousID string                 `json:"anonymous_id"` // 设备ID
    Properties  map[string]interface{}  `json:"properties"`
    Context     Context                `json:"context"`
}

func (s *ShortlinkService) EmitCDPEvent(event CDPEvent) {
    // 多渠道并发投递
    go s.segmentClient.Track(event)
    go s.snowflakeLoader.Insert(event)
    go s.kafkaProducer.Send("martech.events", event)
}

type UTMBuilder struct{}

func (b *UTMBuilder) Generate(params CampaignParams) string {
    utm := url.Values{}
    utm.Set("utm_source", b.resolveSource(params.Channel))
    utm.Set("utm_medium", b.resolveMedium(params.Channel, params.Format))
    utm.Set("utm_campaign", params.CampaignID)
    utm.Set("utm_content", params.CreativeID)
    utm.Set("utm_term", params.Keyword)
    
    // 自定义参数
    utm.Set("sl_channel_id", params.ChannelID)
    utm.Set("sl_creative_id", params.CreativeID)
    utm.Set("sl_user_segment", params.UserSegment)
    
    return utm.Encode()
}

// 自动识别渠道
func (b *UTMBuilder) resolveSource(channel string) string {
    switch {
    case strings.Contains(channel, "wechat"):
        return "wechat"
    case strings.Contains(channel, "douyin"):
        return "douyin"
    case strings.Contains(channel, "email"):
        return "newsletter"
    default:
        return channel
    }
}

<!-- 短链跳转前的中间页 -->
<script>
  // 回传标准事件
  fbq('track', 'PageView', {
    source: '{{utm_source}}',
    campaign: '{{utm_campaign}}',
    link_id: '{{link_id}}'
  });
  
  // 如果是购买转化页
  fbq('track', 'Purchase', {
    value: 99.00,
    currency: 'USD',
    content_ids: ['prod_123'],
    content_type: 'product'
  });
</script>

workflows:
  - name: "hot_lead_nurture"
    trigger:
      event: "link.click"
      conditions:
        - "user.segment == 'warm'"
        - "utm_source == 'newsletter'"
        - "click_count_7d >= 2"
    actions:
      - type: "delay"
        duration: "2h"
      - type: "email"
        template: "nurture_v2"
        subject: "您查看的内容，还有这些补充"
      - type: "slack"
        channel: "#sales-alerts"
        message: "潜在客户 {{user.email}} 进入培育序列"

-- 点击事实表
CREATE TABLE fact_link_clicks (
    click_id BIGINT,
    link_id VARCHAR(32),
    tenant_id VARCHAR(32),
    user_id VARCHAR(64),       -- 统一用户ID
    anonymous_id VARCHAR(64),  -- 设备匿名ID
    timestamp TIMESTAMP,
    utm_source VARCHAR(50),
    utm_medium VARCHAR(50),
    utm_campaign VARCHAR(100),
    device_type ENUM('mobile','desktop','tablet'),
    os VARCHAR(20),
    browser VARCHAR(30),
    country VARCHAR(2),
    city VARCHAR(50),
    referrer_domain VARCHAR(100)
);

-- 用户归因表
CREATE TABLE user_attribution (
    user_id VARCHAR(64),
    first_touch_source VARCHAR(50),
    first_touch_timestamp TIMESTAMP,
    last_touch_source VARCHAR(50),
    last_touch_timestamp TIMESTAMP,
    touch_count INT,
    conversion_flag BOOLEAN
);

func WechatEventCallback(w http.ResponseWriter, r *http.Request) {
    event := parseWechatEvent(r)
    
    // 统一转换为CDP事件
    cdpEvent := CDPEvent{
        EventType: "wechat." + event.EventType,
        UserID:    resolveUnionID(event.OpenID),
        Properties: map[string]interface{}{
            "mp_appid":    event.AppID,
            "scene":       event.Scene,
            "link_id":     event.LinkID,
        },
    }
    
    EmitCDPEvent(cdpEvent)
}