Untitled

from pyrogram.errors import exceptions as pyro_exceptions, rpc_error
from pyrogram.types import ChatPrivileges, ChatEventFilter
from pyrogram.enums import ChatMembersFilter

from pyrogram.raw import functions
from pyrogram import Client
from loguru import logger
import asyncio


filters = {
    "RESTRICTED": ChatMembersFilter.RESTRICTED,
    "ADMINS": ChatMembersFilter.ADMINISTRATORS,
    "BANNED": ChatMembersFilter.BANNED,
    "RECENT": ChatMembersFilter.RECENT
}


async def get_client(userbot_id: int, session_number: int, api_id: str, api_hash: str, bot_token: str):
    client_name = f"{userbot_id}_{session_number}"
    logger.debug(client_name)
    client = Client(
        client_name,
        api_id=api_id,
        api_hash=api_hash,
        bot_token=bot_token,
        workdir="sessions",
        no_updates=True
    )
    await client.start()
    logger.debug("done")
    return client


async def get_full_chat(client: Client, chat_id: int):
    entity = await client.resolve_peer(chat_id)
    full_chat = await client.invoke(functions.channels.GetFullChannel(channel=entity))
    return full_chat


async def get_chat_members(client: Client, chat_id: int, query: str, delay=1, max_retries=5, timeout=60):
    members = []  # Инициализируем список участников

    # Получение списка участников канала, удовлетворяющих текущему запросу
    for _ in range(max_retries):
        try:
            get_members_task = get_chat_members_(client, chat_id, query)

            # Ждем выполнения запроса с установленным временем ожидания
            members = await asyncio.wait_for(get_members_task, timeout)

            logger.debug(f"{chat_id}, {query}, {client.name} ({len(members)})")
            await asyncio.sleep(delay)
            break
        except pyro_exceptions.flood_420.FloodWait as e:
            logger.error(f"[{chat_id}] FloodWait: {e}")
            await asyncio.sleep(e.value)
        except pyro_exceptions.internal_server_error_500.RpcMcgetFail as e:
            logger.error(f"[{chat_id}] RpcMcgetFail: {e}")
            await asyncio.sleep(5)
        except (TimeoutError, asyncio.TimeoutError, rpc_error.RPCError) as e:
            logger.error(f"[{chat_id}] : {e}")
        except ConnectionError as e:
            logger.error(e)
            await asyncio.sleep(5)
        except OSError as e:
            logger.exception(f"RuntimeError: {e}")
            # await remove_client_from_cache(client)
            return query, []
        except pyro_exceptions.bad_request_400.ChannelInvalid as e:
            logger.error(f"[{chat_id}] ChannelInvalid: {e}")
            break

    return query, members


async def get_chat_members_(client: Client, chat_id: int, query: str):
    if query in filters:
        members = client.get_chat_members(chat_id, filter=filters[query])
    else:
        members = client.get_chat_members(chat_id, query=query)

    members_list = []
    counter = 0

    async for member in members:
        members_list.append(member)
        counter += 1

        # Логгируем после каждых 100 элементов
        if counter % 100 == 0:
            logger.debug(f"[{chat_id}] Processed {counter} members")

    return members_list


async def join_chat(client: Client, link: str):
    return await client.join_chat(link)


async def leave_chat(client: Client, chat_id: int):
    return await client.leave_chat(chat_id)


async def promote_chat_member(client: Client, chat_id: int, user_id: int):
    privileges = ChatPrivileges(
        can_invite_users=False,
        can_promote_members=False,
        can_restrict_members=True
    )
    result = await client.promote_chat_member(
        chat_id=chat_id,
        user_id=user_id,
        privileges=privileges
    )
    return result


async def get_users(client: Client, user_ids: list):
    user_ids = await asyncio.gather(*[client.resolve_peer(i) for i in user_ids])
    a = await client.invoke(
        functions.users.GetUsers(
            id=user_ids
        )
    )
    return a


async def get_full_user(client: Client, user_id: int):
    peer = await client.resolve_peer(user_id)
    data = await client.invoke(functions.users.GetFullUser(id=peer))
    return data


async def get_chat_events(client: Client, chat_id: int):
    filters = ChatEventFilter(
        new_members=True,
        leaving_members=True,
        new_restrictions=True,
        new_privileges=True
    )
    # Пример асинхронного генератора для получения событий из чата
    async for event in client.get_chat_event_log(chat_id, filters=filters):
        yield event


async def get_messages(client: Client, chat_id: int, message_ids: list):
    messages = await client.get_messages(chat_id, message_ids)
    return messages


async def get_chat(client: Client, chat_id: int):
    return await client.get_chat(chat_id)


async def get_join_requests(client: Client, chat_id: int):
    async for x in client.get_chat_join_requests(chat_id):
        yield x


async def get_chat_history(client: Client, chat_peer):
    async for post in client.get_chat_history(chat_peer):
        yield post