gitlab-instance-0a899031_payjsapi/app/services/mp_recommend.py

"""
公众号助手（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,
    )