Telegram电脑版“消息批量操作”脚本编写实战：基于TDLib实现自动化管理

import tdlib
import asyncio
import json

# 从环境变量或配置文件中读取你的api_id和api_hash
API_ID = 你的api_id
API_HASH = "你的api_hash"

# 初始化TDLib客户端
client = tdlib.Client(api_id=API_ID, api_hash=API_HASH)

async def main():
    # 启动客户端
    await client.start()
    print("客户端已启动。")
    # ... 后续操作代码将写在这里

    # 运行结束后，确保停止客户端
    await client.stop()

if __name__ == "__main__":
    asyncio.run(main())

async def main():
    await client.start()
    
    # 检查授权状态
    auth_state = await client.get_authorization_state()
    
    if isinstance(auth_state, tdlib.AuthorizationStateWaitPhoneNumber):
        phone_number = input("请输入您的电话号码（国际格式，如+8613012345678）: ")
        await client.send_phone_number(phone_number)
        print("验证码已发送。")
    elif isinstance(auth_state, tdlib.AuthorizationStateWaitCode):
        code = input("请输入您收到的验证码: ")
        await client.send_auth_code(code)
    elif isinstance(auth_state, tdlib.AuthorizationStateWaitPassword):
        password = input("请输入您的两步验证密码: ")
        await client.send_auth_password(password)
    elif isinstance(auth_state, tdlib.AuthorizationStateReady):
        print("用户已登录！")
        me = await client.get_me()
        print(f"登录身份: {me.first_name} {me.last_name} (@{me.usernames.active_usernames[0] if me.usernames else '无'})")
    else:
        print(f"未知的授权状态: {auth_state}")
        await client.stop()
        return
    
    # 确认登录成功
    while True:
        auth_state = await client.get_authorization_state()
        if isinstance(auth_state, tdlib.AuthorizationStateReady):
            break
        await asyncio.sleep(1)
    
    print("登录流程完成，准备执行批量操作。")
    # 此时，client已处于就绪状态，可以调用各种方法。

async def bulk_send_messages(client, chat_ids, message_text, delay_seconds=1):
    """
    向多个聊天批量发送相同文本消息。
    :param client: TDLib 客户端实例
    :param chat_ids: 聊天ID列表
    :param message_text: 要发送的文本
    :param delay_seconds: 每条消息之间的延迟（秒），避免触发风控
    """
    sent_count = 0
    failed_chats = []
    
    for chat_id in chat_ids:
        try:
            # 构造输入消息内容
            input_message_content = tdlib.InputMessageText(
                text=tdlib.FormattedText(text=message_text),
                disable_web_page_preview=True,
                clear_draft=False
            )
            # 发送消息
            sent_message = await client.send_message(
                chat_id=chat_id,
                input_message_content=input_message_content
            )
            print(f"消息已成功发送至聊天 {chat_id}。消息ID: {sent_message.id}")
            sent_count += 1
            
            # 添加延迟，非常重要！
            if delay_seconds > 0:
                await asyncio.sleep(delay_seconds)
                
        except Exception as e:
            print(f"向聊天 {chat_id} 发送消息失败: {e}")
            failed_chats.append(chat_id)
    
    print(f"\n批量发送完成。成功: {sent_count}, 失败: {len(failed_chats)}")
    if failed_chats:
        print(f"失败的聊天ID: {failed_chats}")

# 在主函数中调用示例
# 假设我们已经获取了目标聊天ID列表 target_chat_ids
# await bulk_send_messages(client, target_chat_ids, "大家好，这是一条批量测试消息。", delay_seconds=2)

async def bulk_forward_messages(client, from_chat_id, message_ids, to_chat_ids, delay_seconds=1):
    """
    将指定消息批量转发到多个聊天。
    :param client: TDLib 客户端实例
    :param from_chat_id: 源聊天ID
    :param message_ids: 要转发的消息ID列表
    :param to_chat_ids: 目标聊天ID列表
    :param delay_seconds: 每次转发操作间的延迟
    """
    for to_chat_id in to_chat_ids:
        try:
            # 使用forward_messages方法
            await client.forward_messages(
                chat_id=to_chat_id,
                from_chat_id=from_chat_id,
                message_ids=message_ids,
                disable_notification=True, # 静默转发，不通知目标聊天成员
                send_copy=False # 直接转发，而非复制
            )
            print(f"已批量转发 {len(message_ids)} 条消息至聊天 {to_chat_id}。")
            
            if delay_seconds > 0:
                await asyncio.sleep(delay_seconds)
                
        except Exception as e:
            print(f"向聊天 {to_chat_id} 转发消息失败: {e}")

async def bulk_delete_messages(client, chat_id, message_ids, revoke=True):
    """
    在指定聊天中批量删除消息。
    :param client: TDLib 客户端实例
    :param chat_id: 目标聊天ID
    :param message_ids: 要删除的消息ID列表
    :param revoke: 是否同时撤回（为True时，对方设备上的消息也会被删除）
    """
    # TDLib的deleteMessages方法一次可删除多个消息
    try:
        # 注意：message_ids需要是整数列表
        result = await client.delete_messages(chat_id=chat_id, message_ids=message_ids, revoke=revoke)
        if result:
            print(f"成功在聊天 {chat_id} 中删除了 {len(message_ids)} 条消息。")
    except tdlib.Error as e:
        # 处理特定错误，如无权限删除他人消息
        if e.code == 403:
            print(f"错误: 无权删除聊天 {chat_id} 中的部分或全部消息。")
        else:
            print(f"删除消息时发生未知错误: {e}")
    except Exception as e:
        print(f"操作异常: {e}")

# 使用示例：假设你管理一个群组，需要清理过去24小时内的广告消息ID列表
# await bulk_delete_messages(client, group_chat_id, spam_message_ids, revoke=True)

async def bulk_mark_as_read(client, chat_id, message_ids):
    """批量将消息标记为已读"""
    if message_ids:
        # 将消息列表中的最后一条消息ID标记为已读，会同时将其之前的消息也标记为已读
        last_read_id = max(message_ids)
        await client.view_messages(chat_id=chat_id, message_ids=[last_read_id], force_read=True)
        print(f"已将聊天 {chat_id} 中至消息ID {last_read_id} 的消息标记为已读。")

async def bulk_pin_messages(client, chat_id, message_ids, disable_notification=True):
    """批量置顶消息（注意：大多数聊天通常只允许置顶一条，此示例展示逻辑）"""
    # 实际情况中，通常是一次置顶一条。这里展示循环逻辑。
    for msg_id in message_ids:
        try:
            await client.pin_chat_message(
                chat_id=chat_id,
                message_id=msg_id,
                disable_notification=disable_notification
            )
            print(f"已置顶消息 ID: {msg_id} 在聊天 {chat_id} 中。")
            await asyncio.sleep(0.5) # 简短延迟
        except tdlib.Error as e:
            if e.code == 400 and "CHAT_NOT_MODIFIED" in e.message:
                print(f"消息 {msg_id} 可能已是置顶状态，或已达到置顶数量限制。")
            else:
                print(f"置顶消息 {msg_id} 失败: {e}")