公域互动数据回流到 CRM 的链路设计：采集、清洗、实体对齐与归因

评论、私信、留资表单这些互动散在各平台后台里不回流进 CRM就只是一堆截图和导出的表格。这条链路真正的难点不在采集而在把同一个人在不同平台的身份合并起来——合并错了意向分、跟进记录、历史订单会全部串人。一、采集三种方式的取舍开放接口。稳定、字段规范代价是权限申请周期长且平台愿意给的字段未必是你要的字段。页面自动化采集。覆盖面广能拿到接口不给的内容代价是要持续维护选择器平台一改版就断。人工导出。只适合低频、小体量的场景适合做兜底不适合当主干。不管选哪种幂等必须在采集层就做掉。同一条评论被重复拉进来两次后面所有计数都会翻倍而且越往后越难查。import hashlib def dedup_key(platform, outer_id, content, ts): 同一条互动在任何一次重放中都得到同一个键 raw f{platform}|{outer_id}|{content.strip()}|{ts} return hashlib.md5(raw.encode()).hexdigest() def upsert(conn, rows): sql INSERT INTO raw_interaction (dedup_key, platform, outer_id, content, ts) VALUES (%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE contentVALUES(content) conn.executemany(sql, [(r[k], r[platform], r[outer_id], r[content], r[ts]) for r in rows])幂等键里带上平台和外站 ID重放、补采、多任务并发都不会重复入库。二、清洗四类脏数据在这里处理掉表情与控制字符。存进 UTF-8 库没问题导进表格就乱码落库前替换掉。联系方式。手机号、微信号出现时先脱敏再入库合规风险比数据价值高。时间口径。各平台返回的时间戳时区不一致统一按 UTC 存储展示时再转本地时区。刷屏。同一账号 10 秒内连续 5 条相同内容按一条计。import re PHONE re.compile(r1[3-9]\d{9}) def clean(text): t text.replace(\u200b, ).strip() # 去掉零宽字符 t re.sub(r[\U0001F000-\U0001FAFF], , t) # 去掉表情 t PHONE.sub([手机号], t) return t def burst_filter(rows, window_sec10, same_limit5): 同一账号短时间内的重复内容只保留一条 seen, out {}, [] for r in rows: k (r[outer_user], r[content]) if k in seen and r[ts] - seen[k] window_sec: continue seen[k] r[ts] out.append(r) return out清洗规则要写成可回放的函数而不是一次性脚本。平台侧补数据、修数据时重跑一遍就能对齐。三、实体对齐跨平台同一个人怎么合并这是整条链路里决定可用性的一步。做法分三层从强到弱依次使用。确定性规则。手机号、会员 ID、订单号这类强标识命中即合并不做打分。弱标识打分。昵称、头像哈希、城市、设备指纹加权算相似分。阈值裁决。超过上限合并低于下限新建落在中间的进人工复核队列。def match_score(a, b): if a.get(phone) and a[phone] b.get(phone): return 1.0 if a.get(member_id) and a[member_id] b.get(member_id): return 1.0 s 0.0 if a.get(nickname) and a[nickname] b.get(nickname): s 0.35 if a.get(avatar_hash) and a[avatar_hash] b.get(avatar_hash): s 0.25 if a.get(device) and a[device] b.get(device): s 0.25 if a.get(city) and a[city] b.get(city): s 0.15 return s def align(profiles, hi0.85, lo0.55): clusters, review [], [] for p in profiles: best, bi 0.0, None for i, c in enumerate(clusters): for q in c: sc match_score(p, q) if sc best: best, bi sc, i if best hi: clusters[bi].append(p) elif best lo: review.append((p, bi)) # 不自动合并交人工 else: clusters.append([p]) return clusters, review中间分数段不要自动合并。误合并的代价远高于漏合并两个人并成一个跟进记录串了之后很难回滚一线同事会直接不再信任这份名单。四、意图打标先打标再入库评论和私信的价值在于说了什么。轻量做法是用规则打标未命中的再交分类器不要一上来就全量跑大模型。INTENT_RULES [ (r多少钱|价格|报价|费用, price), (r怎么买|哪里买|下单, purchase), (r售后|坏了|退|换|投诉, complaint), (r合作|加盟|代理, biz), ] def tag(text): for pat, label in INTENT_RULES: if re.search(pat, text): return label, 1.0 return None, 0.0 # 未命中交分类器并记录置信度置信度低于阈值的标为待定不要硬塞一个标签。错误标签会直接带偏下游的名单分层。五、归因这条线索算谁的同一个人可能看过直播、点过评论、加过好友。归因规则要提前定死并写进文档不要每次对不上数再临时商量。def attribute(touches, order_time, half_life7): 时间衰减归因越接近成交的触达权重越高 w {} for t in touches: days (order_time - t[ts]).days if days 0 or days 30: continue w[t[channel]] w.get(t[channel], 0) 0.5 ** (days / half_life) total sum(w.values()) or 1 return {k: round(v / total, 3) for k, v in w.items()}常见口径有三种首次触达、末次触达、时间衰减。选哪一种没那么重要所有报表用同一种才重要——同一个月用两套口径谁都解释不清数字为什么差。六、落库与对账写入 CRM 前做两件事字段映射平台字段对 CRM 字段和写入幂等。之后每天跑一次对账。def reconcile(platform_cnt, raw_cnt, crm_cnt): assert raw_cnt platform_cnt, 原始条数超过平台侧采集逻辑有问题 assert crm_cnt raw_cnt, 落库条数超过原始条数去重失效 return {lost_in_collect: platform_cnt - raw_cnt, merged_by_align: raw_cnt - crm_cnt}平台侧条数、落库条数、合并后条数这三个数每天对一次。差异变大通常是平台改了接口分页或者限流早发现比事后补数省事得多。回流链路做得好不好不看接了几个平台只看一件事——一线同事打开 CRM 能不能直接开始跟进。