modules/client.py - main - volkonsky13/HomeChat

modules/client.py
53 строки2 KB
Maximilian
Fix bug for messages
11 окт 2024, 13:35
11 окт 2024, 13:3500dab97
from telethon import TelegramClient
import tracemalloc


tracemalloc.start()


class ClientTlg:
    def __init__(self, phone, api_id, api_hash):
        self.client = TelegramClient(phone, api_id, api_hash)

    def get_messages(self, dialog_name: str, min_id: int = None) -> dict:
        messages = {"messages": []}
        users = {}
        ids = []
        with self.client.start() as cl:
            chats = cl.iter_dialogs()
            for chat in chats:
                if chat.name == dialog_name:
                    dialog = chat
                    break
            if min_id:
                mess = cl.iter_messages(dialog, min_id=min_id)
            else:
                mess = cl.iter_messages(dialog, limit=100)
            for mes in mess:
                ids.append(mes.id)
                if mes.message is not None and mes.message != '':
                    if mes.from_id.user_id not in users:
                        users[mes.from_id.user_id] = cl.loop.run_until_complete(self.get_name(mes.from_id)).first_name
                    user = users[mes.from_id.user_id]
                    add = {
                        "time": mes.date.strftime("%Y-%m-%d %H:%M:%S"),
                        "text": mes.message.replace('\n', '. '),
                        "user": user
                    }
                    messages['messages'].append(add)
            try:
                counter = max(ids)
                messages['counter'] = counter
            except:
                messages['counter'] = min_id
            return messages

    async def get_name(self, entity):
        return await self.client.get_entity(entity)

    async def _send_message(self, user: str | int, text: str):
        await self.client.send_message(user, text, parse_mode='Markdown')

    def send_message(self, user: str | int, text: str):
        with self.client.start() as cl:
            cl.loop.run_until_complete(self._send_message(user, text))