#!/usr/bin/env python3 """notiMessage 本地调试转发服务 — 在 PC 浏览器查看手机抓取的消息。""" import json import threading from collections import deque from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse HOST = "0.0.0.0" PORT = 8765 MAX_MESSAGES = 500 _messages = deque(maxlen=MAX_MESSAGES) _lock = threading.Lock() def _now_iso(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _resolve_group(payload): group = (payload.get("group") or payload.get("title") or "").strip() if group and "TLRPC$" not in group and "org.telegram.tgnet." not in group: return group app = payload.get("appName") or payload.get("packageName") or "" return app + " / 未分类" if app else "未分类" def _add_message(payload): item = dict(payload) item["group"] = _resolve_group(payload) item["receivedAt"] = _now_iso() with _lock: _messages.appendleft(item) item["id"] = len(_messages) print("[{0}] [{1}] [{2}] {3} | {4}".format( _now_iso(), item["group"], payload.get("source", "?"), payload.get("appName", payload.get("packageName", "")), (payload.get("content", "") or "")[:80], )) return item def _json_response(handler, status, data): body = json.dumps(data, ensure_ascii=False).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Access-Control-Allow-Origin", "*") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def _group_messages(messages): groups = {} for msg in messages: key = msg.get("group") or _resolve_group(msg) if key not in groups: groups[key] = { "key": key, "appName": msg.get("appName") or msg.get("packageName") or "", "count": 0, "latestAt": msg.get("receivedAt", ""), "messages": [], } g = groups[key] g["count"] += 1 g["messages"].append(msg) if (msg.get("receivedAt") or "") > (g.get("latestAt") or ""): g["latestAt"] = msg.get("receivedAt", "") result = sorted(groups.values(), key=lambda x: x.get("latestAt", ""), reverse=True) for g in result: g["messages"].sort(key=lambda m: m.get("receivedAt", ""), reverse=True) return result def _html_page(): html = """ notiMessage 调试台

notiMessage 调试台

0 条 0 群 -

请选择左侧群聊

""" return html.encode("utf-8") class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): if self.path.startswith("/api/messages") and self.command == "GET": return BaseHTTPRequestHandler.log_message(self, fmt, *args) def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): path = urlparse(self.path).path if path == "/": body = _html_page() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if path == "/api/messages": with _lock: data = list(_messages) _json_response(self, 200, data) return if path == "/api/groups": with _lock: data = _group_messages(list(_messages)) _json_response(self, 200, data) return if path == "/health": _json_response(self, 200, {"ok": True}) return _json_response(self, 404, {"error": "not found"}) def do_POST(self): path = urlparse(self.path).path if path not in ("/api/messages", "/api/debug/push", "/api/bills/app-upload"): _json_response(self, 404, {"error": "not found"}) return length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length) if length else b"{}" try: payload = json.loads(raw.decode("utf-8") or "{}") except ValueError: _json_response(self, 400, {"error": "invalid json"}) return if path == "/api/bills/app-upload": data = payload.get("data") or {} normalized = { "source": "app-upload", "packageName": payload.get("packageName", ""), "appName": payload.get("appName", ""), "title": data.get("title", ""), "content": data.get("context", data.get("content", "")), "timestamp": data.get("timestamp"), "raw": payload, } else: normalized = payload if not normalized.get("group"): normalized["group"] = _resolve_group(normalized) item = _add_message(normalized) _json_response(self, 200, {"ok": True, "id": item.get("id")}) def do_DELETE(self): path = urlparse(self.path).path if path != "/api/messages": _json_response(self, 404, {"error": "not found"}) return with _lock: _messages.clear() _json_response(self, 200, {"ok": True}) def main(): server = ThreadingHTTPServer((HOST, PORT), Handler) print("notiMessage debug server: http://127.0.0.1:{0}".format(PORT)) print("浏览器打开上述地址即可查看消息") print("手机经 USB 调试时先执行: adb reverse tcp:8765 tcp:8765") print("Wi-Fi 调试时将 AppConfig.DEBUG_SERVER_URL 改为 http://:8765") try: server.serve_forever() except KeyboardInterrupt: print("\nstopped") if __name__ == "__main__": main()