111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
探索 Telegram BOT 的功能
|
||
"""
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
import requests
|
||
|
||
from env_loader import load_env
|
||
|
||
load_env()
|
||
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
|
||
if not BOT_TOKEN:
|
||
raise RuntimeError("请在 .env 中设置 TELEGRAM_BOT_TOKEN")
|
||
BASE_URL = f"https://api.telegram.org/bot{BOT_TOKEN}"
|
||
|
||
def get_bot_info():
|
||
"""获取 BOT 基本信息"""
|
||
response = requests.get(f"{BASE_URL}/getMe")
|
||
return response.json()
|
||
|
||
def get_updates(offset=None):
|
||
"""获取消息更新"""
|
||
params = {"timeout": 30}
|
||
if offset:
|
||
params["offset"] = offset
|
||
response = requests.get(f"{BASE_URL}/getUpdates", params=params)
|
||
return response.json()
|
||
|
||
def send_message(chat_id, text):
|
||
"""发送消息"""
|
||
data = {
|
||
"chat_id": chat_id,
|
||
"text": text
|
||
}
|
||
response = requests.post(f"{BASE_URL}/sendMessage", json=data)
|
||
return response.json()
|
||
|
||
def get_bot_commands():
|
||
"""获取 BOT 设置的命令列表"""
|
||
response = requests.get(f"{BASE_URL}/getMyCommands")
|
||
return response.json()
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("开始探索 Telegram BOT...")
|
||
print("=" * 60)
|
||
|
||
# 1. 获取 BOT 信息
|
||
print("\n[1] BOT 基本信息:")
|
||
bot_info = get_bot_info()
|
||
print(json.dumps(bot_info, indent=2, ensure_ascii=False))
|
||
|
||
# 2. 获取 BOT 命令列表
|
||
print("\n[2] BOT 命令列表:")
|
||
commands = get_bot_commands()
|
||
print(json.dumps(commands, indent=2, ensure_ascii=False))
|
||
|
||
# 3. 获取最新消息
|
||
print("\n[3] 获取消息更新...")
|
||
updates = get_updates()
|
||
print(f"收到 {len(updates.get('result', []))} 条更新")
|
||
|
||
# 找到一个可用的 chat_id（如果有历史消息）
|
||
chat_id = None
|
||
if updates.get('result'):
|
||
latest_update = updates['result'][-1]
|
||
if 'message' in latest_update:
|
||
chat_id = latest_update['message']['chat']['id']
|
||
print(f"找到 chat_id: {chat_id}")
|
||
|
||
if not chat_id:
|
||
print("\n⚠️ 没有找到历史对话。")
|
||
print("请手动向 BOT 发送一条消息（任意内容），然后我会继续探索...")
|
||
print("等待 30 秒...")
|
||
time.sleep(30)
|
||
|
||
updates = get_updates()
|
||
if updates.get('result'):
|
||
latest_update = updates['result'][-1]
|
||
if 'message' in latest_update:
|
||
chat_id = latest_update['message']['chat']['id']
|
||
print(f"✓ 找到 chat_id: {chat_id}")
|
||
|
||
if chat_id:
|
||
# 发送测试命令
|
||
test_commands = ["/start", "/help"]
|
||
|
||
for cmd in test_commands:
|
||
print(f"\n[4] 测试命令: {cmd}")
|
||
result = send_message(chat_id, cmd)
|
||
print(f"发送结果: {result.get('ok')}")
|
||
time.sleep(2)
|
||
|
||
# 获取响应
|
||
updates = get_updates()
|
||
if updates.get('result'):
|
||
for update in updates['result'][-3:]:
|
||
if 'message' in update and update['message'].get('from', {}).get('is_bot'):
|
||
print(f"\nBOT 响应:")
|
||
print("-" * 40)
|
||
print(update['message'].get('text', ''))
|
||
print("-" * 40)
|
||
else:
|
||
print("\n❌ 无法获取 chat_id，请先与 BOT 对话")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|