#!/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")
HOOK_STATUS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mmp_hook_status.json")
# 与 xposed-module / App MmpModuleInfo 同步
EXPECTED_MODULE_CODE = 5
EXPECTED_MODULE_NAME = "1.2.2"
_messages = deque(maxlen=MAX_MESSAGES)
_mmp_messages = deque(maxlen=MAX_MESSAGES)
_mmp_synced = {} # packetId -> packet dict（手机全量同步，落盘）
_mmp_hook_status = {
"liveCode": 0,
"liveName": "",
"liveAt": 0, # epoch ms
"installedCode": 0,
"installedName": "",
"source": "",
}
_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,
"autoBounceHistoryOnDisconnect": 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"]))
out["autoBounceHistoryOnDisconnect"] = bool(raw.get(
"autoBounceHistoryOnDisconnect", out["autoBounceHistoryOnDisconnect"]))
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)
content_for_mod = ""
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"
content_for_mod = item.get("content") or ""
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"
if content_for_mod:
_note_module_from_content(content_for_mod)
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()
user_id = ""
pool_id = ""
if " #rid=" in right:
before, _, rest = right.partition(" #rid=")
right = before.strip()
rid_part = rest.strip()
if " #pool=" in rid_part:
user_id, _, pool_rest = rid_part.partition(" #pool=")
user_id = user_id.strip()
pool_id = pool_rest.strip()
elif " " in rid_part:
user_id = rid_part.split(" ", 1)[0].strip()
else:
user_id = rid_part
if "#pool=" in user_id:
user_id, _, pool_id = user_id.partition("#pool=")
user_id = user_id.strip()
pool_id = pool_id.strip()
if " #pool=" in right and not pool_id:
before, _, pool_rest = right.partition(" #pool=")
right = before.strip()
pool_id = pool_rest.strip()
claim_time = ""
amount_raw = right
if "(" in right and ")" in right and not right.startswith("{"):
amt, _, rest = right.partition("(")
amount_raw = amt.strip()
claim_time = rest.split(")", 1)[0].strip()
amount_num = _normalize_money(amount_raw)
if nick or user_id:
claims.append({
"nickname": nick or "?",
"userId": user_id,
"poolId": pool_id,
"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:
key = c.get("userId") or c.get("nickname") or "?"
rows.append("{0}={1}".format(key, c.get("amount") or "0"))
rows.sort()
return "|".join(rows)
def _aggregate_claims(claims):
"""同一快照内按 userId（无则昵称）去重，取较大金额，禁止同人累加翻倍。"""
buckets = {}
order = []
for c in claims:
nick = c.get("nickname") or "?"
uid = (c.get("userId") or "").strip()
key = ("id:" + uid) if uid else ("n:" + nick)
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 key not in buckets:
buckets[key] = {
"nickname": nick,
"userId": uid,
"poolId": (c.get("poolId") or "").strip(),
"amount": amt,
"count": 1,
"claimTime": c.get("claimTime") or "",
}
order.append(key)
continue
b = buckets[key]
b["count"] += 1
if uid and not b.get("userId"):
b["userId"] = uid
if c.get("poolId") and not b.get("poolId"):
b["poolId"] = c.get("poolId") or ""
if amt >= b["amount"]:
b["amount"] = amt
if nick and nick != "?":
b["nickname"] = nick
if c.get("claimTime"):
b["claimTime"] = c.get("claimTime") or b["claimTime"]
if uid:
b["userId"] = uid
if c.get("poolId"):
b["poolId"] = c.get("poolId") or b.get("poolId") or ""
elif c.get("claimTime") and not b["claimTime"]:
b["claimTime"] = c.get("claimTime") or ""
result = []
for key in order:
b = buckets[key]
result.append({
"nickname": b["nickname"],
"userId": b.get("userId") or "",
"poolId": b.get("poolId") or "",
"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 _parse_mod_token(mod):
"""mod=1.2.0/3 → (name, code)"""
text = (mod or "").strip()
if not text:
return "", 0
name = text
code = 0
if "/" in text:
name, _, code_part = text.rpartition("/")
name = name.strip()
try:
code = int(code_part.strip())
except (TypeError, ValueError):
code = 0
return name, code
def _note_module_from_content(content):
if not content or "[MMP统计]" not in content:
return
head = content.strip().split("\n", 1)[0]
if "mod=" not in head:
return
raw = head.split("mod=", 1)[1]
for stop in (" | ", " src=", " "):
if stop in raw:
raw = raw.split(stop, 1)[0]
break
name, code = _parse_mod_token(raw)
_update_hook_status({
"liveCode": code,
"liveName": name,
"liveAt": int(time.time() * 1000),
"source": "message",
})
def _load_hook_status():
global _mmp_hook_status
try:
if not os.path.isfile(HOOK_STATUS_PATH):
return
with open(HOOK_STATUS_PATH, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict):
with _lock:
_mmp_hook_status.update({
"liveCode": int(data.get("liveCode") or 0),
"liveName": str(data.get("liveName") or ""),
"liveAt": int(data.get("liveAt") or 0),
"installedCode": int(data.get("installedCode") or 0),
"installedName": str(data.get("installedName") or ""),
"source": str(data.get("source") or ""),
})
except Exception as e:
print("load mmp hook status failed:", e)
def _save_hook_status():
try:
with _lock:
data = dict(_mmp_hook_status)
with open(HOOK_STATUS_PATH, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
print("save mmp hook status failed:", e)
def _update_hook_status(patch):
if not isinstance(patch, dict):
return
with _lock:
try:
incoming_live_at = int(patch.get("liveAt") or 0)
except (TypeError, ValueError):
incoming_live_at = 0
stale_live = (incoming_live_at > 0
and int(_mmp_hook_status.get("liveAt") or 0) > 0
and incoming_live_at < int(_mmp_hook_status.get("liveAt") or 0))
for k in ("liveCode", "liveName", "liveAt", "installedCode", "installedName", "source"):
if k not in patch:
continue
if stale_live and k in ("liveCode", "liveName", "liveAt", "source"):
continue
# 无心跳时不要用 0 覆盖已有运行版本
if incoming_live_at <= 0 and k in ("liveCode", "liveName", "liveAt"):
continue
val = patch[k]
if k in ("liveCode", "liveAt", "installedCode"):
try:
val = int(val or 0)
except (TypeError, ValueError):
val = 0
else:
val = str(val or "")
if k in ("installedCode",) and val <= 0:
continue
if k in ("installedName",) and not val:
continue
_mmp_hook_status[k] = val
_save_hook_status()
def _describe_hook_status():
with _lock:
st = dict(_mmp_hook_status)
installed = int(st.get("installedCode") or 0)
installed_name = st.get("installedName") or ""
live_code = int(st.get("liveCode") or 0)
live_name = st.get("liveName") or ""
live_at = int(st.get("liveAt") or 0)
now_ms = int(time.time() * 1000)
if installed > 0:
apk = "模块APK：" + (installed_name or str(installed))
if installed < EXPECTED_MODULE_CODE:
apk += "(旧，期望{0})".format(EXPECTED_MODULE_NAME)
else:
apk = "模块APK：未知（等手机同步）"
age = (now_ms - live_at) if live_at > 0 else -1
if age < 0:
live = "运行中：未检测到（请强停再开 TNG）"
ok = False
elif age > 3 * 60 * 1000:
live = "运行中：心跳偏旧({0}分钟前 {1})".format(
age // 60000, live_name or live_code or "?")
ok = False
elif live_code >= EXPECTED_MODULE_CODE:
live = "运行中：最新({0})".format(live_name or live_code)
ok = True
elif live_code > 0:
live = "运行中：旧版({0}，期望{1})".format(
live_name or live_code, EXPECTED_MODULE_NAME)
ok = False
else:
live = "运行中：未知"
ok = False
return {
"ok": ok,
"text": apk + " · " + live,
"expectedCode": EXPECTED_MODULE_CODE,
"expectedName": EXPECTED_MODULE_NAME,
"liveCode": live_code,
"liveName": live_name,
"liveAt": live_at,
"installedCode": installed,
"installedName": installed_name,
"ageMs": age,
}
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,
"userId": (row.get("userId") or "").strip(),
"poolId": (row.get("poolId") or "").strip(),
"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 ""
expires = raw.get("expiresAt") or raw.get("expire") or raw.get("expireTime") or ""
status = raw.get("status") or raw.get("activityStatus") or ""
finished = bool(raw.get("finished")) or _is_terminal_status(status) or _is_expire_passed(expires)
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,
"expiresAt": expires,
"status": status,
"updatedAt": raw.get("updatedAt") or "",
"fetchedAt": raw.get("updatedAt") or raw.get("fetchedAt") or _now_iso(),
"latestAt": issued or raw.get("updatedAt") or "",
"finished": finished,
"snapshots": int(raw.get("snapshots") or 1),
"claimedCount": int(raw.get("claimedCount") or 0),
"totalCount": int(raw.get("totalCount") or 0),
"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 []
hook = payload.get("hookStatus")
if isinstance(hook, dict):
_update_hook_status({
"liveCode": hook.get("liveCode") or hook.get("moduleVersionCode") or 0,
"liveName": hook.get("liveName") or hook.get("moduleVersionName") or "",
"liveAt": hook.get("liveAt") or hook.get("updatedAt") or int(time.time() * 1000),
"installedCode": hook.get("installedCode") or 0,
"installedName": hook.get("installedName") or "",
"source": "phone-sync",
})
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 _is_terminal_status(meta.get("status") or "") \
or _is_expire_passed(meta.get("expire") or meta.get("expireTime") or "")
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,
"expiresAt": meta.get("expire") or meta.get("expireTime") or "",
"status": meta.get("status") or "",
"fetchedAt": msg.get("receivedAt") or "",
"latestAt": issued or (msg.get("receivedAt") or ""),
"finished": finished,
"snapshots": 1,
"claims": claims,
}
_apply_counts_from_meta(item, meta)
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 ""
if not p.get("claimedCount"):
p["claimedCount"] = len(ranked)
if p.get("totalCount") and int(p.get("claimedCount") or 0) >= int(p.get("totalCount") or 0):
p["finished"] = True
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())
for p in result:
_refresh_packet_closed(p)
p["statusLabel"] = _packet_status_label(p)
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 _parse_mmp_time_ms(text):
s = (text or "").strip()
if not s:
return 0
m = re.match(r"^(\d{2})/(\d{2})/(\d{4})(?:\s+(\d{2}):(\d{2})(?::(\d{2}))?)?", s)
if m:
try:
dt = datetime(
int(m.group(3)), int(m.group(2)), int(m.group(1)),
int(m.group(4) or 23), int(m.group(5) or 59), int(m.group(6) or 59))
return int(dt.timestamp() * 1000)
except ValueError:
return 0
m = re.match(r"^(\d{4})-(\d{2})-(\d{2})(?:[ T](\d{2}):(\d{2})(?::(\d{2}))?)?", s)
if m:
try:
dt = datetime(
int(m.group(1)), int(m.group(2)), int(m.group(3)),
int(m.group(4) or 0), int(m.group(5) or 0), int(m.group(6) or 0))
return int(dt.timestamp() * 1000)
except ValueError:
return 0
if re.match(r"^\d{13}$", s):
return int(s)
if re.match(r"^\d{10}$", s):
return int(s) * 1000
return 0
def _is_terminal_status(status):
s = (status or "").upper()
if not s:
return False
return any(k in s for k in (
"FINISH", "COMPLETE", "EXPIRE", "ENDED", "CLOSED", "CANCEL", "DONE"))
def _is_expire_passed(expire_text):
ms = _parse_mmp_time_ms(expire_text)
return ms > 0 and int(time.time() * 1000) >= ms
def _refresh_packet_closed(p):
if not isinstance(p, dict):
return p
if p.get("finished"):
return p
if _is_terminal_status(p.get("status") or "") or _is_expire_passed(
p.get("expiresAt") or p.get("expire") or ""):
p["finished"] = True
return p
def _packet_status_label(p):
_refresh_packet_closed(p)
claimed = int(p.get("claimedCount") or 0)
total = int(p.get("totalCount") or 0)
board_n = len(p.get("leaderboard") or []) if isinstance(p.get("leaderboard"), list) else 0
claimed = max(claimed, board_n, int(p.get("claimantCount") or 0))
fully = total > 0 and claimed >= total
if fully:
progress = "已全部领取"
elif total > 0:
progress = "已领取 {0}/{1} 人".format(claimed, total)
elif claimed > 0:
progress = "已领取 {0} 人".format(claimed)
else:
progress = "未领取" if p.get("finished") else "领取中"
expired = _is_expire_passed(p.get("expiresAt") or "") or any(
k in (p.get("status") or "").upper() for k in ("EXPIRE", "ENDED", "CLOSED", "CANCEL"))
if expired and not fully:
return progress + " · 已过期"
return progress
def _apply_counts_from_meta(item, meta):
counts = (meta.get("counts") or "").strip()
if counts and "/" in counts:
left, right = counts.split("/", 1)
try:
item["claimedCount"] = int(left.strip() or 0)
except ValueError:
pass
try:
item["totalCount"] = int(right.strip() or 0)
except ValueError:
pass
board = item.get("claims") or item.get("leaderboard") or []
if not item.get("claimedCount") and isinstance(board, list):
item["claimedCount"] = len(board)
if item.get("totalCount") and item.get("claimedCount", 0) >= item["totalCount"]:
item["finished"] = True
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 = "