Telegram Bot 启动时定制化操作与信息获取指南

from telegram import Update, Application
from telegram.ext import ApplicationBuilder, PicklePersistence
import asyncio

# 假设您有一个配置文件存储了bot_token和persistent_data_file_path
bot_token = "YOUR_BOT_TOKEN"
persistent_data_file_path = "bot_data.pkl"
TARGET_USER_ID = 123456789 # 替换为实际的用户ID

async def post_init_handler(application: Application) -> None:
    """
    在 Bot 启动前执行的初始化逻辑。
    """
    print("Bot 应用程序初始化中...")

    # 访问 Bot 实例并获取 Bot ID
    bot_info = await application.bot.get_me()
    print(f"Bot ID: {bot_info.id}, Username: @{bot_info.username}")

    # 向特定用户发送一条启动消息
    try:
        await application.bot.send_message(
            chat_id=TARGET_USER_ID,
            text=f"Bot 已启动！Bot ID: {bot_info.id}, Username: @{bot_info.username}"
        )
        print(f"启动消息已发送至用户 {TARGET_USER_ID}")
    except Exception as e:
        print(f"发送启动消息失败: {e}")

async def post_stop_handler(application: Application) -> None:
    """
    在 Bot 停止后执行的清理逻辑。
    """
    print("Bot 应用程序已停止。")

def main() -> None:
    persistence_object = PicklePersistence(filepath=persistent_data_file_path)
    application = (
        ApplicationBuilder()
        .token(bot_token)
        .persistence(persistence=persistence_object)
        .post_init(post_init_handler) # 注册 post_init_handler
        .post_stop(post_stop_handler) # 注册 post_stop_handler
        .build()
    )

    # run_polling() 将在 post_init_handler 执行完成后开始
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

from telegram import Update, Application, ChatMemberUpdated, Chat
from telegram.ext import ApplicationBuilder, PicklePersistence, ChatMemberHandler
import asyncio
import logging

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

bot_token = "YOUR_BOT_TOKEN"
persistent_data_file_path = "bot_data.pkl"
ADMIN_USER_ID = 123456789 # 替换为您的管理员用户ID

# 用于存储 Bot 已知聊天的字典 {chat_id: chat_info_tuple}
# chat_info_tuple 结构: (chat_id, chat_title, chat_type, is_bot_owner, bot_admin_rights)
known_chats = {}

async def post_init_handler(application: Application) -> None:
    """
    Bot 启动前的初始化逻辑。
    """
    logger.info("Bot 应用程序初始化中...")

    # 从持久化存储中加载已知的聊天列表
    # 注意：PTB 的 persistence 默认会处理 bot_data, chat_data, user_data
    # 如果 known_chats 是单独维护的，需要手动加载或集成到 bot_data 中
    if "known_chats" in application.bot_data:
        global known_chats
        known_chats = application.bot_data["known_chats"]
        logger.info(f"从持久化存储加载了 {len(known_chats)} 个已知聊天。")
    else:
        logger.info("未找到已知聊天列表，从头开始。")

    # 获取 Bot 自身信息
    bot_info = await application.bot.get_me()
    logger.info(f"Bot ID: {bot_info.id}, Username: @{bot_info.username}")

    # 准备要发送的启动信息
    startup_message_parts = [
        f"Bot 已启动！",
        f"Bot ID: {bot_info.id}",
        f"Username: @{bot_info.username}",
        "\n--- 已知聊天列表 ---"
    ]

    if known_chats:
        for chat_id, chat_data in known_chats.items():
            # 格式化聊天信息
            chat_id_str, title, chat_type, is_owner, admin_rights = chat_data
            admin_info = "(ignored)"
            if admin_rights:
                admin_info = f"拥有权限: {admin_rights.to_dict()}" # 假设 admin_rights 是 ChatMemberAdministrator 对象
            elif is_owner:
                admin_info = "是所有者"

            startup_message_parts.append(
                f"{chat_id_str},{title},{chat_type},{is_owner},{admin_info}"
            )
    else:
        startup_message_parts.append("无已知聊天。请等待 Bot 加入新的群组或频道。")

    # 将信息发送给管理员
    try:
        await application.bot.send_message(
            chat_id=ADMIN_USER_ID,
            text="\n".join(startup_message_parts)
        )
        logger.info(f"启动信息已发送至管理员 {ADMIN_USER_ID}")
    except Exception as e:
        logger.error(f"发送启动信息失败: {e}")

async def post_stop_handler(application: Application) -> None:
    """
    Bot 停止后的清理逻辑。
    """
    logger.info("Bot 应用程序已停止。")
    # 确保在停止前保存 known_chats 到持久化存储
    application.bot_data["known_chats"] = known_chats
    logger.info("已知聊天列表已保存到持久化存储。")


async def track_chats_handler(update: Update, context) -> None:
    """
    处理 ChatMemberUpdated 更新，维护 Bot 所在聊天列表。
    """
    chat_member: ChatMemberUpdated = update.chat_member

    # 确保是 Bot 自身的成员状态变化
    if chat_member.new_chat_member.user.id != context.bot.id:
        return

    chat: Chat = chat_member.chat
    chat_id = str(chat.id)
    chat_title = chat.title or chat.username or f"Private Chat with {chat.first_name}"
    chat_type = chat.type

    # 检查 Bot 的新状态
    new_status = chat_member.new_chat_member.status
    old_status = chat_member.old_chat_member.status if chat_member.old_chat_member else None

    is_bot_owner = False
    bot_admin_rights = None # None 或 ChatMemberAdministrator 对象

    if new_status == ChatMemberUpdated.MEMBER or new_status == ChatMemberUpdated.ADMINISTRATOR:
        # Bot 加入或成为管理员
        if new_status == ChatMemberUpdated.ADMINISTRATOR:
            bot_admin_rights = chat_member.new_chat_member.can_manage_chat # 示例，实际应获取具体权限
            # 进一步判断是否为所有者，通常通过 is_creator 字段判断
            if hasattr(chat_member.new_chat_member, 'is_creator') and chat_member.new_chat_member.is_creator:
                is_bot_owner = True

        # 存储或更新聊天信息
        known_chats[chat_id] = (chat_id, chat_title, chat_type, is_bot_owner, bot_admin_rights)
        logger.info(f"Bot 加入或更新了聊天: {chat_title} ({chat_id}), 类型: {chat_type}")

    elif new_status == ChatMemberUpdated.LEFT or new_status == ChatMemberUpdated.KICKED:
        # Bot 离开或被踢出
        if chat_id in known_chats:
            del known_chats[chat_id]
            logger.info(f"Bot 离开了聊天: {chat_title} ({chat_id})")

    # 将更新后的 known_chats 存储到 bot_data，以便持久化
    context.application.bot_data["known_chats"] = known_chats


def main() -> None:
    persistence_object = PicklePersistence(filepath=persistent_data_file_path)
    application = (
        ApplicationBuilder()
        .token(bot_token)
        .persistence(persistence=persistence_object)
        .post_init(post_init_handler)
        .post_stop(post_stop_handler)
        .build()
    )

    # 注册 ChatMemberHandler 来追踪 Bot 所在聊天
    application.add_handler(ChatMemberHandler(track_chats_handler, ChatMemberHandler.CHAT_MEMBER))

    logger.info("Bot 正在启动...")
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()