客户发来一个小程序卡片或外部链接里面可能藏着高价值信号——小程序里的商品链接、公众号文章里提到的需求、表单链接里漏填的字段。程序能帮你把这些结构化信息沉淀下来运营不用手动看。一、消息解析从原始结构里提取可用字段小程序消息回调通常带 title、desc、path跳转路径、thumb_url。path 里藏着参数比如 ?product_id123要 URL 解码后提取。外部链接消息带 title网页标题和 urlurl 里可能有产品 ID、活动 ID 等结构化参数。提取出来存一条结构化记录发消息的人、发送时间、消息类型、提取到的所有字段、原始消息快照。原始快照留着以便以后字段补全或校验。二、链接 meta 抓取标题、摘要、图片外部链接除了抓 url 本身还应该实际访问一次抓 meta 标签——title、meta description、meta og:image。这些能让你知道客户发的到底是啥内容而不只是拿到一堆长串 url。抓取用 HTTP HEAD 解析 HTML head 部分别全量下载页面体。遇到需要 JS 渲染的页面SPAmeta 抓不到就退回只用 url 本身。三、沉淀入库和客户关联、做标签提取的结构化字段和客户库的客户关联——谁、什么时候、发了什么链接/小程序、里面包含什么。然后跑标签规则链接里带 pricing 打 价格关注 标签、小程序跳转路径带 product/耳机 打 耳机意向 标签。标签规则可以在后台配置新增规则不用改代码。结构化消息提取对照消息类型提取字段附加抓取典型沉淀小程序卡片title/path/desc-path 参数拆产品 ID外部链接url/titlemeta 抓取og:title/description公众号文章urlmeta 摘要文章主题关键词结构化提取实现import re, urllib.parse class StructuredExtractor: def __init__(self, store, tag_rules): self.store store self.tag_rules tag_rules # 可配置的标签规则 def extract(self, event): msg_type event.get(type) if msg_type miniprogram: record self._parse_miniprogram(event) elif msg_type in (url, link): record self._parse_external_link(event) else: return # 沉淀入库 record[from] event[from] record[ts] event[ts] record[msg_id] event[msg_id] self.store.save_structured(record) # 打标签 new_tags self._match_tags(record) if new_tags: self.store.add_tags(event[from], new_tags) def _parse_miniprogram(self, event): path event.get(path, ) # 从 path 里拆参数 params {} if ? in path: qs path.split(?, 1)[1] params dict(urllib.parse.parse_qsl(qs)) return { type: miniprogram, title: event.get(title, ), desc: event.get(desc, ), path: path, path_params: params, original: event, } def _parse_external_link(self, event): url event.get(url) or self._find_url(event.get(content, )) title event.get(title) or meta {} try: meta self._fetch_meta(url) except Exception: pass return { type: link, url: url, title: title or meta.get(title, ), meta: meta, original: event, } def _find_url(self, text): m re.search(rhttps?://[^\s], text or ) return m.group(0) if m else None def _fetch_meta(self, url): import requests r requests.head(url, timeout5, allow_redirectsTrue) if r.status_code 400: return {} html requests.get(url, timeout10).text[:4000] return { title: self._extract_tag(html, title), og_title: self._extract_meta(html, og:title), og_desc: self._extract_meta(html, og:description), } def _extract_tag(self, html, tag): ... def _extract_meta(self, html, name): ... def _match_tags(self, record): tags [] for rule in self.tag_rules.all(): if self._rule_match(rule, record): tags.append(rule[tag]) return list(set(tags)) def _rule_match(self, rule, record): field record.get(rule[field], ) return rule[pattern] in field落地建议小程序和外部链接消息能提就提——path 里的 URL 编码参数、url 本身的 query 参数、html head 里的 og:title/description 都抓一下meta 抓不到也能退回用 url 本身。每条结构化记录落库时和客户关联谁、什么时候再跑可配置的标签规则把客户标记成价格关注耳机意向等让沉淀的数据能复用。原始消息快照一起存以后加新字段不用回头找。微信侧的结构化消息接口由 Eyun这类个人微信API平台 提供解析提取与标签匹配逻辑在自建服务实现接口字段以平台开发文档为准。