如何判断Telegram用户是否允许被添加至群组？批量加群代码优化

核心思路 #

通过获取用户隐私设置中的can_be_added_to_groups字段，在执行添加操作前判断用户是否允许被加入群组，直接跳过无权限用户，避免无效请求、提升操作效率。

修改后的完整代码 #

import asyncio
import random
from pyrogram.errors import (
    FloodWait, UsernameNotOccupied, UserNotMutualContact, 
    PeerFlood, ChatAdminRequired, UserPrivacyRestrictedError,
    UserNotFound
)

addcount = 0

async def check_user_permission(client, user_id):
    """预检查用户是否允许被添加至群组"""
    try:
        user = await client.get_users(user_id)
        # 返回用户是否允许被添加到群组的布尔值
        return user.can_be_added_to_groups
    except UserNotFound:
        print("用户不存在")
        return False
    except UserPrivacyRestrictedError:
        print("用户隐私设置禁止获取信息，无法判断权限")
        return False
    except FloodWait as e:
        print(f"预检查触发FloodWait: 等待{e.seconds}秒")
        await asyncio.sleep(e.seconds)
        # 等待后重试一次预检查
        try:
            user = await client.get_users(user_id)
            return user.can_be_added_to_groups
        except Exception as e:
            print(f"重试预检查失败: {e}")
            return False
    except Exception as e:
        print(f"预检查出错: {e}")
        return False

async def add(client, user_id, chat_id):
    global addcount
    # 先执行预检查，不满足条件直接跳过
    can_add = await check_user_permission(client, user_id)
    if not can_add:
        print(f"用户{user_id}无法被添加，跳过")
        await asyncio.sleep(random.randint(2, 4))
        return
    
    try:
        result = await client.add_chat_members(chat_id, user_id)
        print(result)
        if result:
            addcount += 1
            print(f"已添加成员数 = {addcount}")
        await asyncio.sleep(random.randint(8, 12))
    except FloodWait as e:
        print(f"FloodWait: 等待{e.seconds}秒")
        await asyncio.sleep(e.seconds)
    except UsernameNotOccupied:
        print("用户名不存在!")
        await asyncio.sleep(random.randint(6, 10))
    except UserNotMutualContact:
        print("用户不是双向联系人!")
        await asyncio.sleep(random.randint(6, 10))
    except PeerFlood:
        print("PeerFlood: 账号受限!")
        await asyncio.sleep(random.randint(6, 10))
    except ChatAdminRequired:
        print("ChatAdminRequired: 需要管理员权限!")
        await asyncio.sleep(random.randint(6, 10))
    except UserPrivacyRestrictedError:
        print("用户隐私设置禁止被添加至群组")
        await asyncio.sleep(random.randint(2, 4))
    except Exception as e:
        print(f"意外错误: {e}")
        await asyncio.sleep(random.randint(6, 10))

关键说明 #

• 预检查函数：check_user_permission通过get_users接口获取用户对象，直接读取官方提供的权限字段判断，同时处理用户不存在、隐私限制、Flood等待等异常，保证预检查的稳定性。
• 前置判断逻辑：在add函数开头加入预检查，不满足条件的用户直接跳过，减少无效的加群请求，大幅提升批量操作的效率。
• 异常补充：新增UserPrivacyRestrictedError捕获，覆盖预检查未涉及的隐私限制场景，避免遗漏报错。

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