Pyrogram调用join_chat时链接无效问题求助

核心问题分析 #

1. Telegram频率限制：你遇到的30个链接有效后失效，大概率是触发了Telegram的加群频率限制，固定1.5秒的sleep不足以规避这个限制。
2. 链接解析错误：第二种方法直接用split("/")[-1]提取username会出错，因为部分链接是临时邀请链接（比如https://t.me/joinchat/XXXXXX）或者带参数（比如https://t.me/MyGroup?invite=XXX），提取的内容不是有效username，导致UsernameInvalid错误。
3. 已加入判断逻辑错误：第一种方法里me.id not in chat.members完全不对，Pyrogram的Chat对象的members属性不是用户ID列表，不能这么判断。

修正方案 #

1. 正确解析Telegram链接

区分两种链接类型处理：

• 私有邀请链接（比如https://t.me/joinchat/XXXXXX）：必须用完整链接加入
• 公开群组/频道链接（比如https://t.me/GroupName）：提取纯username，去掉多余参数

2. 调整频率控制策略

增加基础sleep时长，加入批量暂停机制，同时捕获FloodWait异常，严格按照Telegram要求的时间等待，避免触发更严格的限制。

3. 修正已加入群组的判断逻辑

用app.get_chat_member检查自身是否在群内，抛出UserNotParticipant异常则说明未加入。

修正后的代码示例 #

import asyncio
from urllib.parse import urlparse
from pyrogram import Client
from pyrogram.errors import UserNotParticipant, UsernameInvalid, FloodWait

async def main():
    await app.start()
    me = await app.get_me()
    
    # 读取并清理链接列表
    with open('links.txt', 'r', encoding='utf-8') as f:
        spic = [line.strip() for line in f if line.strip()]
    
    for idx, link in enumerate(spic, 1):
        try:
            # 解析链接，提取有效加入参数
            parsed_url = urlparse(link)
            if parsed_url.path.startswith('/joinchat/'):
                join_param = link
            else:
                # 提取纯username，去掉链接后的参数
                username = parsed_url.path.lstrip('/').split('?')[0]
                join_param = username
            
            # 检查是否已在群内
            try:
                chat = await app.get_chat(join_param)
                await app.get_chat_member(chat.id, me.id)
                print(f"[{idx}] 已在群组：{chat.title}")
                await app.archive_chats(chat.id)
                continue
            except UserNotParticipant:
                # 未加入则执行加群操作
                chat = await app.join_chat(join_param)
                print(f"[{idx}] 成功加入群组：{chat.title}")
                await app.archive_chats(chat.id)
                
                # 每处理10个群暂停30秒，降低触发限制的概率
                if idx % 10 == 0:
                    print("暂停30秒规避频率限制...")
                    await asyncio.sleep(30)
            
        except UsernameInvalid:
            print(f"[{idx}] 无效链接/用户名：{link}")
        except FloodWait as e:
            # 遇到官方限制，严格按照要求等待
            print(f"[{idx}] 触发频率限制，等待{e.value}秒...")
            await asyncio.sleep(e.value)
        except Exception as e:
            print(f"[{idx}] 处理失败：{link}，错误：{str(e)}")
        
        # 基础等待时间，可根据实际情况调整
        await asyncio.sleep(2)
    
    await app.stop()

if __name__ == "__main__":
    # 替换为你的客户端配置
    app = Client("my_account")
    asyncio.run(main())

额外注意事项 #

• 确保Pyrogram客户端的API ID、API Hash已正确配置，且账号处于正常状态。
• 如果仍频繁触发限制，可进一步延长基础sleep时间，或者分批次处理链接（比如每次处理40个，间隔5分钟再继续）。

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