跨境电商 Facebook 营销的工程化实践：商品目录同步、事件回传与合规治理

import csv
from dataclasses import dataclass, asdict
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path

REQUIRED = ("id", "title", "description", "availability",
            "condition", "price", "link", "image_link", "brand")

@dataclass
class CatalogItem:
    id: str
    title: str
    description: str
    availability: str          # in stock / out of stock
    condition: str             # new / refurbished
    price: str                 # "19.99 USD"
    link: str
    image_link: str
    brand: str
    gtin: str = ""

    def validate(self) -> list[str]:
        issues = []
        for field in REQUIRED:
            if not getattr(self, field, ""):
                issues.append(f"{self.id}: 缺少字段 {field}")
        if self.availability not in ("in stock", "out of stock"):
            issues.append(f"{self.id}: availability 非法")
        if self.condition not in ("new", "refurbished", "used"):
            issues.append(f"{self.id}: condition 非法")
        try:
            amount, currency = self.price.split()
            if Decimal(amount) <= 0:
                issues.append(f"{self.id}: 价格必须为正")
            if len(currency) != 3:
                issues.append(f"{self.id}: 币种格式错误")
        except ValueError:
            issues.append(f"{self.id}: price 格式应为 '19.99 USD'")
        return issues

def format_price(amount: Decimal, currency: str) -> str:
    value = amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    return f"{value} {currency}"

def write_feed(items: list[CatalogItem], out: Path) -> dict:
    errors = [e for it in items for e in it.validate()]
    if errors:
        return {"ok": False, "errors": errors[:20]}

    fields = list(asdict(items[0]).keys())
    with out.open("w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        for it in items:
            writer.writerow(asdict(it))
    return {"ok": True, "count": len(items), "path": str(out)}

import os, hashlib, time, requests
from flask import Flask, request, jsonify

app = Flask(__name__)
PIXEL_ID = os.getenv("FB_PIXEL_ID")
TOKEN = os.getenv("FB_ACCESS_TOKEN")
API = "v20.0"

def sha256(value: str) -> str:
    return hashlib.sha256(value.strip().lower().encode()).hexdigest()

def build_user_data(payload: dict, consent: dict) -> dict:
    """只有获得授权的字段才上报"""
    data = {}
    if consent.get("email") and payload.get("email"):
        data["em"] = [sha256(payload["email"])]
    if consent.get("phone") and payload.get("phone"):
        data["ph"] = [sha256(payload["phone"])]
    if payload.get("fbp"):
        data["fbp"] = payload["fbp"]
    if payload.get("fbc"):
        data["fbc"] = payload["fbc"]
    return data

@app.post("/capi/purchase")
def capi_purchase():
    body = request.get_json(force=True)
    consent = body.get("consent", {})

    # 未授权则不回传，宁可损失信号也不违规
    if not any(consent.values()):
        return jsonify({"status": "skipped", "reason": "no consent"}), 200

    user_data = build_user_data(body, consent)
    if not user_data:
        return jsonify({"status": "skipped", "reason": "no usable data"}), 200

    event = {
        "event_name": "Purchase",
        "event_time": int(time.time()),
        "event_id": body["order_id"],          # 幂等键
        "action_source": "website",
        "user_data": user_data,
        "custom_data": {
            "currency": body.get("currency", "USD"),
            "value": str(body.get("value", "0")),
        },
    }

    resp = requests.post(
        f"https://graph.facebook.com/{API}/{PIXEL_ID}/events",
        params={"access_token": TOKEN},
        json={"data": [event]},
        timeout=10,
    )
    return jsonify(resp.json()), resp.status_code

import os, requests
from decimal import Decimal

TOKEN = os.getenv("FB_ACCESS_TOKEN")
ACCOUNT = os.getenv("FB_AD_ACCOUNT_ID")

def fetch_insights(date_preset: str = "today") -> list[dict]:
    resp = requests.get(
        f"https://graph.facebook.com/v20.0/act_{ACCOUNT}/insights",
        params={
            "access_token": TOKEN,
            "fields": "campaign_id,campaign_name,spend,ctr,cpc,actions",
            "level": "campaign",
            "date_preset": date_preset,
            "limit": 200,
        },
        timeout=20,
    )
    resp.raise_for_status()
    return resp.json().get("data", [])

def rule_engine(rows: list[dict],
                max_cpc: Decimal = Decimal("1.5"),
                min_ctr: float = 0.008) -> list[dict]:
    alerts = []
    for r in rows:
        spend = Decimal(str(r.get("spend", "0")))
        cpc = Decimal(str(r.get("cpc", "0")))
        ctr = float(r.get("ctr", 0))

        if spend < 10:          # 样本太小不下结论
            continue
        if cpc > max_cpc:
            alerts.append({"campaign": r["campaign_name"],
                           "type": "cpc_high", "value": str(cpc)})
        if ctr < min_ctr:
            alerts.append({"campaign": r["campaign_name"],
                           "type": "ctr_low", "value": ctr})
    return alerts

import re

ABSOLUTE = {"best", "No.1", "第一", "最好", "100%", "guaranteed"}
PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

def audit_creative(text: str) -> list[str]:
    issues = []
    blob = text.lower()
    for w in ABSOLUTE:
        if w.lower() in blob:
            issues.append(f"含绝对化用语：{w}")
    if PII.search(text):
        issues.append("含个人信息，需脱敏")
    return issues