190 lines
8.2 KiB
Python
190 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""通过已登录抖音页的官方 IM SDK 读取私信、发送文本；不导航或点击用户主页。"""
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
from cdp_explicit import ( # pyright: ignore[reportMissingImports]
|
|
add_cdp_args,
|
|
configure,
|
|
evaluate_script,
|
|
)
|
|
from follow_user import validate_uid
|
|
|
|
EXPRESSION = r"""
|
|
(async () => {
|
|
const p = __PARAMS__;
|
|
const fail = code => { throw new Error(code); };
|
|
try {
|
|
if (location.origin !== 'https://www.douyin.com') fail('WRONG_ORIGIN');
|
|
const response = await fetch('/aweme/v1/web/user/profile/self/?device_platform=webapp&aid=6383',
|
|
{credentials: 'include', signal: AbortSignal.timeout(15000)});
|
|
if (!response.ok) fail('LOGIN_CHECK_FAILED');
|
|
const profile = await response.json();
|
|
if (profile.status_code !== 0 || !profile.user?.uid || profile.user.uid === '0')
|
|
fail('LOGIN_REQUIRED');
|
|
if (String(profile.user.uid) === p.uid) fail('SELF_TARGET');
|
|
|
|
// 按模块特征发现当前版本，不固定 chunk 名称、模块数字 ID 或账号。
|
|
let service;
|
|
for (const key of Object.keys(window).filter(k => k.startsWith('@pc-im/im:'))) {
|
|
const chunks = window[key];
|
|
if (!Array.isArray(chunks)) continue;
|
|
let req;
|
|
chunks.push([['python_im_' + crypto.randomUUID()], {}, r => { req = r; }]);
|
|
for (const [id, module] of Object.entries(req?.c || {})) {
|
|
if (!String(req.m[id]).includes('getOrCreatePrivateConversationByUid')) continue;
|
|
for (const exported of Object.values(module.exports || {})) {
|
|
if (exported?.instance?.imSdkService) service = exported.instance.imSdkService;
|
|
}
|
|
}
|
|
}
|
|
if (!service) fail('IM_SDK_NOT_READY');
|
|
const sdk = service.imSdkManager.getImSdkInstance();
|
|
const sendManager = service.sendMessageManager;
|
|
if (!sdk || !sendManager || typeof sendManager.createMessageBuilder !== 'function' ||
|
|
typeof sendManager.sendMessage !== 'function') fail('IM_SDK_NOT_READY');
|
|
const meta = c => ({id: String(c.id), short_id: String(c.shortId),
|
|
uid: String(c.toParticipantUserId), type: c.type});
|
|
const pack = m => ({server_id: String(m.serverId || ''), client_id: m.clientId || null,
|
|
sender: String(m.sender), type: m.type, content: m.content,
|
|
created_at: m.createdAt, server_status: m.serverStatus});
|
|
let conversation = sdk.getConversationList().find(c =>
|
|
c.type === 1 && String(c.toParticipantUserId) === p.uid);
|
|
if (p.action === 'send' && !p.confirm) {
|
|
return JSON.stringify({action: 'preview', sender_uid: String(profile.user.uid),
|
|
uid: p.uid, text: p.text, conversation: conversation ? meta(conversation) : null,
|
|
note: '未创建会话、未创建或发送消息；添加 --confirm 才会发送'});
|
|
}
|
|
if (!conversation && p.action === 'history') {
|
|
conversation = await service.conversationManager.getConversationByUidOrShortIdIfOnline(
|
|
{participantId: p.uid});
|
|
}
|
|
if (!conversation && p.action === 'send') {
|
|
conversation = await service.conversationManager.getOrCreatePrivateConversationByUid(p.uid);
|
|
}
|
|
if (!conversation) fail('CONVERSATION_NOT_FOUND');
|
|
if (conversation.type !== 1 || String(conversation.toParticipantUserId) !== p.uid)
|
|
fail('TARGET_MISMATCH');
|
|
if (p.action === 'history') {
|
|
const result = await sdk.getMessagesByConversation({conversation,
|
|
cursor: p.cursor, limit: p.limit});
|
|
if (!Array.isArray(result?.messages)) fail('HISTORY_FAILED');
|
|
return JSON.stringify({action: 'history', conversation: meta(conversation),
|
|
messages: result.messages.map(pack), has_more: result.hasMore,
|
|
next_cursor: result.cursor == null ? null : String(result.cursor)});
|
|
}
|
|
// 复用站内 sendMessageManager，不直接调用 sdk.sendMessage：站内流程会在发送前
|
|
// 更新 identity security headers，并处理官方一次性风控验证/重试回调。
|
|
const builder = sendManager.createMessageBuilder({
|
|
conversation, messageType: 7, enterMethod: 'douyin_accounts_worker'});
|
|
builder.toUid(p.uid).content(JSON.stringify({
|
|
aweType: 700, type: 0, richTextInfos: [], text: p.text}));
|
|
const result = await sendManager.sendMessage(builder);
|
|
const message = builder.getMessage();
|
|
if (!message) fail('MESSAGE_BUILD_FAILED');
|
|
let checkInfo = {};
|
|
if (typeof result?.checkMsg === 'string' && result.checkMsg) {
|
|
try { checkInfo = JSON.parse(result.checkMsg); } catch (_) { /* fixed fields below */ }
|
|
}
|
|
// 仅明确的 SDK 成功结果算成功；超时或未知状态不能由本程序自动重发。
|
|
// 不返回原始 checkMsg：它可能包含二次验证配置或其他敏感参数。
|
|
return JSON.stringify({action: 'send', success: result?.success === true,
|
|
status_code: result?.statusCode ?? null, check_code: String(result?.checkCode ?? ''),
|
|
check_message_present: Boolean(result?.checkMsg),
|
|
check_status_code: checkInfo?.status_code ?? null,
|
|
verification_required: Boolean(checkInfo?.decision_conf),
|
|
conversation: meta(conversation), message: pack(message)});
|
|
} catch (error) {
|
|
const known = ['WRONG_ORIGIN', 'LOGIN_CHECK_FAILED', 'LOGIN_REQUIRED', 'SELF_TARGET',
|
|
'IM_SDK_NOT_READY', 'CONVERSATION_NOT_FOUND', 'TARGET_MISMATCH', 'HISTORY_FAILED',
|
|
'MESSAGE_BUILD_FAILED'];
|
|
return JSON.stringify({error: known.includes(error.message) ? error.message : 'SDK_REQUEST_FAILED'});
|
|
}
|
|
})()
|
|
"""
|
|
|
|
|
|
def run_im(
|
|
action: str,
|
|
uid: str,
|
|
*,
|
|
text: str = "",
|
|
confirm: bool = False,
|
|
limit: int = 20,
|
|
cursor: str = "9223372036854775807",
|
|
) -> dict:
|
|
"""单次请求；历史不会标记已读，发送默认只预览。"""
|
|
validate_uid(uid)
|
|
if action not in ("history", "send"):
|
|
raise ValueError("action 必须为 history 或 send")
|
|
if not isinstance(cursor, str) or not cursor.isascii() or not cursor.isdecimal():
|
|
raise ValueError("cursor 必须是非负整数字符串")
|
|
normalized = cursor.lstrip("0") or "0"
|
|
if (
|
|
len(normalized) > 19
|
|
or (len(normalized) == 19 and normalized > "9223372036854775807")
|
|
or not 1 <= limit <= 100
|
|
):
|
|
raise ValueError("cursor 超出 int64 范围或 limit 不在 1..100")
|
|
if action == "send" and (not isinstance(text, str) or not text.strip()):
|
|
raise ValueError("消息不能为空")
|
|
params = {
|
|
"action": action,
|
|
"uid": uid,
|
|
"text": text,
|
|
"confirm": confirm,
|
|
"limit": limit,
|
|
"cursor": normalized,
|
|
}
|
|
expression = EXPRESSION.replace("__PARAMS__", json.dumps(params, ensure_ascii=True))
|
|
try:
|
|
result = evaluate_script(expression)
|
|
payload = json.loads(result.stdout.strip().splitlines()[-1])
|
|
except (subprocess.SubprocessError, OSError, ValueError, IndexError) as exc:
|
|
raise RuntimeError(
|
|
"浏览器调用失败：检查 CDP、已登录首页和 SDK；发送结果可能未知，勿自动重发"
|
|
) from exc
|
|
if not isinstance(payload, dict):
|
|
raise RuntimeError("浏览器返回格式异常；勿自动重发")
|
|
if payload.get("error"):
|
|
raise RuntimeError(
|
|
f"{payload['error']}：停止操作，请手动检查登录态/会话/SDK；勿自动重发"
|
|
)
|
|
return payload
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
|
|
sub = parser.add_subparsers(dest="action", required=True)
|
|
history = sub.add_parser("history", help="获取指定 UID 的一页历史消息")
|
|
history.add_argument("uid")
|
|
history.add_argument("--limit", type=int, default=20)
|
|
history.add_argument("--cursor", default="9223372036854775807")
|
|
send = sub.add_parser("send", help="发送文本，默认只预览")
|
|
send.add_argument("uid")
|
|
send.add_argument("--text", required=True)
|
|
send.add_argument("--confirm", action="store_true", help="确认实际发送一次")
|
|
add_cdp_args(history)
|
|
add_cdp_args(send)
|
|
args = parser.parse_args()
|
|
configure(args)
|
|
params = {
|
|
k: v
|
|
for k, v in vars(args).items()
|
|
if k not in ("cdp_url", "expected_uid", "page_url")
|
|
}
|
|
try:
|
|
result = run_im(**params)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return int(result.get("action") == "send" and not result.get("success"))
|
|
except (ValueError, RuntimeError) as exc:
|
|
print(f"失败：{exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|