109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""发送「点赞比心」互动消息测试（aweType=11400）。
|
||
|
||
PowerShell 一行命令（请用 backend 虚拟环境 Python）:
|
||
.\.venv\Scripts\python.exe send_interaction_like_heart.py --account-id 1 --conversation-id "0:1:869032150442612:2461179734661520"
|
||
|
||
或双击运行: send_interaction_like_heart.bat
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import sqlite3
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
load_dotenv(ROOT.parent / ".env")
|
||
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from rpa_engine.credential import build_im_session_from_storage
|
||
from rpa_engine.device_profiles import resolve_user_agent
|
||
from rpa_engine.douyin_im.http_client import DouyinImHttpClient
|
||
from rpa_engine.douyin_im.interaction_messages import LIKE_HEART_SAMPLE
|
||
from rpa_engine.douyin_im.reply_payload import serialize_reply_content
|
||
from utils.cookie_store import read_cookie_file
|
||
|
||
|
||
def _load_account(account_id: int) -> dict:
|
||
db_path = ROOT / "kefu.db"
|
||
if not db_path.is_file():
|
||
raise SystemExit(f"数据库不存在: {db_path}")
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
row = conn.execute(
|
||
"SELECT id, cookie_data, im_session_data, user_agent FROM accounts WHERE id = ?",
|
||
(account_id,),
|
||
).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
raise SystemExit(f"账号 id={account_id} 不存在")
|
||
cookie_data = row["cookie_data"] or read_cookie_file(account_id)
|
||
if not cookie_data:
|
||
raise SystemExit(f"账号 id={account_id} 无 Cookie，请先登录采集凭证")
|
||
return {
|
||
"id": int(row["id"]),
|
||
"cookie_data": cookie_data,
|
||
"im_session_data": row["im_session_data"],
|
||
"user_agent": row["user_agent"],
|
||
}
|
||
|
||
|
||
def _build_session(account: dict):
|
||
storage = json.loads(account["cookie_data"]) if account["cookie_data"] else {}
|
||
session = build_im_session_from_storage(storage, account.get("im_session_data"))
|
||
session.user_agent = resolve_user_agent(account.get("user_agent") or session.user_agent)
|
||
return session
|
||
|
||
|
||
async def _send(account_id: int, conversation_id: str, content: str) -> None:
|
||
account = _load_account(account_id)
|
||
session = _build_session(account)
|
||
async with DouyinImHttpClient(session, account_id=account_id) as http:
|
||
ok = await http.send_text_message(conversation_id, content)
|
||
print("sent:", ok)
|
||
if not ok:
|
||
print(http.last_error or "发送失败")
|
||
raise SystemExit(1)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="发送抖音 IM 点赞比心互动消息")
|
||
parser.add_argument("--account-id", type=int, required=True)
|
||
parser.add_argument("--conversation-id", required=True)
|
||
parser.add_argument("--item-id", default=LIKE_HEART_SAMPLE["itemId"])
|
||
parser.add_argument("--message-type", type=int, default=8, help="protobuf message_type，默认 8")
|
||
parser.add_argument("--raw-file", help="完整 content JSON 文件（原样发送）")
|
||
args = parser.parse_args()
|
||
|
||
if args.raw_file:
|
||
with open(args.raw_file, encoding="utf-8") as f:
|
||
payload = json.load(f)
|
||
content = serialize_reply_content(
|
||
{
|
||
"type": "raw_im",
|
||
"message_type": args.message_type,
|
||
"payload": payload,
|
||
}
|
||
)
|
||
else:
|
||
content = serialize_reply_content(
|
||
{
|
||
"type": "interaction",
|
||
"variant": "like_heart",
|
||
"item_id": args.item_id,
|
||
"message_type": args.message_type,
|
||
}
|
||
)
|
||
|
||
print("content preview:", content[:200], "...")
|
||
asyncio.run(_send(args.account_id, args.conversation_id, content))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|