""" 公众号助手（com.tencent.mp）「收到XXX的推荐」通知入库。 仅需 PocketBase 一个集合：mp_recommend_events（每个 mp_uid + recommender_key 一行，累计计数）。 - username Text 用户名（展示名，如从标题解析的 Eric） - cumulative_count Number 该用户的累计触发次数（同一 mp_uid + recommender_key 每次 webhook 写入时 = 旧值 + 1） - recommender_key Text 归一化键（unicode 归一化 + 清洗 + casefold），与 username 对应，便于筛选，建议建索引 - mp_uid Text 用户ID（用于把同名推荐人按不同账号隔离统计） 字段（在 PocketBase 中新建，名称一致）： - recommender_key Text - username Text - cumulative_count Number - package_name Text - title_line Text - summary_line Text - mp_uid Text - device_time_str Text - device_model Text - forward_timestamp_ms Number 可选 - raw_content Text - source_from Text - client_ip Text """ from __future__ import annotations import logging import re import unicodedata from typing import Any from urllib.parse import parse_qs from .pb import get_pb_client logger = logging.getLogger(__name__) COL_EVENTS = "mp_recommend_events" PACKAGE_MP = "com.tencent.mp" _TITLE_REC = re.compile(r"收到(.+?)的推荐") _UID_REC = re.compile(r"UID[：:]\s*(\S+)") def _pb_escape(s: str) -> str: return (s or "").replace("\\", "\\\\").replace('"', '\\"') def _normalize_username(raw: str) -> str: """ 尽量稳定地把展示名归一化到可用于统计/过滤的形式。 - 兼容 emoji/颜文字/特殊字母：保留字符本身，不做 ASCII 限制 - 统一空白：压缩为单个空格 - Unicode 归一化：NFKC（把全角等兼容形态归一） - 大小写：仅在 key 上 casefold """ s = (raw or "") s = unicodedata.normalize("NFKC", s) s = s.replace("\u00A0", " ") s = re.sub(r"\s+", " ", s).strip() return s def _username_to_key(username: str) -> str: u = _normalize_username(username) return u.casefold() if u else "unknown" def parse_form_urlencoded_like(body: str) -> dict[str, str]: q = parse_qs(body, keep_blank_values=True, strict_parsing=False) return { "from": (q.get("from") or [""])[0], "content": (q.get("content") or [""])[0], "timestamp": (q.get("timestamp") or [""])[0], } def parse_mp_recommend_content(content: str) -> dict[str, Any]: text = (content or "").replace("\r\n", "\n").replace("\r", "\n") lines = text.split("\n") stripped = [ln.strip() for ln in lines] package_name = stripped[0] if stripped else "" title_line = stripped[1] if len(stripped) > 1 else "" summary_line = stripped[2] if len(stripped) > 2 else "" device_time_str = stripped[4] if len(stripped) > 4 else "" device_model = stripped[5] if len(stripped) > 5 else "" recommender_name = "" m = _TITLE_REC.search(title_line or "") if m: recommender_name = (m.group(1) or "").strip() mp_uid = "" um = _UID_REC.search(text) if um: mp_uid = (um.group(1) or "").strip() return { "package_name": package_name, "recommender_name": recommender_name, "title_line": title_line, "summary_line": summary_line, "mp_uid": mp_uid, "device_time_str": device_time_str, "device_model": device_model, "raw_content": text, } def _parse_forward_ts_ms(raw: str) -> int | None: s = (raw or "").strip() if not s.isdigit(): return None try: return int(s) except ValueError: return None def persist_mp_recommend( *, source_from: str, content: str, timestamp_raw: str, client_ip: str, ) -> dict[str, Any]: """按 mp_uid + recommender_key 聚合更新累计次数（存在则 +1，不存在则创建）。""" parsed = parse_mp_recommend_content(content) forward_ts = _parse_forward_ts_ms(timestamp_raw) username = _normalize_username(parsed.get("recommender_name") or "") or "unknown" key = _username_to_key(username if username != "unknown" else "") mp_uid = (parsed.get("mp_uid") or "").strip() pb = get_pb_client() events = pb.collection(COL_EVENTS) filt_parts = [f'recommender_key = "{_pb_escape(key)}"'] if mp_uid: filt_parts.append(f'mp_uid = "{_pb_escape(mp_uid)}"') filt = " && ".join(filt_parts) lst = events.get_list(1, 1, {"filter": filt, "sort": "-updated"}) items = list(getattr(lst, "items", []) or []) existing = items[0] if items else None old_count = 0 existing_id = None if existing is not None: existing_id = getattr(existing, "id", None) try: old_count = int(getattr(existing, "cumulative_count", 0) or 0) except (TypeError, ValueError): old_count = 0 cumulative_count = old_count + 1 event_payload: dict[str, Any] = { "recommender_key": key, "username": username, "cumulative_count": cumulative_count, "package_name": parsed["package_name"], "title_line": parsed["title_line"], "summary_line": parsed["summary_line"], "mp_uid": mp_uid, "device_time_str": parsed["device_time_str"], "device_model": parsed["device_model"], "raw_content": parsed["raw_content"], "source_from": source_from, "client_ip": client_ip or "", } if forward_ts is not None: event_payload["forward_timestamp_ms"] = forward_ts if existing_id: rec = events.update(existing_id, event_payload) else: rec = events.create(event_payload) event_id = getattr(rec, "id", None) or str(rec) logger.info( "[like] mp_recommend stored event_id=%s username=%s cumulative_count=%s", event_id, username, cumulative_count, ) return { "event_id": event_id, "recommender_key": key, "username": username, "cumulative_count": cumulative_count, } def is_mp_recommend_webhook_body(body_text: str, content_type: str) -> bool: """是否为公众号助手 SmsForwarder 表单（可尝试入库）。""" low = (content_type or "").lower() if "application/x-www-form-urlencoded" not in low: return False fields = parse_form_urlencoded_like(body_text) if (fields.get("from") or "").strip() != PACKAGE_MP: return False return bool((fields.get("content") or "").strip()) def try_persist_from_like_body( body_text: str, content_type: str, client_ip: str, ) -> dict[str, Any] | None: """若为公众号助手 urlencoded 通知则入库并返回摘要；否则返回 None。""" low = (content_type or "").lower() if "application/x-www-form-urlencoded" not in low: return None fields = parse_form_urlencoded_like(body_text) source_from = (fields.get("from") or "").strip() if source_from != PACKAGE_MP: return None content = fields.get("content") or "" if not content.strip(): return None return persist_mp_recommend( source_from=source_from, content=content, timestamp_raw=fields.get("timestamp") or "", client_ip=client_ip, )