微信小程序推广系统的工程化实践：渠道追踪、归因结算与合规风控

import re
import hashlib
from dataclasses import dataclass

CHANNEL_RE = re.compile(r"^[a-z0-9_]{1,32}$")

@dataclass(frozen=True)
class Channel:
    channel_id: str
    campaign_id: str = ""
    source: str = "unknown"

def parse_channel(raw: dict) -> Channel | None:
    ch = (raw.get("ch") or "").strip().lower()
    if not CHANNEL_RE.match(ch):
        return None
    cid = (raw.get("cid") or "").strip()[:64]
    return Channel(channel_id=ch, campaign_id=cid,
                   source=raw.get("src", "unknown"))

def device_fingerprint(openid: str, appid: str) -> str:
    raw = f"{appid}:{openid}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

from datetime import datetime, timedelta

ATTRIBUTION_WINDOW = timedelta(days=7)

def attribute(user_id: str, order_time: datetime,
              clicks: list[dict]) -> str | None:
    valid = [
        c for c in clicks
        if order_time - c["click_time"] <= ATTRIBUTION_WINDOW
    ]
    if not valid:
        return None
    valid.sort(key=lambda c: c["click_time"], reverse=True)
    return valid[0]["channel_id"]

def dedup_key(user_id: str, order_id: str) -> str:
    return hashlib.sha256(
        f"{user_id}:{order_id}".encode()
    ).hexdigest()[:24]

from decimal import Decimal, ROUND_HALF_UP

COMMISSION_RATE = {
    "default": Decimal("0.05"),
    "vip_channel": Decimal("0.08"),
}

def calc_commission(amount: Decimal, channel_id: str) -> Decimal:
    rate = COMMISSION_RATE.get(channel_id, COMMISSION_RATE["default"])
    return (amount * rate).quantize(Decimal("0.01"),
                                    rounding=ROUND_HALF_UP)

SETTLED: dict[str, str] = {}

def settle(order_id: str, user_id: str, amount: Decimal,
           channel_id: str) -> dict:
    key = dedup_key(user_id, order_id)
    if key in SETTLED:
        return {"status": "duplicate", "settle_id": SETTLED[key]}

    commission = calc_commission(amount, channel_id)
    if commission <= 0:
        return {"status": "rejected", "reason": "zero commission"}

    settle_id = f"st_{key}"
    SETTLED[key] = settle_id
    return {
        "status": "ok",
        "settle_id": settle_id,
        "commission": str(commission),
        "channel_id": channel_id,
    }

def reconcile(daily_orders: list[dict],
              daily_settlements: list[dict]) -> dict:
    order_total = sum(Decimal(o["amount"]) for o in daily_orders)
    settle_total = sum(Decimal(s["amount"]) for s in daily_settlements)
    diff = order_total - settle_total
    return {
        "order_total": str(order_total),
        "settle_total": str(settle_total),
        "diff": str(diff),
        "ok": diff == 0,
    }

import time
from collections import defaultdict

RATE_LIMIT = defaultdict(list)

def check_rate(user_id: str, action: str,
               window: int = 60, limit: int = 10) -> bool:
    now = time.time()
    key = f"{user_id}:{action}"
    RATE_LIMIT[key] = [t for t in RATE_LIMIT[key] if now - t < window]
    if len(RATE_LIMIT[key]) >= limit:
        return False
    RATE_LIMIT[key].append(now)
    return True

BLACKLIST = {"device_x", "device_y"}

def risk_check(user_id: str, device_id: str,
               amount: Decimal) -> list[str]:
    flags = []
    if device_id in BLACKLIST:
        flags.append("device_blacklisted")
    if not check_rate(user_id, "order"):
        flags.append("rate_exceeded")
    if amount > Decimal("10000"):
        flags.append("high_amount")
    return flags

import re

PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def mask(text: str) -> str:
    return PII.sub("[REDACTED]", text)

def audit_promotion(text: str) -> None:
    bad = {"最好", "第一", "国家级", "保证收益", "稳赚"}
    if any(w in text for w in bad):
        raise ValueError("含绝对化或承诺性用语")