74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""测试 funstat text 搜索功能"""
|
|
|
|
import asyncio
|
|
import time
|
|
from datetime import datetime
|
|
from telethon import TelegramClient
|
|
|
|
# 配置
|
|
API_ID = 24660516
|
|
API_HASH = "eae564578880a59c9963916ff1bbbd3a"
|
|
SESSION_PATH = "/Users/lucas/telegram_sessions/funstat_bot"
|
|
BOT_USERNAME = "@openaiw_bot"
|
|
|
|
|
|
async def send_command_and_wait(client, bot_entity, command, timeout=15):
|
|
"""发送命令到 BOT 并等待响应"""
|
|
print(f"📤 发送命令: {command}")
|
|
|
|
# 记录发送前的最新消息 ID
|
|
last_message_id = 0
|
|
async for message in client.iter_messages(bot_entity, limit=1):
|
|
last_message_id = message.id
|
|
break
|
|
|
|
# 发送消息
|
|
await client.send_message(bot_entity, command)
|
|
|
|
# 等待响应
|
|
await asyncio.sleep(2)
|
|
|
|
# 获取新消息
|
|
start_time = time.time()
|
|
while time.time() - start_time < timeout:
|
|
async for message in client.iter_messages(bot_entity, limit=5):
|
|
if message.id > last_message_id:
|
|
if not message.out and message.text:
|
|
print(f"\n✅ 收到响应 ({len(message.text)} 字符):\n")
|
|
print("=" * 60)
|
|
print(message.text)
|
|
print("=" * 60)
|
|
return message.text
|
|
|
|
await asyncio.sleep(0.5)
|
|
|
|
raise TimeoutError(f"等待 BOT 响应超时 ({timeout}秒)")
|
|
|
|
|
|
async def main():
|
|
"""主函数"""
|
|
print("初始化 Telegram 客户端...")
|
|
|
|
# 创建客户端
|
|
client = TelegramClient(SESSION_PATH, API_ID, API_HASH)
|
|
await client.start()
|
|
|
|
# 获取 bot 实体
|
|
bot_entity = await client.get_entity(BOT_USERNAME)
|
|
print(f"✅ 已连接到: {bot_entity.first_name}\n")
|
|
|
|
# 发送 text 搜索命令
|
|
try:
|
|
await send_command_and_wait(client, bot_entity, "/text 翻译")
|
|
except Exception as e:
|
|
print(f"❌ 错误: {e}")
|
|
|
|
# 断开连接
|
|
await client.disconnect()
|
|
print("\n✅ 完成")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|