Telegram Bot v20 启动时执行任务与信息获取指南

import logging
from telegram import Update, Bot
from telegram.ext import Application, ApplicationBuilder, ContextTypes, PicklePersistence

# 配置日志
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

# 替换为您的 Bot Token 和目标用户 ID
BOT_TOKEN = "YOUR_BOT_TOKEN"
TARGET_USER_ID = 123456789 # 替换为实际的用户ID，Bot 会向此用户发送启动消息
PERSISTENCE_FILE_PATH = "bot_data.pkl" # 持久化文件路径

async def post_init_handler(application: Application) -> None:
    """
    在 Bot 启动并构建 Application 之后，但在开始轮询之前执行的异步回调。
    """
    logger.info("Bot 启动中：执行 post_init_handler...")
    bot_id = application.bot.id # 获取 Bot 自身的 ID
    logger.info(f"Bot ID: {bot_id}")

    # 使用 application.bot 发送消息
    try:
        await application.bot.send_message(
            chat_id=TARGET_USER_ID,
            text=f"Bot 已启动！我的 ID 是：{bot_id}\n准备好接收更新。"
        )
        logger.info(f"成功向用户 {TARGET_USER_ID} 发送了启动消息。")
    except Exception as e:
        logger.error(f"发送启动消息失败: {e}")

async def post_stop_handler(application: Application) -> None:
    """
    在 Bot 停止后执行的异步回调。
    """
    logger.info("Bot 已停止：执行 post_stop_handler...")
    try:
        await application.bot.send_message(
            chat_id=TARGET_USER_ID,
            text="Bot 已停止运行。"
        )
        logger.info(f"成功向用户 {TARGET_USER_ID} 发送了停止消息。")
    except Exception as e:
        logger.error(f"发送停止消息失败: {e}")

def main() -> None:
    """主函数，设置并运行 Bot"""
    persistence_object = PicklePersistence(filepath=PERSISTENCE_FILE_PATH)

    application = (
        ApplicationBuilder()
        .token(BOT_TOKEN)
        .persistence(persistence=persistence_object)
        .post_init(post_init_handler)  # 注册启动前回调
        .post_stop(post_stop_handler)  # 注册停止后回调
        .build()
    )

    logger.info("Bot Application 已构建，即将开始轮询...")
    application.run_polling(allowed_updates=Update.ALL_TYPES) # 开始轮询
    logger.info("Bot 轮询已停止。")

if __name__ == "__main__":
    main()

# ... (前面的导入和 main 函数结构保持不变) ...

async def _load_chats_from_persistence(application: Application) -> list[tuple]:
    """
    模拟从持久化存储中加载 Bot 所属的聊天信息。
    实际应用中，您需要在此处实现从数据库或文件读取逻辑。
    """
    # 假设 persistence_object 存储了 chat_data
    # 实际中，您可能需要定义一个更复杂的结构来存储这些数据
    if application.persistence and application.persistence.bot_data:
        # 这是一个简化的示例，假设 chat_data 直接存储在 bot_data 中
        # 实际可能需要一个特定的键，如 application.persistence.bot_data.get("known_chats", [])
        known_chats = application.persistence.bot_data.get("known_chats", {})
        logger.info(f"从持久化加载了 {len(known_chats)} 个已知聊天。")
        # 转换为所需的格式 (chat id, username/title, chat type, is_owner, admin_rights)
        formatted_chats = []
        for chat_id, chat_info in known_chats.items():
            formatted_chats.append(
                (
                    chat_id,
                    chat_info.get("title", chat_info.get("username", "N/A")),
                    chat_info.get("type", "N/A"),
                    chat_info.get("is_owner", "N/A"), # 需要在 chat_member 更新中判断并存储
                    chat_info.get("admin_rights", "N/A") # 需要在 chat_member 更新中判断并存储
                )
            )
        return formatted_chats
    return []

async def post_init_handler(application: Application) -> None:
    """
    在 Bot 启动并构建 Application 之后，但在开始轮询之前执行的异步回调。
    """
    logger.info("Bot 启动中：执行 post_init_handler...")
    bot_id = application.bot.id
    logger.info(f"Bot ID: {bot_id}")

    # 1. 发送启动消息
    try:
        await application.bot.send_message(
            chat_id=TARGET_USER_ID,
            text=f"Bot 已启动！我的 ID 是：{bot_id}\n准备好接收更新。"
        )
        logger.info(f"成功向用户 {TARGET_USER_ID} 发送了启动消息。")
    except Exception as e:
        logger.error(f"发送启动消息失败: {e}")

    # 2. 获取并发送 Bot 已知聊天列表（从持久化存储中加载）
    known_chats_data = await _load_chats_from_persistence(application)
    if known_chats_data:
        message_parts = ["Bot 已知的聊天列表："]
        for chat_id, title, chat_type, is_owner, admin_rights in known_chats_data:
            message_parts.append(
                f"{chat_id},{title},{chat_type},{is_owner},{admin_rights}"
            )
        full_message = "\n".join(message_parts)
        try:
            await application.bot.send_message(
                chat_id=TARGET_USER_ID,
                text=full_message
            )
            logger.info(f"成功向用户 {TARGET_USER_ID} 发送了已知聊天列表。")
        except Exception as e:
            logger.error(f"发送已知聊天列表失败: {e}")
    else:
        logger.info("没有从持久化存储中找到已知的聊天数据。")

# ... (main 函数保持不变) ...

from telegram.ext import ChatMemberHandler, MessageHandler, filters

async def track_chats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """处理 chat_member 更新以维护 Bot 所属的聊天列表。"""
    if not update.chat_member:
        return

    chat = update.chat_member.chat
    old_member = update.chat_member.old_chat_member
    new_member = update.chat_member.new_chat_member

    bot_data = context.application.persistence.bot_data # 获取持久化存储的 bot_data
    if "known_chats" not in bot_data:
        bot_data["known_chats"] = {}

    chat_id = chat.id
    chat_info = {
        "title": chat.title,
        "username": chat.username,
        "type": chat.type,
        # ... 更多您想存储的聊天信息
    }

    if new_member.user.id == context.bot.id: # 针对 Bot 自身的成员状态变化
        if new_member.status == "member": # Bot 被添加到聊天
            bot_data["known_chats"][chat_id] = chat_info
            logger.info(f"Bot 加入了聊天: {chat.title} ({chat_id})")
        elif new_member.status == "administrator": # Bot 成为管理员
            bot_data["known_chats"][chat_id] = chat_info # 更新信息
            # 可以在这里解析并存储 admin_rights
            logger.info(f"Bot 在聊天 {chat.title} ({chat_id}) 中成为管理员。")
        elif new_member.status == "left" or new_member.status == "kicked": # Bot 被移除
            if chat_id in bot_data["known_chats"]:
                del bot_data["known_chats"][chat_id]
            logger.info(f"Bot 离开了聊天: {chat.title} ({chat_id})")

    # 确保持久化数据被保存
    context.application.persistence.flush()

def main() -> None:
    # ... (前面的 ApplicationBuilder 部分) ...
    application.add_handler(ChatMemberHandler(track_chats, ChatMemberHandler.MY_CHAT_MEMBER))
    # ... (run_polling 部分) ...