's'

app/services/mp_recommend.py

@@ -0,0 +1,186 @@
+"""
+公众号助手（com.tencent.mp）「收到XXX的推荐」通知入库。
+
+仅需 PocketBase 一个集合：mp_recommend_events（每条通知一行）。
+
+- username            Text   用户名（展示名，如从标题解析的 Eric）
+- cumulative_count    Number 该用户截至本条通知的累计触发次数（每次 webhook 写入时 = 旧总数 + 1）
+- recommender_key     Text   归一化键（casefold），与 username 对应，便于筛选，建议建索引
+
+字段（在 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
+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 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]:
+    """写入一条事件；username、cumulative_count 写入本行（累计次数 = 同 key 已有条数 + 1）。"""
+    parsed = parse_mp_recommend_content(content)
+    forward_ts = _parse_forward_ts_ms(timestamp_raw)
+
+    username = (parsed["recommender_name"] or "").strip() or "unknown"
+    key = username.casefold() if username != "unknown" else "unknown"
+
+    pb = get_pb_client()
+    events = pb.collection(COL_EVENTS)
+    filt = f'recommender_key = "{_pb_escape(key)}"'
+    lst = events.get_list(1, 1, {"filter": filt})
+    total_before = int(getattr(lst, "total_items", 0) or 0)
+    cumulative_count = total_before + 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": parsed["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
+
+    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,
+    )