228 lines
7.2 KiB
Python
228 lines
7.2 KiB
Python
"""
|
||
公众号助手（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,
|
||
)
|