下面给出一个可运行的参考脚本，用于整合四类数据源（销售CRM、运营日志、客服工单、数据仓库），并输出标准化后的维表与事实表。脚本采用“抽取-标准化-实体解析-整合-输出”的流程，尽量保持对具体厂商/系统无关，便于按需对接实际环境。
说明与假设
- 依赖库：pandas、requests、SQLAlchemy、python-dateutil（pandas 自带解析也可）、pyarrow（可选，用于写 Parquet；若未安装将回退为 CSV）。
- 认证与连接信息通过环境变量配置，避免硬编码凭据。
- API 端点与字段命名存在差异，脚本提供字段映射与注释，便于按实际系统字段调整。
- 实体解析采用“确定性匹配优先”的策略：优先使用数据仓库中的客户主数据（若有）；否则以域名（来自邮箱/公司网址）为锚点，生成稳定客户ID。该策略透明、可复现，便于审计与迭代。
- 输出结果：
- dim_customer：客户维表（去重、稳定ID）
- fact_sales：销售机会/订单事实（含客户ID）
- fact_interactions：交互事实（运营日志与客服工单统一为“交互”）
- 如果需要将结果回写数据仓库，可开启示例中的 to_sql 段落（以 SQLAlchemy URL 为准）。
使用方法
- 设置环境变量后直接运行：python integrate_sources.py
- 核对并调整“字段映射”段落以与实际数据对齐。
- 在生产使用前，先在开发环境对小样本运行，校验覆盖率与正确性。
代码
#!/usr/bin/env python3
-- coding: utf-8 --
"""
整合销售CRM、运营日志、客服工单、数据仓库的数据，输出统一的客户维表和交互/销售事实表。
环境变量（示例）：
CRM
CRM_API_URL=https://your-crm.example.com/api
CRM_API_TOKEN=xxxx
客服工单
TICKET_API_URL=https://your-ticket.example.com/api
TICKET_API_TOKEN=yyyy
运营日志
OPS_LOG_PATH=./data/ops_logs/*.jsonl # 支持 glob；可为 csv/jsonl
数据仓库（只读或读写）
DW_SQLALCHEMY_URL=postgresql+psycopg2://user:pwd@host:5432/dbname
控制增量
SINCE_DAYS=30
运行：
python integrate_sources.py
输出：
./output/dim_customer.parquet (或 csv)
./output/fact_sales.parquet (或 csv)
./output/fact_interactions.parquet (或 csv)
"""
import os
import re
import json
import glob
import math
import uuid
import time
import logging
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
import pandas as pd
import requests
from sqlalchemy import create_engine, text
-------------------------
全局设置
-------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s"
)
UTC = timezone.utc
-------------------------
配置读取
-------------------------
CRM_API_URL = os.getenv("CRM_API_URL", "").rstrip("/")
CRM_API_TOKEN = os.getenv("CRM_API_TOKEN", "")
TICKET_API_URL = os.getenv("TICKET_API_URL", "").rstrip("/")
TICKET_API_TOKEN = os.getenv("TICKET_API_TOKEN", "")
OPS_LOG_PATH = os.getenv("OPS_LOG_PATH", "./data/ops_logs/*.jsonl")
DW_SQLALCHEMY_URL = os.getenv("DW_SQLALCHEMY_URL", "")
SINCE_DAYS = int(os.getenv("SINCE_DAYS", "30"))
SINCE_DT = datetime.now(tz=UTC) - timedelta(days=SINCE_DAYS)
OUTPUT_DIR = "./output"
os.makedirs(OUTPUT_DIR, exist_ok=True)
-------------------------
实用函数：清洗/标准化
-------------------------
def normalize_email(email: Optional[str]) -> Optional[str]:
if not email or not isinstance(email, str):
return None
e = email.strip().lower()
# 简单校验
if "@" not in e:
return None
return e
def extract_domain_from_email(email: Optional[str]) -> Optional[str]:
e = normalize_email(email)
if not e:
return None
try:
return e.split("@", 1)[1]
except Exception:
return None
def normalize_domain(domain: Optional[str]) -> Optional[str]:
if not domain or not isinstance(domain, str):
return None
d = domain.strip().lower()
d = re.sub(r"^https?://", "", d)
d = d.split("/")[0]
# 去除常见前缀
for prefix in ["www.", "m."]:
if d.startswith(prefix):
d = d[len(prefix):]
if "." not in d:
return None
return d
def extract_domain_from_url(url: Optional[str]) -> Optional[str]:
if not url or not isinstance(url, str):
return None
return normalize_domain(url)
def coalesce_domain(email: Optional[str], website: Optional[str]) -> Optional[str]:
d = extract_domain_from_email(email)
if d:
return d
return extract_domain_from_url(website)
def parse_ts_to_utc(ts) -> Optional[pd.Timestamp]:
if ts is None or (isinstance(ts, float) and math.isnan(ts)):
return None
try:
t = pd.to_datetime(ts, utc=True, errors="coerce")
if pd.isna(t):
return None
return t.tz_convert("UTC") if t.tzinfo else t.tz_localize("UTC")
except Exception:
return None
def stable_id_from_domain(domain: str) -> str:
# 依据域名生成稳定 UUID（可复现）
return uuid.uuid5(uuid.NAMESPACE_DNS, domain).hex
def safe_to_parquet_or_csv(df: pd.DataFrame, path_base: str):
try:
df.to_parquet(f"{path_base}.parquet", index=False)
logging.info(f"written: {path_base}.parquet")
except Exception as e:
logging.warning(f"fall back to CSV for {path_base}: {e}")
df.to_csv(f"{path_base}.csv", index=False)
logging.info(f"written: {path_base}.csv")
-------------------------
数据抽取：通用 API 工具
-------------------------
def api_get_json(url: str, headers: Dict[str, str], params: Dict) -> dict:
resp = requests.get(url, headers=headers, params=params, timeout=30)
resp.raise_for_status()
return resp.json()
def fetch_paginated_list(
base_url: str, endpoint: str, headers: Dict[str, str],
params: Dict, page_param: str = "page", per_page_param: str = "per_page",
per_page: int = 200, data_path: Optional[List[str]] = None, max_pages: int = 100
) -> List[dict]:
"""
通用分页拉取。根据你实际的 API 规则调整参数名/翻页逻辑/data_path。
data_path: 用于从返回的 JSON 中取列表的路径，如 ["data","items"]。
"""
out = []
page = 1
while page <= max_pages:
p = dict(params)
p[page_param] = page
p[per_page_param] = per_page
data = api_get_json(f"{base_url}{endpoint}", headers, p)
# 取数据列表
items = data
if data_path:
for k in data_path:
items = items.get(k, [])
if not isinstance(items, list):
items = []
break
elif isinstance(items, dict):
# 若不清楚结构，可尝试常见键
for k in ["data", "items", "results"]:
if k in items and isinstance(items[k], list):
items = items[k]
break
if isinstance(items, dict):
items = []
if not items:
break
out.extend(items)
page += 1
# 简单限速，必要时增强
time.sleep(0.1)
return out
-------------------------
数据抽取：CRM
-------------------------
def fetch_crm_contacts(since_iso: str) -> pd.DataFrame:
if not CRM_API_URL or not CRM_API_TOKEN:
logging.warning("CRM 配置缺失，跳过 contacts 抽取")
return pd.DataFrame()
headers = {"Authorization": f"Bearer {CRM_API_TOKEN}"}
# 请按实际 API 调整 endpoint、入参、data_path
items = fetch_paginated_list(
base_url=CRM_API_URL, endpoint="/contacts",
headers=headers, params={"updated_since": since_iso},
data_path=None
)
df = pd.json_normalize(items)
# 字段映射：根据你实际系统字段调整
# 目标字段：contact_id, contact_name, contact_email, account_name, account_website, updated_at
field_map = {
"id": "contact_id",
"name": "contact_name",
"email": "contact_email",
"account.name": "account_name",
"account.website": "account_website",
"updated_at": "updated_at",
}
df = df.rename(columns={k: v for k, v in field_map.items() if k in df.columns})
# 只保留目标字段
keep = list(field_map.values())
df = df[[c for c in keep if c in df.columns]].copy()
return df
def fetch_crm_deals(since_iso: str) -> pd.DataFrame:
if not CRM_API_URL or not CRM_API_TOKEN:
logging.warning("CRM 配置缺失，跳过 deals 抽取")
return pd.DataFrame()
headers = {"Authorization": f"Bearer {CRM_API_TOKEN}"}
items = fetch_paginated_list(
base_url=CRM_API_URL, endpoint="/deals",
headers=headers, params={"updated_since": since_iso},
data_path=None
)
df = pd.json_normalize(items)
# 字段映射：根据你实际系统字段调整
# 目标字段：deal_id, deal_name, amount, stage, close_date, owner, account_name, account_website, contact_email, updated_at
field_map = {
"id": "deal_id",
"name": "deal_name",
"amount": "amount",
"stage": "stage",
"close_date": "close_date",
"owner.name": "owner",
"account.name": "account_name",
"account.website": "account_website",
"primary_contact.email": "contact_email",
"updated_at": "updated_at",
}
df = df.rename(columns={k: v for k, v in field_map.items() if k in df.columns})
keep = list(field_map.values())
df = df[[c for c in keep if c in df.columns]].copy()
return df
-------------------------
数据抽取：客服工单
-------------------------
def fetch_tickets(since_iso: str) -> pd.DataFrame:
if not TICKET_API_URL or not TICKET_API_TOKEN:
logging.warning("Ticket 配置缺失，跳过工单抽取")
return pd.DataFrame()
headers = {"Authorization": f"Bearer {TICKET_API_TOKEN}"}
items = fetch_paginated_list(
base_url=TICKET_API_URL, endpoint="/tickets",
headers=headers, params={"updated_since": since_iso},
data_path=None
)
df = pd.json_normalize(items)
# 字段映射：根据你实际系统字段调整
# 目标字段：ticket_id, subject, status, priority, created_at, updated_at, requester_email, assignee, organization, channel
field_map = {
"id": "ticket_id",
"subject": "subject",
"status": "status",
"priority": "priority",
"created_at": "created_at",
"updated_at": "updated_at",
"requester.email": "requester_email",
"assignee.name": "assignee",
"organization.name": "organization",
"channel": "channel",
}
df = df.rename(columns={k: v for k, v in field_map.items() if k in df.columns})
keep = list(field_map.values())
df = df[[c for c in keep if c in df.columns]].copy()
return df
-------------------------
数据抽取：运营日志（本地/对象存储）支持 CSV/JSONL
-------------------------
def load_ops_logs(path_glob: str) -> pd.DataFrame:
paths = glob.glob(path_glob)
if not paths:
logging.warning(f"未找到运营日志文件：{path_glob}")
return pd.DataFrame()
frames = []
for p in paths:
if p.lower().endswith(".csv"):
tmp = pd.read_csv(p)
elif p.lower().endswith(".jsonl") or p.lower().endswith(".ndjson"):
tmp = pd.read_json(p, lines=True)
elif p.lower().endswith(".json"):
tmp = pd.read_json(p)
else:
logging.warning(f"不支持的文件类型，跳过：{p}")
continue
frames.append(tmp)
if not frames:
return pd.DataFrame()
df = pd.concat(frames, ignore_index=True)
# 字段映射：根据实际日志格式调整
# 目标字段：event_id, event_type, occurred_at, user_email, account_domain, metadata(json)
# 尝试映射常见字段
candidate_map = {
"id": "event_id",
"event_id": "event_id",
"type": "event_type",
"event_type": "event_type",
"timestamp": "occurred_at",
"occurred_at": "occurred_at",
"user.email": "user_email",
"user_email": "user_email",
"account.domain": "account_domain",
"account_domain": "account_domain",
"metadata": "metadata",
"data": "metadata",
}
rename_map = {k: v for k, v in candidate_map.items() if k in df.columns}
df = df.rename(columns=rename_map)
keep = ["event_id", "event_type", "occurred_at", "user_email", "account_domain", "metadata"]
df = df[[c for c in keep if c in df.columns]].copy()
# metadata 标准化为 JSON 字符串，便于落地
if "metadata" in df.columns:
df["metadata"] = df["metadata"].apply(lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, (dict, list)) else (x if isinstance(x, str) else None))
return df
-------------------------
数据抽取：数据仓库（读取客户主数据等）
-------------------------
def query_dw(sql: str) -> pd.DataFrame:
if not DW_SQLALCHEMY_URL:
logging.warning("DW_SQLALCHEMY_URL 未设置，跳过数据仓库查询")
return pd.DataFrame()
engine = create_engine(DW_SQLALCHEMY_URL)
with engine.connect() as conn:
df = pd.read_sql(text(sql), conn)
return df
def fetch_dw_customers() -> pd.DataFrame:
# 请按实际客户主数据表调整查询
sql = """
SELECT
customer_id,
customer_name,
domain, -- 建议在主数据中维护主域名
website,
updated_at
FROM dim_customer
"""
try:
return query_dw(sql)
except Exception as e:
logging.warning(f"查询数据仓库客户主数据失败：{e}")
return pd.DataFrame()
-------------------------
标准化与实体解析
-------------------------
def standardize_crm_contacts(df: pd.DataFrame) -> pd.DataFrame:
if df.empty:
return df
df["contact_email"] = df["contact_email"].apply(normalize_email) if "contact_email" in df.columns else None
if "account_website" in df.columns:
df["account_domain"] = df["account_website"].apply(extract_domain_from_url)
else:
df["account_domain"] = None
if "updated_at" in df.columns:
df["updated_at"] = df["updated_at"].apply(parse_ts_to_utc)
return df
def standardize_crm_deals(df: pd.DataFrame) -> pd.DataFrame:
if df.empty:
return df
if "contact_email" in df.columns:
df["contact_email"] = df["contact_email"].apply(normalize_email)
# 优先从 contact_email 提取域名，回退到 account_website
email_domain = df["contact_email"].apply(extract_domain_from_email) if "contact_email" in df.columns else pd.Series([None]*len(df))
website_domain = df["account_website"].apply(extract_domain_from_url) if "account_website" in df.columns else pd.Series([None]*len(df))
df["account_domain"] = email_domain.fillna(website_domain)
if "close_date" in df.columns:
df["close_date"] = df["close_date"].apply(parse_ts_to_utc)
if "updated_at" in df.columns:
df["updated_at"] = df["updated_at"].apply(parse_ts_to_utc)
# 数值规范
if "amount" in df.columns:
df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
return df
def standardize_tickets(df: pd.DataFrame) -> pd.DataFrame:
if df.empty:
return df
if "requester_email" in df.columns:
df["requester_email"] = df["requester_email"].apply(normalize_email)
df["requester_domain"] = df["requester_email"].apply(extract_domain_from_email)
else:
df["requester_domain"] = None
for col in ["created_at", "updated_at"]:
if col in df.columns:
df[col] = df[col].apply(parse_ts_to_utc)
return df
def standardize_ops_logs(df: pd.DataFrame) -> pd.DataFrame:
if df.empty:
return df
if "user_email" in df.columns:
df["user_email"] = df["user_email"].apply(normalize_email)
df["user_domain"] = df["user_email"].apply(extract_domain_from_email)
else:
df["user_domain"] = None
if "account_domain" in df.columns:
df["account_domain"] = df["account_domain"].apply(normalize_domain)
if "occurred_at" in df.columns:
df["occurred_at"] = df["occurred_at"].apply(parse_ts_to_utc)
return df
def build_dim_customer(dw_customers: pd.DataFrame,
crm_contacts: pd.DataFrame,
crm_deals: pd.DataFrame,
tickets: pd.DataFrame,
ops_logs: pd.DataFrame) -> pd.DataFrame:
"""构建客户维表：优先使用DW主数据；补全来自CRM/Ticket/Logs的域名集合，生成稳定ID"""
candidates = []
# DW 主数据（优先）
if not dw_customers.empty:
df = dw_customers.copy()
if "domain" in df.columns:
df["domain"] = df["domain"].apply(normalize_domain)
else:
df["domain"] = df["website"].apply(extract_domain_from_url) if "website" in df.columns else None
df = df[["customer_id", "customer_name", "domain", "website", "updated_at"]].copy()
candidates.append(df)
# CRM 账户域（来自 contacts 和 deals）
if not crm_contacts.empty:
tmp = crm_contacts.copy()
tmp["domain"] = tmp["account_domain"]
tmp["customer_name"] = tmp.get("account_name")
tmp["website"] = tmp.get("account_website")
tmp = tmp[["customer_name", "domain", "website"]]
candidates.append(tmp)
if not crm_deals.empty:
tmp = crm_deals.copy()
tmp["domain"] = tmp["account_domain"]
tmp["customer_name"] = tmp.get("account_name")
tmp["website"] = tmp.get("account_website")
tmp = tmp[["customer_name", "domain", "website"]]
candidates.append(tmp)
# Tickets 域名
if not tickets.empty:
tmp = tickets.copy()
tmp["domain"] = tmp["requester_domain"]
tmp["customer_name"] = tmp.get("organization")
tmp["website"] = None
tmp = tmp[["customer_name", "domain", "website"]]
candidates.append(tmp)
# Logs 域名
if not ops_logs.empty:
tmp = ops_logs.copy()
# 优先使用 account_domain，其次 user_domain
tmp["domain"] = tmp["account_domain"].fillna(tmp["user_domain"])
tmp["customer_name"] = None
tmp["website"] = None
tmp = tmp[["customer_name", "domain", "website"]]
candidates.append(tmp)
if not candidates:
return pd.DataFrame(columns=["customer_id", "customer_name", "domain", "website", "source"])
all_cand = pd.concat(candidates, ignore_index=True)
all_cand["domain"] = all_cand["domain"].apply(normalize_domain)
all_cand = all_cand.dropna(subset=["domain"]).drop_duplicates(subset=["domain"]).reset_index(drop=True)
# 合并 DW 主数据的 customer_id，若缺失则根据域名生成稳定 ID
if not dw_customers.empty and "customer_id" in dw_customers.columns:
dw_map = dw_customers.copy()
dw_map["domain"] = dw_map["domain"].apply(normalize_domain)
dw_map = dw_map.dropna(subset=["domain"])
dim = all_cand.merge(dw_map[["domain", "customer_id", "customer_name", "website"]], on="domain", how="left", suffixes=("", "_dw"))
# 优先 DW 名称/网站
dim["customer_name"] = dim["customer_name_dw"].fillna(dim["customer_name"])
dim["website"] = dim["website_dw"].fillna(dim["website"])
dim = dim.drop(columns=[c for c in dim.columns if c.endswith("_dw")])
else:
dim = all_cand.copy()
dim["customer_id"] = None
# 生成缺失 ID
mask = dim["customer_id"].isna()
dim.loc[mask, "customer_id"] = dim.loc[mask, "domain"].apply(stable_id_from_domain)
# 规范列
dim["customer_name"] = dim["customer_name"].fillna(dim["domain"])
dim["website"] = dim["website"].apply(lambda x: x if isinstance(x, str) else None)
dim = dim[["customer_id", "customer_name", "domain", "website"]].copy()
return dim
def map_customer_id(df: pd.DataFrame, dim_customer: pd.DataFrame, domain_col: str, out_col: str = "customer_id") -> pd.DataFrame:
if df.empty:
df[out_col] = None
return df
tmp = df.copy()
tmp = tmp.merge(dim_customer[["domain", "customer_id"]], left_on=domain_col, right_on="domain", how="left")
tmp = tmp.drop(columns=["domain_y"], errors="ignore").rename(columns={"domain_x": domain_col})
# 对无域名匹配的记录，保持 customer_id 为缺失
return tmp
-------------------------
事实表构建
-------------------------
def build_fact_sales(crm_deals: pd.DataFrame, dim_customer: pd.DataFrame) -> pd.DataFrame:
if crm_deals.empty:
return pd.DataFrame(columns=["deal_id", "deal_name", "amount", "stage", "close_date", "owner", "customer_id", "account_domain"])
df = crm_deals.copy()
df = map_customer_id(df, dim_customer, domain_col="account_domain", out_col="customer_id")
cols = ["deal_id", "deal_name", "amount", "stage", "close_date", "owner", "customer_id", "account_domain"]
df = df[[c for c in cols if c in df.columns]]
return df
def build_fact_interactions(tickets: pd.DataFrame, ops_logs: pd.DataFrame, dim_customer: pd.DataFrame) -> pd.DataFrame:
frames = []
# 工单 -> 交互
if not tickets.empty:
t = tickets.copy()
t["interaction_type"] = "ticket"
t["occurred_at"] = t["created_at"] if "created_at" in t.columns else None
t["actor"] = t["requester_email"] if "requester_email" in t.columns else None
t["domain"] = t["requester_domain"]
t["payload"] = t.apply(lambda r: json.dumps({
"ticket_id": r.get("ticket_id"),
"subject": r.get("subject"),
"status": r.get("status"),
"priority": r.get("priority"),
"assignee": r.get("assignee"),
"channel": r.get("channel"),
}, ensure_ascii=False), axis=1)
t = map_customer_id(t, dim_customer, domain_col="domain", out_col="customer_id")
frames.append(t[["interaction_type", "occurred_at", "actor", "customer_id", "domain", "payload"]])
# 运营日志 -> 交互
if not ops_logs.empty:
o = ops_logs.copy()
o["interaction_type"] = o.get("event_type", "log")
o["actor"] = o.get("user_email")
# 优先 account_domain 其后 user_domain
dom = o.get("account_domain")
if dom is None:
dom = o.get("user_domain")
o["domain"] = dom
# payload 保留 metadata
o["payload"] = o.get("metadata")
o = map_customer_id(o, dim_customer, domain_col="domain", out_col="customer_id")
# 统一 occurred_at 列名
if "occurred_at" not in o.columns:
o["occurred_at"] = None
frames.append(o[["interaction_type", "occurred_at", "actor", "customer_id", "domain", "payload"]])
if not frames:
return pd.DataFrame(columns=["interaction_type", "occurred_at", "actor", "customer_id", "domain", "payload"])
inter = pd.concat(frames, ignore_index=True)
return inter
-------------------------
主流程
-------------------------
def main():
since_iso = SINCE_DT.isoformat()
logging.info(f"增量抽取起始时间：{since_iso}")
# 抽取
crm_contacts = fetch_crm_contacts(since_iso)
crm_deals = fetch_crm_deals(since_iso)
tickets = fetch_tickets(since_iso)
ops_logs = load_ops_logs(OPS_LOG_PATH)
dw_customers = fetch_dw_customers()
logging.info(f"CRM contacts: {len(crm_contacts)}, CRM deals: {len(crm_deals)}, tickets: {len(tickets)}, logs: {len(ops_logs)}, dw_customers: {len(dw_customers)}")
# 标准化
crm_contacts = standardize_crm_contacts(crm_contacts)
crm_deals = standardize_crm_deals(crm_deals)
tickets = standardize_tickets(tickets)
ops_logs = standardize_ops_logs(ops_logs)
# 实体解析（客户维表）
dim_customer = build_dim_customer(dw_customers, crm_contacts, crm_deals, tickets, ops_logs)
logging.info(f"dim_customer 行数：{len(dim_customer)}")
# 事实表
fact_sales = build_fact_sales(crm_deals, dim_customer)
fact_interactions = build_fact_interactions(tickets, ops_logs, dim_customer)
logging.info(f"fact_sales 行数：{len(fact_sales)}, fact_interactions 行数：{len(fact_interactions)}")
# 输出
safe_to_parquet_or_csv(dim_customer, os.path.join(OUTPUT_DIR, "dim_customer"))
safe_to_parquet_or_csv(fact_sales, os.path.join(OUTPUT_DIR, "fact_sales"))
safe_to_parquet_or_csv(fact_interactions, os.path.join(OUTPUT_DIR, "fact_interactions"))
# 如需将结果回写数据仓库，取消以下注释并确保 DW_SQLALCHEMY_URL 可写，且目标表结构匹配或允许 if_exists="replace"/"append"
"""
if DW_SQLALCHEMY_URL:
engine = create_engine(DW_SQLALCHEMY_URL)
with engine.begin() as conn:
dim_customer.to_sql("dim_customer_unified", conn, if_exists="replace", index=False)
fact_sales.to_sql("fact_sales_unified", conn, if_exists="replace", index=False)
fact_interactions.to_sql("fact_interactions_unified", conn, if_exists="replace", index=False)
logging.info("已将统一表写回数据仓库")
"""
logging.info("整合流程完成")
if name == "main":
main()
实施与校验建议
- 字段映射对齐：根据你的CRM、客服系统与日志字段，调整字段映射段落（fetch_crm_contacts/fetch_crm_deals/fetch_tickets/load_ops_logs 中的 field_map）。
- 分页与限流：实际API的分页参数名与返回结构可能不同，调整 fetch_paginated_list 的参数与 data_path。
- 实体解析覆盖率评估：输出 fact_sales 与 fact_interactions 中 customer_id 的缺失比例，并回看无法匹配的域名，优化数据质量（例如补齐网站、规范邮箱、维护域名别名映射）。
- 时间统一：确保所有时间戳统一为 UTC，避免跨系统对齐误差。
- 数据敏感性：输出时谨慎处理 PII；如需脱敏或字段白名单，请在构建 payload 时定制。
