如何检查Telethon Bot运行状态？求补全代码实现跨文件检测及数据传网页

一、安装依赖 #

先安装所需的Python包：

pip install telethon redis flask

二、改造Telethon Bot（集成Redis） #

修改你的Bot代码，加入Redis用于存储消息和维护心跳：

from telethon.sync import TelegramClient, events
import cfg
import redis
import time
from threading import Thread

# 初始化Redis连接
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

api_id = cfg.API_ID
api_hash = cfg.API_HASH
bot = TelegramClient('bot', api_id, api_hash)

# 心跳线程：每隔10秒更新Redis中的心跳状态
def heartbeat():
    while True:
        r.setex('bot_heartbeat', 20, 'alive')  # 过期时间20秒，确保心跳中断后能检测到
        time.sleep(10)

# 启动心跳线程
Thread(target=heartbeat, daemon=True).start()

@bot.on(events.NewMessage)
async def my_event_handler(event):
    if 'hello' in event.raw_text:
        await event.reply('hi!')
    
    # 将收到的消息存入Redis列表，最多保留100条
    message_data = {
        'text': event.raw_text,
        'sender': str(event.sender_id),
        'time': time.strftime('%Y-%m-%d %H:%M:%S')
    }
    r.lpush('telegram_messages', str(message_data))
    r.ltrim('telegram_messages', 0, 99)  # 截断列表，只保留最新100条

bot.start()
bot.run_until_disconnected()

三、网页服务（Flask示例） #

创建一个web_server.py文件，从Redis读取消息并展示：

from flask import Flask, render_template_string
import redis

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 简单的网页模板
HTML_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
    <title>Telegram Bot Messages</title>
</head>
<body>
    <h1>Latest Telegram Messages</h1>
    <ul>
        {% for msg in messages %}
            <li>{{ msg }}</li>
        {% endfor %}
    </ul>
</body>
</html>
"""

@app.route('/')
def show_messages():
    # 从Redis获取最新20条消息
    messages = r.lrange('telegram_messages', 0, 19)
    return render_template_string(HTML_TEMPLATE, messages=messages)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

启动网页服务后，访问http://localhost:5000就能看到Bot收到的消息。

四、Bot运行状态检测脚本 #

创建check_bot_status.py文件，检测Bot是否在线：

import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def check_bot_alive():
    if r.exists('bot_heartbeat'):
        print("Bot is running")
        return True
    else:
        print("Bot is offline")
        return False

if __name__ == '__main__':
    check_bot_alive()

注意事项 #

• 确保Redis服务已启动并运行在本地默认端口（6379），如果Redis配置不同，修改代码中的连接参数。
• 心跳线程用了守护线程，会随Bot进程退出而终止，避免残留进程。
• 消息存储用Redis列表，适合按时间顺序保存最新消息，也可以根据需求改用其他数据结构。

内容的提问来源于stack exchange，提问作者Amirhossein shokri