#!/usr/bin/env python3
"""notiMessage 本地调试转发服务 — 在 PC 浏览器查看手机抓取的消息。"""
import json
import os
import re
import threading
import time
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
SETTINGS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mmp_settings.json")
PACKETS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mmp_packets.json")
_messages = deque(maxlen=MAX_MESSAGES)
_mmp_messages = deque(maxlen=MAX_MESSAGES)
_mmp_synced = {} # packetId -> packet dict（手机全量同步，落盘）
_lock = threading.Lock()
_last_dedup = {"key": None, "ts": 0.0}
_mmp_last_dedup = {"key": None, "ts": 0.0}
_mmp_settings_lock = threading.Lock()
_DEFAULT_MMP_SETTINGS = {
"historyCooldownSec": 8,
"detailCooldownSec": 8,
"dedupSec": 3,
"detailGapMs": 80,
"pagePollMs": 1000,
"openHistoryIfNoTemplate": True,
}
_mmp_settings = dict(_DEFAULT_MMP_SETTINGS)
def _load_mmp_settings():
global _mmp_settings
try:
if os.path.isfile(SETTINGS_PATH):
with open(SETTINGS_PATH, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict):
merged = dict(_DEFAULT_MMP_SETTINGS)
merged.update(data)
_mmp_settings = _normalize_mmp_settings(merged)
except Exception as e:
print("load mmp settings failed:", e)
def _save_mmp_settings():
try:
with open(SETTINGS_PATH, "w", encoding="utf-8") as f:
json.dump(_mmp_settings, f, ensure_ascii=False, indent=2)
except Exception as e:
print("save mmp settings failed:", e)
def _normalize_mmp_settings(raw):
out = dict(_DEFAULT_MMP_SETTINGS)
if not isinstance(raw, dict):
return out
try:
out["historyCooldownSec"] = max(1, min(300, int(raw.get("historyCooldownSec", out["historyCooldownSec"]))))
out["detailCooldownSec"] = max(1, min(300, int(raw.get("detailCooldownSec", out["detailCooldownSec"]))))
out["dedupSec"] = max(0, min(120, int(raw.get("dedupSec", out["dedupSec"]))))
out["detailGapMs"] = max(0, min(5000, int(raw.get("detailGapMs", out["detailGapMs"]))))
out["pagePollMs"] = max(500, min(30000, int(raw.get("pagePollMs", out["pagePollMs"]))))
out["openHistoryIfNoTemplate"] = bool(raw.get(
"openHistoryIfNoTemplate", out["openHistoryIfNoTemplate"]))
except (TypeError, ValueError):
pass
return out
def _get_mmp_settings():
with _mmp_settings_lock:
return dict(_mmp_settings)
def _set_mmp_settings(raw):
global _mmp_settings
with _mmp_settings_lock:
_mmp_settings = _normalize_mmp_settings(raw)
_save_mmp_settings()
return dict(_mmp_settings)
_load_mmp_settings()
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()
dedup_key = "|".join([
str(payload.get("source") or ""),
str(payload.get("packageName") or ""),
str(payload.get("title") or ""),
str(payload.get("content") or ""),
])
now = time.time()
is_mmp = _is_mmp_message(item)
with _lock:
if is_mmp:
if (_mmp_last_dedup["key"] == dedup_key
and now - float(_mmp_last_dedup["ts"]) < 3.0):
if _mmp_messages:
return _mmp_messages[0]
_mmp_last_dedup["key"] = dedup_key
_mmp_last_dedup["ts"] = now
_mmp_messages.appendleft(item)
item["id"] = len(_mmp_messages)
channel = "MMP"
else:
if (_last_dedup["key"] == dedup_key
and now - float(_last_dedup["ts"]) < 3.0):
if _messages:
return _messages[0]
_last_dedup["key"] = dedup_key
_last_dedup["ts"] = now
_messages.appendleft(item)
item["id"] = len(_messages)
channel = "MSG"
print("[{0}] [{1}] [{2}] [{3}] {4} | {5}".format(
_now_iso(),
channel,
item["group"],
payload.get("source", "?"),
payload.get("appName", payload.get("packageName", "")),
(payload.get("content", "") or "")[:80],
))
return item
def _clear_mmp_messages():
with _lock:
_mmp_messages.clear()
_mmp_synced.clear()
# 兼容：顺带清掉旧版混入通用队列的 MMP
keep = [m for m in _messages if not _is_mmp_message(m)]
_messages.clear()
_messages.extend(keep)
_save_mmp_synced()
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:
if _is_mmp_message(msg):
continue
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 _is_mmp_message(msg):
source = (msg.get("source") or "").lower()
content = msg.get("content") or ""
title = msg.get("title") or ""
if "tng_mmp" in source or "xposed_tng_mmp" in source:
return True
if "[MMP统计]" in content or "Money Packet" in title:
return True
lower = content.lower()
return "receiverlist" in lower or ("claimedamount" in lower and "mmp" in lower)
def _normalize_money(value):
"""把 1.23 / RM1.23 / {"amount":"0.71","cent":"71"} 统一成 float 或 None。"""
if value is None:
return None
text = str(value).strip()
if not text or text.lower() == "null":
return None
text = text.replace("RM", "").replace("rm", "").strip()
if text.startswith("{") and "amount" in text:
try:
obj = json.loads(text)
if isinstance(obj, dict):
if obj.get("amount") not in (None, ""):
return float(str(obj["amount"]).replace(",", ""))
if obj.get("cent") not in (None, ""):
return int(str(obj["cent"])) / 100.0
except Exception:
return None
try:
return float(text.replace(",", ""))
except ValueError:
return None
def _parse_mmp_content(content):
"""解析 TngMoneyPacketHook 转发文本 → {meta, claims[]}。"""
text = (content or "").strip()
meta = {}
claims = []
if not text:
return meta, claims
lines = text.splitlines()
head = lines[0] if lines else ""
if head.startswith("[MMP统计]"):
head = head[len("[MMP统计]"):].strip()
# 新格式用 " | " 分隔；旧格式兼容空格拆 token
if " | " in head:
parts = [p.strip() for p in head.split(" | ") if p.strip()]
else:
parts = []
buf = ""
for part in head.split():
if buf:
buf += " " + part
if buf.count("{") <= buf.count("}"):
parts.append(buf)
buf = ""
continue
if "=" in part and part.split("=", 1)[1].startswith("{") and part.count("{") > part.count("}"):
buf = part
continue
parts.append(part)
if buf:
parts.append(buf)
for part in parts:
if "=" not in part:
continue
k, v = part.split("=", 1)
meta[k.strip()] = v.strip()
# 旧报文 sender=LIAO RUICHAO 被空格截断时，尝试从原文还原
if meta.get("sender") and "sender=" in head:
raw_sender = head.split("sender=", 1)[1]
for stop in (" | ", " total=", " via=", " group=", " packet=", " src="):
if stop in raw_sender:
raw_sender = raw_sender.split(stop, 1)[0]
break
raw_sender = raw_sender.strip()
if raw_sender and len(raw_sender) > len(meta.get("sender") or ""):
meta["sender"] = raw_sender
if "total" in meta:
total_num = _normalize_money(meta["total"])
meta["total"] = ("{0:.2f}".format(total_num) if total_num is not None else meta["total"])
# 发放时间：issued= / createTime= / issuedAt=
for key in ("issued", "createTime", "issuedAt", "create"):
if meta.get(key):
meta["issued"] = meta[key]
break
for line in lines[1:]:
line = line.strip()
if not line or "->" not in line:
continue
left, right = line.split("->", 1)
nick = left.strip()
right = right.strip()
claim_time = ""
if right.endswith(")") and "(" in right and not right.startswith("{"):
amt, _, rest = right.partition("(")
amount_raw = amt.strip()
claim_time = rest.rstrip(")").strip()
else:
amount_raw = right
amount_num = _normalize_money(amount_raw)
if nick:
claims.append({
"nickname": nick,
"amount": ("{0:.2f}".format(amount_num) if amount_num is not None else "0.00"),
"amountValue": amount_num if amount_num is not None else 0.0,
"claimTime": claim_time,
})
return meta, claims
def _claims_fingerprint(claims):
rows = []
for c in claims:
rows.append("{0}={1}".format(c.get("nickname") or "?", c.get("amount") or "0"))
rows.sort()
return "|".join(rows)
def _aggregate_claims(claims):
"""同一快照内同昵称只保留一笔（取较大金额），红包每人只领一次，禁止累加导致翻倍。"""
buckets = {}
order = []
for c in claims:
nick = c.get("nickname") or "?"
try:
amt = float(c.get("amountValue") if c.get("amountValue") is not None
else _normalize_money(c.get("amount")) or 0.0)
except (TypeError, ValueError):
amt = 0.0
if nick not in buckets:
buckets[nick] = {"nickname": nick, "amount": amt, "count": 1,
"claimTime": c.get("claimTime") or ""}
order.append(nick)
continue
b = buckets[nick]
b["count"] += 1
if amt >= b["amount"]:
b["amount"] = amt
if c.get("claimTime"):
b["claimTime"] = c.get("claimTime") or b["claimTime"]
elif c.get("claimTime") and not b["claimTime"]:
b["claimTime"] = c.get("claimTime") or ""
result = []
for nick in order:
b = buckets[nick]
result.append({
"nickname": nick,
"amount": round(b["amount"], 4),
"amountText": "{0:.2f}".format(b["amount"]),
"claimCount": b["count"],
"claimTime": b.get("claimTime") or "",
})
result.sort(key=lambda x: x["amount"], reverse=True)
for i, row in enumerate(result, 1):
row["rank"] = i
return result
def _packet_score(it):
board = it.get("leaderboard") or it.get("claims") or []
return (
1 if it.get("finished") else 0,
1 if it.get("issuedAt") else 0,
len(board),
it.get("updatedAt") or it.get("fetchedAt") or it.get("latestAt") or "",
)
def _normalize_synced_packet(raw):
if not isinstance(raw, dict):
return None
pid = str(raw.get("packetId") or raw.get("packet") or "").strip()
if not pid:
return None
board = raw.get("leaderboard") or []
if not isinstance(board, list):
board = []
norm_board = []
for i, row in enumerate(board):
if not isinstance(row, dict):
continue
amt = row.get("amount")
if amt is None:
amt = _normalize_money(row.get("amountText"))
try:
amt = float(amt or 0)
except (TypeError, ValueError):
amt = 0.0
nick = (row.get("nickname") or "?").strip() or "?"
norm_board.append({
"nickname": nick,
"amount": round(amt, 4),
"amountText": row.get("amountText") or "{0:.2f}".format(amt),
"claimTime": row.get("claimTime") or "",
"rank": int(row.get("rank") or (i + 1)),
})
norm_board.sort(key=lambda x: x["amount"], reverse=True)
for i, row in enumerate(norm_board, 1):
row["rank"] = i
best = norm_board[0] if norm_board else None
worst = norm_board[-1] if norm_board else None
sum_claimed = raw.get("sumClaimed")
try:
sum_claimed = float(sum_claimed) if sum_claimed is not None else round(
sum(x["amount"] for x in norm_board), 4)
except (TypeError, ValueError):
sum_claimed = round(sum(x["amount"] for x in norm_board), 4)
total = raw.get("total") or ""
total_num = _normalize_money(total)
if total_num is not None:
total = "{0:.2f}".format(total_num)
elif sum_claimed:
total = "{0:.2f}".format(sum_claimed)
issued = raw.get("issuedAt") or raw.get("issued") or ""
return {
"packetId": pid,
"title": raw.get("title") or "TNG 红包",
"sender": raw.get("sender") or "",
"group": raw.get("group") or "",
"total": total,
"via": raw.get("via") or "phone-sync",
"issuedAt": issued,
"updatedAt": raw.get("updatedAt") or "",
"fetchedAt": raw.get("updatedAt") or raw.get("fetchedAt") or _now_iso(),
"latestAt": issued or raw.get("updatedAt") or "",
"finished": bool(raw.get("finished")),
"snapshots": int(raw.get("snapshots") or 1),
"leaderboard": norm_board,
"claimantCount": len(norm_board) if norm_board else int(raw.get("claimantCount") or 0),
"sumClaimed": sum_claimed,
"bestNick": (best or {}).get("nickname") or "",
"bestAmount": (best or {}).get("amountText") or "",
"worstNick": (worst or {}).get("nickname") or "",
"worstAmount": (worst or {}).get("amountText") or "",
"fromSync": True,
}
def _load_mmp_synced():
global _mmp_synced
try:
if not os.path.isfile(PACKETS_PATH):
return
with open(PACKETS_PATH, "r", encoding="utf-8") as f:
data = json.load(f)
items = data.get("packets") if isinstance(data, dict) else data
if not isinstance(items, list):
return
synced = {}
for raw in items:
p = _normalize_synced_packet(raw)
if p:
synced[p["packetId"]] = p
with _lock:
_mmp_synced = synced
print("loaded mmp synced packets:", len(synced))
except Exception as e:
print("load mmp packets failed:", e)
def _save_mmp_synced():
try:
with _lock:
packets = list(_mmp_synced.values())
with open(PACKETS_PATH, "w", encoding="utf-8") as f:
json.dump({"packets": packets, "savedAt": _now_iso()}, f, ensure_ascii=False, indent=2)
except Exception as e:
print("save mmp packets failed:", e)
def _sync_mmp_packets(payload):
"""手机全量同步：合并进落盘镜像，电脑刷新 /mmp 即可看到。"""
items = []
if isinstance(payload, dict):
items = payload.get("packets") or []
elif isinstance(payload, list):
items = payload
if not isinstance(items, list):
return {"ok": False, "error": "packets must be list", "count": 0}
merged = 0
with _lock:
for raw in items:
p = _normalize_synced_packet(raw)
if not p:
continue
key = p["packetId"]
old = _mmp_synced.get(key)
if old is None or _packet_score(p) >= _packet_score(old):
if old and old.get("finished"):
p["finished"] = True
if old and not p.get("issuedAt") and old.get("issuedAt"):
p["issuedAt"] = old["issuedAt"]
p["latestAt"] = p["issuedAt"] or p.get("latestAt") or ""
_mmp_synced[key] = p
merged += 1
total = len(_mmp_synced)
_save_mmp_synced()
return {"ok": True, "count": merged, "total": total}
def _mmp_packets():
with _lock:
# 独立队列 + 兼容旧版混入通用消息的 MMP
msgs = list(_mmp_messages) + [m for m in _messages if _is_mmp_message(m)]
# 每个红包只保留「最新一次完整领取榜」快照
packets = {}
for msg in msgs:
meta, claims = _parse_mmp_content(msg.get("content") or "")
if not claims:
continue
packet_id = meta.get("packet") or meta.get("packetId") or ""
if not packet_id or packet_id in ("unknown", "TNG 红包", "TNG Money Packet"):
packet_id = "红包-" + _claims_fingerprint(claims)[:48]
key = str(packet_id)
issued = meta.get("issued") or ""
finished = str(meta.get("done") or "").lower() in ("1", "true") \
or str(meta.get("status") or "").upper() in ("FINISHED", "COMPLETE", "COMPLETED", "EXPIRED")
item = {
"packetId": key,
"title": msg.get("title") or "TNG 红包",
"sender": meta.get("sender") or "",
"group": meta.get("group") or "",
"total": meta.get("total") or "",
"via": meta.get("via") or "",
"issuedAt": issued,
"fetchedAt": msg.get("receivedAt") or "",
"latestAt": issued or (msg.get("receivedAt") or ""),
"finished": finished,
"snapshots": 1,
"claims": claims,
}
existing = packets.get(key)
def _score(it):
return (
1 if it.get("issuedAt") else 0,
len(it.get("claims") or []),
it.get("fetchedAt") or "",
)
if existing is None or _score(item) >= _score(existing):
if existing is not None:
item["snapshots"] = int(existing.get("snapshots") or 1) + 1
if not item.get("issuedAt") and existing.get("issuedAt"):
item["issuedAt"] = existing["issuedAt"]
item["latestAt"] = item["issuedAt"] or item.get("fetchedAt") or ""
if existing.get("finished"):
item["finished"] = True
packets[key] = item
else:
existing["snapshots"] = int(existing.get("snapshots") or 1) + 1
if not existing.get("issuedAt") and issued:
existing["issuedAt"] = issued
existing["latestAt"] = issued
if finished:
existing["finished"] = True
result = []
for p in packets.values():
ranked = _aggregate_claims(p.get("claims") or [])
p["leaderboard"] = ranked
p["claimantCount"] = len(ranked)
p["claimEventCount"] = len(p.get("claims") or [])
try:
p["sumClaimed"] = round(sum(x["amount"] for x in ranked), 4)
except Exception:
p["sumClaimed"] = 0
total_num = _normalize_money(p.get("total"))
if total_num is None or (p["sumClaimed"] > 0 and abs(total_num - p["sumClaimed"]) > 0.001
and total_num <= max((x["amount"] for x in ranked), default=0) + 1e-9):
p["total"] = "{0:.2f}".format(p["sumClaimed"])
elif total_num is not None:
p["total"] = "{0:.2f}".format(total_num)
best = ranked[0] if ranked else None
worst = ranked[-1] if ranked else None
p["bestNick"] = (best or {}).get("nickname") or ""
p["bestAmount"] = (best or {}).get("amountText") or ""
p["worstNick"] = (worst or {}).get("nickname") or ""
p["worstAmount"] = (worst or {}).get("amountText") or ""
del p["claims"]
result.append(p)
# 合并手机全量同步落盘数据（电脑重启后仍可看）
with _lock:
synced_items = list(_mmp_synced.values())
by_id = {p["packetId"]: p for p in result}
for sp in synced_items:
key = sp.get("packetId")
if not key:
continue
cur = by_id.get(key)
if cur is None or _packet_score(sp) >= _packet_score(cur):
merged = dict(sp)
if cur and cur.get("finished"):
merged["finished"] = True
if cur and not merged.get("issuedAt") and cur.get("issuedAt"):
merged["issuedAt"] = cur["issuedAt"]
merged["latestAt"] = merged["issuedAt"]
by_id[key] = merged
result = list(by_id.values())
result.sort(key=lambda x: _mmp_time_sort_key(x.get("issuedAt") or x.get("latestAt") or ""), reverse=True)
return result
def _mmp_time_sort_key(text):
"""把 13/07/2026 11:54:44 / ISO 等统一成可比较字符串。"""
s = (text or "").strip()
m = re.match(r"^(\d{2})/(\d{2})/(\d{4})(?:\s+(\d{2}:\d{2}(?::\d{2})?))?", s)
if m:
return "{0}-{1}-{2} {3}".format(m.group(3), m.group(2), m.group(1), m.group(4) or "")
return s
def _mmp_html_page(filename="mmp_page.html"):
page = os.path.join(os.path.dirname(os.path.abspath(__file__)), filename)
try:
with open(page, "r", encoding="utf-8") as f:
return f.read().encode("utf-8")
except Exception as e:
body = "