From a0085ba9a20be4693b9290a0a4fe4a469113bd28 Mon Sep 17 00:00:00 2001 From: eric Date: Wed, 1 Apr 2026 14:22:29 -0500 Subject: [PATCH] 's' --- app/routers/chatbot.py | 79 ++++++++++++++- app/services/mp_recommend.py | 186 +++++++++++++++++++++++++++++++++++ 2 files changed, 264 insertions(+), 1 deletion(-) create mode 100644 app/services/mp_recommend.py diff --git a/app/routers/chatbot.py b/app/routers/chatbot.py index 8d957e8..566feb8 100644 --- a/app/routers/chatbot.py +++ b/app/routers/chatbot.py @@ -1,3 +1,4 @@ +import asyncio import base64 import json import logging @@ -12,9 +13,10 @@ from urllib.parse import parse_qs import httpx from Crypto.Cipher import AES from fastapi import APIRouter, BackgroundTasks, HTTPException, Request -from fastapi.responses import HTMLResponse, PlainTextResponse +from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse from ..config import PB_ADMIN_EMAIL, PB_ADMIN_PASSWORD, PB_URL +from ..services.mp_recommend import is_mp_recommend_webhook_body, try_persist_from_like_body router = APIRouter(prefix="/chatbot", tags=["chatbot"]) @@ -461,6 +463,81 @@ def chatbot_health(): return {"status": "ok", "module": "chatbot-clone", "profile": ACTIVE_PROFILE} +def _log_like_incoming(request: Request, body: bytes) -> None: + """SmsForwarder Webhook 调试：打印完整入参（公众号助手等通知转发）。""" + client_host = request.client.host if request.client else None + client_port = request.client.port if request.client else None + query = {k: v for k, v in request.query_params.multi_items()} + headers = {k: v for k, v in request.headers.items()} + ct = headers.get("content-type", "") or headers.get("Content-Type", "") + + logger.info( + "[like] ----- incoming POST ----- path=%s full_url=%s client=%s:%s query=%r", + request.url.path, + str(request.url), + client_host, + client_port, + query, + ) + logger.info("[like] content_type=%r content_length=%s body_bytes=%s", ct, headers.get("content-length"), len(body)) + logger.info("[like] headers=%s", headers) + + if not body: + logger.info("[like] body=(empty)") + return + + text = None + try: + text = body.decode("utf-8") + except UnicodeDecodeError: + hex_preview = body[:400].hex() + suffix = "..." if len(body) > 400 else "" + logger.info("[like] body=(non-utf8, hex prefix) %s%s", hex_preview, suffix) + return + + logger.info("[like] body_text=\n%s", text) + low = (ct or "").lower() + if "json" in low: + try: + parsed = json.loads(text) + logger.info("[like] body_json=%s", json.dumps(parsed, ensure_ascii=False, indent=2)) + except json.JSONDecodeError as e: + logger.info("[like] body_json_parse_failed: %s", e) + + +def _client_ip_for_like(request: Request) -> str: + xff = request.headers.get("x-forwarded-for") or request.headers.get("X-Forwarded-For") + if xff: + return xff.split(",")[0].strip() + if request.client: + return request.client.host or "" + return "" + + +@router.post("/like") +async def like_webhook_sink(request: Request): + """接收 SmsForwarder 等 Webhook POST，详细打日志；公众号助手通知写入 PocketBase。""" + body = await request.body() + _log_like_incoming(request, body) + ct = request.headers.get("content-type", "") or "" + payload: dict = {"ok": True, "received_bytes": len(body)} + try: + text = body.decode("utf-8") + except UnicodeDecodeError: + return JSONResponse(status_code=200, content=payload) + + if is_mp_recommend_webhook_body(text, ct): + try: + summary = await asyncio.to_thread(try_persist_from_like_body, text, ct, _client_ip_for_like(request)) + if summary: + payload["mp_recommend"] = summary + except Exception: + logger.exception("[like] PocketBase 写入 mp_recommend 失败") + payload["mp_recommend_error"] = "persist_failed" + + return JSONResponse(status_code=200, content=payload) + + @router.post("/api/private-message/query") async def chatbot_private_message_query(payload: dict): authtoken = str(payload.get("authtoken", "")).strip() diff --git a/app/services/mp_recommend.py b/app/services/mp_recommend.py new file mode 100644 index 0000000..10022bb --- /dev/null +++ b/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, + )