259 lines
8.9 KiB
Python
259 lines
8.9 KiB
Python
"""极简 protobuf wire 解码器（无第三方依赖）。
|
||
|
||
用于解析抖音 IM 发送私信的响应：官方 Response.proto 只建模了
|
||
create/get_info/new_message_notify 三种 body，没有“发送消息响应”，
|
||
导致仅凭 error_desc 为空就误判为发送成功。这里直接按 wire 格式解码，
|
||
读取真实的 status_code / server_message_id，判定是否真的投递成功。
|
||
"""
|
||
|
||
import json
|
||
from typing import Any, Optional
|
||
|
||
|
||
def _format_status_json(status_json: dict) -> str:
|
||
"""把内嵌 status JSON 格式化为更可读的失败原因。"""
|
||
code = status_json.get("status_code")
|
||
raw_check = status_json.get("raw_check_code")
|
||
decision = status_json.get("decision_type")
|
||
parts = [f"status_code={code}"]
|
||
if raw_check is not None:
|
||
parts.append(f"raw_check_code={raw_check}")
|
||
if decision:
|
||
parts.append(f"decision_type={decision}")
|
||
return "；".join(parts)
|
||
|
||
|
||
def _read_varint(buf: bytes, i: int) -> tuple[int, int]:
|
||
shift = 0
|
||
result = 0
|
||
n = len(buf)
|
||
while i < n:
|
||
b = buf[i]
|
||
i += 1
|
||
result |= (b & 0x7F) << shift
|
||
if not (b & 0x80):
|
||
return result, i
|
||
shift += 7
|
||
if shift > 70:
|
||
break
|
||
raise ValueError("truncated varint")
|
||
|
||
|
||
def decode_fields(buf: bytes) -> list[tuple[int, int, Any]]:
|
||
"""返回 [(field_num, wire_type, value), ...]。
|
||
|
||
wire_type: 0=varint(int), 1=64bit(int), 2=length-delimited(bytes), 5=32bit(int)
|
||
"""
|
||
out: list[tuple[int, int, Any]] = []
|
||
i = 0
|
||
n = len(buf)
|
||
while i < n:
|
||
key, i = _read_varint(buf, i)
|
||
field = key >> 3
|
||
wt = key & 7
|
||
if wt == 0:
|
||
val, i = _read_varint(buf, i)
|
||
out.append((field, wt, val))
|
||
elif wt == 2:
|
||
ln, i = _read_varint(buf, i)
|
||
val = buf[i:i + ln]
|
||
i += ln
|
||
out.append((field, wt, val))
|
||
elif wt == 5:
|
||
val = int.from_bytes(buf[i:i + 4], "little")
|
||
i += 4
|
||
out.append((field, wt, val))
|
||
elif wt == 1:
|
||
val = int.from_bytes(buf[i:i + 8], "little")
|
||
i += 8
|
||
out.append((field, wt, val))
|
||
else:
|
||
raise ValueError(f"unsupported wire type {wt}")
|
||
return out
|
||
|
||
|
||
def _collect_big_varints(buf: bytes, acc: list[int], depth: int = 0) -> None:
|
||
"""递归收集疑似 ID 的大整数（server_message_id / short_id 等都是大数）。"""
|
||
if depth > 6:
|
||
return
|
||
try:
|
||
fields = decode_fields(buf)
|
||
except Exception:
|
||
return
|
||
for _field, wt, val in fields:
|
||
if wt == 0 and isinstance(val, int) and val > 10 ** 12:
|
||
acc.append(val)
|
||
elif wt == 2 and isinstance(val, (bytes, bytearray)) and val:
|
||
_collect_big_varints(bytes(val), acc, depth + 1)
|
||
|
||
|
||
def _extract_status_json(raw: bytes) -> Optional[dict]:
|
||
"""抖音发送响应的 body 内嵌一段 JSON：{"status_code":x,"tips":"...","status_msg":{...}}。
|
||
|
||
这才是“消息是否真正投递”的权威结论（status_code=0 才是真成功）。
|
||
顶层 message=OK 只是接口层面的“已受理”，不代表已投递。
|
||
"""
|
||
marker = b'"status_code"'
|
||
idx = raw.find(marker)
|
||
if idx < 0:
|
||
return None
|
||
start = raw.rfind(b"{", 0, idx)
|
||
if start < 0:
|
||
return None
|
||
depth = 0
|
||
in_str = False
|
||
esc = False
|
||
end = -1
|
||
for i in range(start, len(raw)):
|
||
c = raw[i]
|
||
if in_str:
|
||
if esc:
|
||
esc = False
|
||
elif c == 0x5C: # backslash
|
||
esc = True
|
||
elif c == 0x22: # quote
|
||
in_str = False
|
||
continue
|
||
if c == 0x22:
|
||
in_str = True
|
||
elif c == 0x7B: # {
|
||
depth += 1
|
||
elif c == 0x7D: # }
|
||
depth -= 1
|
||
if depth == 0:
|
||
end = i + 1
|
||
break
|
||
if end < 0:
|
||
return None
|
||
try:
|
||
return json.loads(raw[start:end].decode("utf-8", "ignore"))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def analyze_send_response(raw: bytes) -> dict:
|
||
"""分析“发送私信”的 protobuf 响应，判定是否真正投递成功。
|
||
|
||
顶层 Response 字段（见 Response.proto）：
|
||
1=cmd, 2=sequence_id, 3=error_desc(string), 4=message(string),
|
||
5=inbox_type, 6=body(ResponseBody)。
|
||
|
||
注意：顶层没有 status_code 字段。真正“消息已写入服务端”的标志是
|
||
body(field 6) 里带有服务端分配的 server_message_id（大整数）。
|
||
sequence_id(field 2) 也是大整数，因此只在 body 内部查找 message_id，
|
||
避免把 sequence_id 误当成投递成功标志。
|
||
|
||
返回:
|
||
ok: 是否真正发送成功（body 内带服务端 message_id，或 message=OK 且有 body）
|
||
cmd: 顶层 cmd
|
||
message: 顶层 message 文本（field 4）
|
||
error_desc: 顶层 error_desc 文本（field 3）
|
||
server_message_id: body 内服务端消息 ID（投递成功的强信号）
|
||
has_body: 是否带 body
|
||
summary: 顶层字段概览 + hex 片段，便于排查
|
||
"""
|
||
info = {
|
||
"ok": False,
|
||
"cmd": None,
|
||
"status": None,
|
||
"status_code": None,
|
||
"raw_check_code": None,
|
||
"delivered_with_notice": False,
|
||
"status_reason": "",
|
||
"message": "",
|
||
"error_desc": "",
|
||
"server_message_id": None,
|
||
"has_body": False,
|
||
"summary": "",
|
||
}
|
||
if not raw:
|
||
info["summary"] = "空响应"
|
||
return info
|
||
try:
|
||
fields = decode_fields(raw)
|
||
except Exception as e:
|
||
info["summary"] = f"解码失败: {e}; hex={raw[:120].hex()}"
|
||
return info
|
||
|
||
body = None
|
||
parts = []
|
||
for field, wt, val in fields:
|
||
if field == 1 and wt == 0:
|
||
info["cmd"] = val
|
||
elif field == 3 and wt == 2:
|
||
try:
|
||
info["error_desc"] = bytes(val).decode("utf-8", "ignore")
|
||
except Exception:
|
||
pass
|
||
elif field == 4 and wt == 2:
|
||
try:
|
||
info["message"] = bytes(val).decode("utf-8", "ignore")
|
||
except Exception:
|
||
pass
|
||
elif field == 6 and wt == 2:
|
||
body = bytes(val)
|
||
info["has_body"] = len(body) > 0
|
||
|
||
if wt == 0:
|
||
parts.append(f"{field}=int:{val}")
|
||
elif wt == 2:
|
||
parts.append(f"{field}=bytes[{len(val)}]")
|
||
else:
|
||
parts.append(f"{field}={val}")
|
||
info["summary"] = " ".join(parts) + f" | hex={raw[:120].hex()}"
|
||
|
||
# 只在 body 内部查找服务端 message_id（避免误用顶层 sequence_id）
|
||
if body:
|
||
ids: list[int] = []
|
||
_collect_big_varints(body, ids)
|
||
if ids:
|
||
info["server_message_id"] = max(ids)
|
||
|
||
# 权威结论：body 内嵌 JSON 的 status_code（0 才是真成功）
|
||
status_json = _extract_status_json(raw)
|
||
if status_json is not None:
|
||
info["status_code"] = status_json.get("status_code")
|
||
info["raw_check_code"] = status_json.get("raw_check_code")
|
||
tips = (status_json.get("tips") or "").strip()
|
||
status_msg = status_json.get("status_msg")
|
||
msg_text = ""
|
||
if isinstance(status_msg, dict):
|
||
# 抖音把人类可读提示放在 status_msg.msg_content.tips
|
||
mc = status_msg.get("msg_content")
|
||
if isinstance(mc, dict):
|
||
msg_text = (mc.get("tips") or mc.get("content") or "").strip()
|
||
if not msg_text:
|
||
msg_text = (
|
||
status_msg.get("toast")
|
||
or status_msg.get("content")
|
||
or status_msg.get("msg")
|
||
or ""
|
||
)
|
||
elif isinstance(status_msg, str):
|
||
msg_text = status_msg
|
||
info["status_reason"] = tips or msg_text or _format_status_json(status_json)
|
||
|
||
msg_ok = info["message"].strip().upper() == "OK"
|
||
|
||
# 优先用 status_code 判定：明确给了 status_code 就以它为准（0=成功，非0=另判）
|
||
if info["status_code"] is not None:
|
||
if info["status_code"] == 0 and not info["error_desc"]:
|
||
info["ok"] = True
|
||
elif info["raw_check_code"] == 0 and msg_ok and not info["error_desc"]:
|
||
# raw_check_code=0 表示已通过抖音风控/安全校验；配合 message=OK，
|
||
# 说明消息已实际投递。此时非零 status_code 只是“业务侧提示”
|
||
# （如营销/陌生人限制提醒），对方仍能收到，不应判为发送失败。
|
||
info["ok"] = True
|
||
info["delivered_with_notice"] = True
|
||
else:
|
||
# raw_check_code=1（被风控拦截）或缺少 OK 标志：判为未送达
|
||
info["ok"] = False
|
||
else:
|
||
# 没有内嵌 status_code 时，退回“message=OK 且 body 内有服务端 message_id”
|
||
info["ok"] = bool(
|
||
not info["error_desc"]
|
||
and msg_ok
|
||
and info["server_message_id"] is not None
|
||
)
|
||
return info
|