108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
查看与 BOT 的历史消息
|
|
"""
|
|
import json
|
|
import os
|
|
|
|
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_updates(offset=None, limit=100):
|
|
"""获取消息更新"""
|
|
params = {"limit": limit}
|
|
if offset:
|
|
params["offset"] = offset
|
|
response = requests.get(f"{BASE_URL}/getUpdates", params=params)
|
|
return response.json()
|
|
|
|
def main():
|
|
print("=" * 80)
|
|
print("查看历史消息")
|
|
print("=" * 80)
|
|
|
|
updates = get_updates()
|
|
|
|
if not updates.get('result'):
|
|
print("没有历史消息")
|
|
return
|
|
|
|
print(f"\n总共 {len(updates['result'])} 条消息\n")
|
|
|
|
# 按对话分组
|
|
conversations = {}
|
|
|
|
for update in updates['result']:
|
|
if 'message' in update:
|
|
msg = update['message']
|
|
chat_id = msg.get('chat', {}).get('id')
|
|
|
|
if chat_id not in conversations:
|
|
conversations[chat_id] = []
|
|
|
|
conversations[chat_id].append(msg)
|
|
|
|
# 显示每个对话
|
|
for chat_id, messages in conversations.items():
|
|
print(f"\n{'='*80}")
|
|
print(f"Chat ID: {chat_id}")
|
|
print(f"消息数: {len(messages)}")
|
|
print('='*80)
|
|
|
|
for msg in messages[-20:]: # 只显示最近20条
|
|
from_user = msg.get('from', {})
|
|
is_bot = from_user.get('is_bot', False)
|
|
username = from_user.get('username', 'unknown')
|
|
first_name = from_user.get('first_name', '')
|
|
|
|
sender = f"{'🤖 BOT' if is_bot else '👤 User'} ({username or first_name})"
|
|
|
|
text = msg.get('text', '')
|
|
photo = msg.get('photo')
|
|
document = msg.get('document')
|
|
|
|
print(f"\n{sender}:")
|
|
|
|
if text:
|
|
# 限制显示长度
|
|
if len(text) > 200:
|
|
print(f" {text[:200]}...")
|
|
else:
|
|
print(f" {text}")
|
|
|
|
if photo:
|
|
print(f" [图片消息]")
|
|
|
|
if document:
|
|
print(f" [文档: {document.get('file_name', 'unknown')}]")
|
|
|
|
if 'reply_markup' in msg:
|
|
markup = msg['reply_markup']
|
|
if 'inline_keyboard' in markup:
|
|
print(f" [内联键盘:]")
|
|
for row in markup['inline_keyboard']:
|
|
for btn in row:
|
|
print(f" - {btn.get('text')}")
|
|
if 'keyboard' in markup:
|
|
print(f" [回复键盘:]")
|
|
for row in markup['keyboard']:
|
|
for btn in row:
|
|
btn_text = btn.get('text') if isinstance(btn, dict) else str(btn)
|
|
print(f" - {btn_text}")
|
|
|
|
# 保存完整历史
|
|
with open("bot_message_history.json", "w", encoding="utf-8") as f:
|
|
json.dump(updates, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"\n\n✓ 完整历史已保存到 bot_message_history.json")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|