Upload 9 files

app.py

@@ -0,0 +1,10 @@
+from dotenv import load_dotenv
+load_dotenv()
+
+import utils.setup_logging
+from utils.interface import create_interface
+
+
+if __name__ == '__main__':
+    interface = create_interface()
+    interface.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio==5.20.1
+Telethon==1.39.0
+python-dotenv

utils/auth.py

@@ -0,0 +1,205 @@
+import asyncio
+import os
+import logging
+from dataclasses import dataclass
+
+from pathlib import Path
+
+from telethon import TelegramClient, errors
+from telethon.sessions import SQLiteSession, MemorySession
+from telethon.sessions.abstract import Session
+
+from utils.validation import Validator
+
+
+@dataclass
+class AuthState:
+    session_type: str = 'sqlite'
+    session_name: str = 'telegram_api_session'
+    memory_session: MemorySession | None = None
+    is_logging: bool = False
+
+    is_auth: bool = False
+    need_send_code: bool = False
+    need_verify_code: bool = False
+    need_verify_2fa: bool = False
+    message: str | None = None
+    client: TelegramClient | None = None
+
+    def __post_init__(self):
+        self.session_dir = Path('sessions')
+        self.session_dir.mkdir(exist_ok=True)
+
+    def check_start_auth_status(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.check_is_auth())
+
+    async def check_is_auth(self) -> None:
+        if Validator.validate_env_vars().is_valid:
+            client = ClientConnector.get_client(self.get_session(), os.getenv('API_ID'), os.getenv('API_HASH'))
+            validation_result = await Validator.validate_auth(client)
+            if validation_result.is_valid:
+                self.set_auth_success()
+
+    def get_session(self) -> Session:
+        if self.session_type == 'sqlite':
+            session_filepath = self.session_dir / self.session_name
+            return SQLiteSession(str(session_filepath))
+        elif self.session_type == 'memory':
+            if self.memory_session is None:
+                self.memory_session = MemorySession()
+            return self.memory_session
+
+    def change_session_type(self, session_type):
+        if session_type != self.session_type:
+            self.session_type = session_type
+
+    def reset_state(self) -> None:
+        defaults = self.__class__()
+        # self.__dict__.update(defaults.__dict__)
+        self.is_auth = defaults.is_auth
+        self.need_send_code = defaults.need_send_code
+        self.need_verify_code = defaults.need_verify_code
+        self.need_verify_2fa = defaults.need_verify_2fa
+        self.message = defaults.message
+        self.client = defaults.client
+
+    def _log(self) -> None:
+        if self.is_logging and self.message:
+            logging.info(self.message)
+
+    def set_auth_failed(self, message: str | None = None) -> None:
+        if message:
+            self.message = message
+        self._log()
+
+    def set_start_auth(self) -> None:
+        self.reset_state()
+        self.message = 'Начата процедура аутентификации'
+
+    def set_client(self, client: TelegramClient) -> None:
+        self.client = client
+        if self.session_type == 'memory':
+            self.memory_session = client.session
+
+    def set_need_send_code(self) -> None:
+        self.need_send_code = True
+        self.message = 'Проверка соединения клиента завершена успешно. Отправка проверочного кода'
+        self._log()
+
+    def set_need_verify_code(self) -> None:
+        self.need_verify_code = True
+        self.message = 'Код отправлен в Telegram. Введите его в поле Проверочный код'
+        self._log()
+
+    def set_need_verify_2fa(self) -> None:
+        self.need_verify_2fa = True
+        self.need_verify_code = False
+        self.message = 'Требуется 2FA-пароль. Введите его в поле Облачный пароль'
+        self._log()
+
+    def set_auth_success(self, message: str | None = None) -> None:
+        self.is_auth = True
+        self.need_send_code = False
+        self.need_verify_code = False
+        self.need_verify_2fa = False
+        self.message = 'Клиент авторизован' if message is None else message
+        self._log()
+
+    async def delete_session(self) -> None:
+        if self.client is not None:
+            await ClientConnector.log_out(self.client)
+        if self.session_type == 'sqlite':
+            session_filepath = self.session_dir / f'{self.session_name}.session'
+            if session_filepath.is_file():
+                session_filepath.unlink(missing_ok=True)
+        elif self.session_type == 'memory':
+            self.memory_session = None
+        self.reset_state()
+        self.message = 'Сессия удалена'
+        self._log()
+
+
+class ClientConnector:
+    @staticmethod
+    def get_client(session: Session, api_id: str, api_hash: str) -> TelegramClient:
+        client = TelegramClient(session, api_id, api_hash, system_version='4.16.30-vxCUSTOM')
+        return client
+    
+    @staticmethod
+    async def connect(client: TelegramClient) -> None:
+        if not client.is_connected():
+            await client.connect()
+
+    @staticmethod
+    async def disconnect(client: TelegramClient) -> None:
+        if client.is_connected():
+            await client.disconnect()
+
+    @classmethod
+    async def log_out(cls, client: TelegramClient) -> None:
+        await cls.connect(client)
+        await client.log_out()
+        await cls.disconnect(client)
+
+    @classmethod
+    async def start_auth(cls, state: AuthState, api_id: str, api_hash: str) -> AuthState:
+        if not api_id or not api_hash:
+            message = 'Не заданы api_id и/или api_hash'
+            state.set_auth_failed(message=message)
+            return state
+        state.set_start_auth()
+        client = cls.get_client(state.get_session(), api_id, api_hash)
+        validation_result = await Validator.validate_auth(client)
+        if validation_result.is_valid:
+            message = 'Клиент авторизован'
+            state.set_auth_success(message)
+        elif not validation_result.is_valid and validation_result.is_error:
+            state.set_auth_failed(message=validation_result.message)
+        elif not validation_result.is_valid and not validation_result.is_error:
+            state.set_client(client)
+            state.set_need_send_code()
+        return state
+
+    @classmethod
+    async def send_code(cls, state: AuthState, phone_number: str) -> AuthState:
+        if not state.need_send_code:
+            return state
+        try:
+            await cls.connect(state.client)
+            await state.client.send_code_request(phone_number)
+            state.set_need_verify_code()
+        except Exception as ex:
+            message = f'Ошибка при отправке кода подтверждения, код ошибки: {ex}'
+            state.set_auth_failed(message)
+        return state
+
+    @classmethod
+    async def verify_code(cls, state: AuthState, phone_number: str, code: str) -> AuthState:
+        if not state.need_verify_code:
+            return state
+        try:
+            await state.client.sign_in(phone=phone_number, code=code)
+            await cls.disconnect(state.client)
+            state.set_auth_success()
+        except errors.SessionPasswordNeededError:
+            state.set_need_verify_2fa()
+        except Exception as ex:
+            message = f'Ошибка при верификации кода подтверждения, код ошибки: {ex}'
+            state.set_auth_failed(message)
+        return state
+
+    @classmethod
+    async def verify_2fa(cls, state: AuthState, password_2fa: str) -> AuthState:
+        if not state.need_verify_2fa:
+            return state
+        try:
+            await state.client.sign_in(password=password_2fa)
+            state.set_auth_success()
+            await cls.disconnect(state.client)
+        except Exception as ex:
+            message = f'Ошибка при верификации облачного пароля, код ошибки: {ex}'
+            state.set_auth_failed(message)
+        return state
+

utils/components.py

@@ -0,0 +1,270 @@
+import os
+from pathlib import Path
+from typing import Callable, Collection
+
+import gradio as gr
+from telethon.sessions import SQLiteSession, MemorySession
+
+from utils.auth import AuthState
+from utils.parser import Parser
+from utils.validation import Validator
+
+
+class Components:
+
+    welcome_message_markdown = '''
+<h5 style='text-align: center'>
+    Парсер сообщений Telegram
+</h5>
+<h6 style='text-align: center'>
+    <a 
+        href="https://github.com/sergey21000/telegram-message-parser" 
+        target='_blank'>GitHub
+    </a>
+    проекта с инструкциями по получению `API_ID` и `API_HASH` приложения Telegram
+</h6>
+    '''
+
+    @staticmethod
+    def _create_env_var_textbox(
+            env_var: str | None = None,
+            validator_method: Callable | None = None,
+            **kwargs,
+            ) -> gr.Textbox:
+        value = None
+        if env_var and validator_method:
+            validation_result = validator_method()
+            if validation_result.is_valid:
+                value = os.getenv(env_var)
+        curr_kwargs = dict(value=value)
+        curr_kwargs.update(kwargs)
+        return gr.Textbox(**curr_kwargs)
+
+    @classmethod
+    def api_id(cls) -> gr.Textbox:
+        return cls._create_env_var_textbox(
+            env_var='API_ID',
+            validator_method=Validator.validate_env_id,
+            label='API_ID Telegram app',
+            placeholder='API_ID приложения Telegram',
+            scale=1,
+            )
+
+    @classmethod
+    def api_hash(cls) -> gr.Textbox:
+        return cls._create_env_var_textbox(
+            env_var='API_HASH',
+            validator_method=Validator.validate_env_hash,
+            label='API_HASH Telegram app',
+            placeholder='API_HASH приложения Telegram',
+            scale=1,
+            )
+
+    @classmethod
+    def phone_number(cls) -> gr.Textbox:
+        return cls._create_env_var_textbox(
+            env_var='PHONE_NUMBER',
+            validator_method=Validator.validate_env_phone_number,
+            label='PHONE_NUMBER Telegram app',
+            placeholder='PHONE_NUMBER приложения Telegram',
+            scale=1,
+            )
+
+    @staticmethod
+    def code(visible: str = False, render: bool = True) -> gr.Textbox:
+        component = gr.Textbox(
+            value=None,
+            label='Проверочный код',
+            visible=visible,
+            render=render,
+            placeholder=None,
+            scale=1,
+            )
+        return component
+
+    @staticmethod
+    def password_2fa(visible: str = False, render: bool = True) -> gr.Textbox:
+        component = gr.Textbox(
+            type='password',
+            value=None,
+            label='Облачный пароль',
+            visible=visible,
+            render=render,
+            placeholder=None,
+            scale=1,
+            )
+        return component
+
+    @staticmethod
+    def auth_status(value: str | None = None) -> gr.Textbox:
+        component = gr.Textbox(
+            value=value,
+            label='Статус авторизации',
+            placeholder=None,
+            interactive=False,
+            scale=1,
+            )
+        return component
+
+    @staticmethod
+    def auth_btn(interactive: bool = True) -> gr.Button:
+        component = gr.Button(
+            value='Авторизация',
+            interactive=interactive,
+            scale=1,
+            )
+        return component
+
+    @staticmethod
+    def code_btn(visible: bool = False, render: bool = True) -> gr.Button:
+        component = gr.Button(
+            value='Подтвержение кода',
+            visible=visible,
+            render=render,
+            scale=0,
+            )
+        return component
+
+    @staticmethod
+    def password_2fa_btn(visible: bool = False, render: bool = True) -> gr.Button:
+        component = gr.Button(
+            value='Подтверждение облачного пароля',
+            visible=visible,
+            render=render,
+            scale=0,
+            )
+        return component
+
+    @staticmethod
+    def delete_session_btn(visible: bool = False, render: bool = True) -> gr.Button:
+        component = gr.Button(
+            value='Удаление сессии',
+            visible=visible,
+            render=render,
+            scale=1,
+            )
+        return component
+
+    @classmethod
+    def session_type_radio(cls) -> gr.Radio:
+        session_types = {'sqlite': SQLiteSession, 'memory': MemorySession}
+        component = gr.Radio(
+            choices=session_types,
+            value='sqlite',
+            label='Тип сессии',
+            info=None,  # cls.session_type_markdown
+            )
+        return component
+
+    @staticmethod
+    def chats_usernames() -> gr.Textbox:
+        component = gr.Textbox(
+            label='Адреса чатов',
+            placeholder='Ссылки или ID чатов/каналов через пробел или перенос строки',
+            scale=1,
+            lines=2,
+            )
+        return component
+
+    @staticmethod
+    def add_chat_btn() -> gr.Button:
+        component = gr.Button(
+            value='Добавить чат/чаты',
+            scale=0,
+            )
+        return component
+
+    @staticmethod
+    def chats_list_status() -> gr.Textbox:
+        component = gr.Textbox(
+            label='Добавленные чаты',
+            placeholder='Здесь будет список чатов для парсинга',
+            scale=1,
+            lines=3,
+            )
+        return component
+
+    @staticmethod
+    def parse_status() -> gr.Textbox:
+        component = gr.Textbox(
+            label='Статус парсинга',
+            placeholder='Здесь будет отчет о результатах парсинга',
+            scale=1,
+            lines=8,
+            )
+        return component
+
+    @staticmethod
+    def start_parse_btn() -> gr.Button:
+        component = gr.Button(
+            value='Начать парсинг',
+            scale=0,
+            )
+        return component
+
+    @staticmethod
+    def download_btn(value: str | None = None) -> gr.Button:
+        component = gr.DownloadButton(
+            label='Загрузить csv результаты',
+            value=value,
+            visible=value is not None,
+            scale=0,
+            )
+        return component
+
+    @staticmethod
+    def get_parse_args() -> list[gr.component]:
+        limit = gr.Number(
+            value=None,
+            label='limit',
+            info='Сколько сообщений парсить',
+            )
+        offset_date = gr.DateTime(
+            value=None,
+            label='offset_date',
+            info='До какой даты парсить',
+            timezone='Europe/Moscow',
+            )
+        reverse = gr.Checkbox(
+            value=False,
+            label='reverse',
+            info='Парсить от сегодняшнего сообщения к самому раннему',
+        )
+        parse_args = [limit, offset_date, reverse]
+        return parse_args
+
+
+class ComponentsFn(Components):
+    @staticmethod
+    def update_status(auth_state: AuthState) -> str | None:
+        return auth_state.message
+
+    @classmethod
+    def get_dynamic_visible_components(cls, auth_state: AuthState, render: bool = True) -> tuple[gr.component]:
+        code = cls.code(visible=auth_state.need_verify_code, render=render)
+        code_btn = cls.code_btn(visible=auth_state.need_verify_code, render=render)
+
+        password_2fa = cls.password_2fa(visible=auth_state.need_verify_2fa, render=render)
+        password_2fa_btn = cls.password_2fa_btn(visible=auth_state.need_verify_2fa, render=render)
+
+        delete_session_btn = cls.delete_session_btn(visible=auth_state.is_auth, render=render)
+        return code, code_btn, password_2fa, password_2fa_btn, delete_session_btn
+
+    @staticmethod
+    def update_auth_state_session_type(auth_state: AuthState, session_type: str) -> None:
+        auth_state.change_session_type(session_type)
+
+    @staticmethod
+    async def delete_session(auth_state: AuthState) -> None:
+        await auth_state.delete_session()
+
+    @classmethod
+    def update_download_btn(cls, csv_paths: Collection[Path]) -> gr.Button | None:
+        if len(csv_paths) == 0:
+            return None
+        elif len(csv_paths) == 1:
+            filepath = csv_paths[0]
+        else:
+            filepath = Parser.zip_files(csv_paths)
+        component = cls.download_btn(value=filepath)
+        return component

utils/interface.py

@@ -0,0 +1,152 @@
+import gradio as gr
+
+from utils.auth import AuthState, ClientConnector
+from utils.components import Components, ComponentsFn
+from utils.parser import Parser
+
+
+def create_interface() -> gr.Blocks:
+    auth_state = AuthState()
+    auth_state.check_start_auth_status()
+
+    # css = '.gradio-container {width: 60% !important}'
+    css = '''
+    .gradio-container {
+        width: 70% !important;
+        margin: 0 auto !important; /* Центрирование по горизонтали */
+    }
+    '''
+
+    with gr.Blocks(css=css) as interface:
+        gr.Markdown(Components.welcome_message_markdown)
+        
+        auth_state = gr.State(auth_state)
+        chats_list = gr.State([])
+        csv_names = gr.State([])
+
+        dynamic_visible_components = ComponentsFn.get_dynamic_visible_components(auth_state.value, render=False)
+        code, code_btn, password_2fa, password_2fa_btn, delete_session_btn = dynamic_visible_components
+
+        with gr.Group():
+            gr.Markdown('Авторизация')
+            with gr.Row():
+                with gr.Column():
+                    session_type = Components.session_type_radio()
+                    auth_status = Components.auth_status(value=auth_state.value.message)
+                    with gr.Row():
+                        auth_btn = Components.auth_btn()
+                        delete_session_btn.render()
+                    code.render()
+                    code_btn.render()
+                    password_2fa.render()
+                    password_2fa_btn.render()
+
+                with gr.Column():
+                    api_id = Components.api_id()
+                    api_hash = Components.api_hash()
+                    phone_number = Components.phone_number()
+
+        auth_btn.click(
+            fn=ClientConnector.start_auth,
+            inputs=[auth_state, api_id, api_hash],
+            outputs=[auth_state],
+        ).then(
+            fn=ClientConnector.send_code,
+            inputs=[auth_state, phone_number],
+            outputs=[auth_state],
+        ).then(
+            fn=ComponentsFn.get_dynamic_visible_components,
+            inputs=[auth_state],
+            outputs=dynamic_visible_components,
+        ).then(
+            fn=ComponentsFn.update_status,
+            inputs=[auth_state],
+            outputs=[auth_status],
+        )
+
+        code_btn.click(
+            fn=ClientConnector.verify_code,
+            inputs=[auth_state, phone_number, code],
+            outputs=[auth_state],
+        ).then(
+            fn=ComponentsFn.get_dynamic_visible_components,
+            inputs=[auth_state],
+            outputs=dynamic_visible_components,
+        ).then(
+            fn=ComponentsFn.update_status,
+            inputs=[auth_state],
+            outputs=[auth_status],
+        )
+
+        password_2fa_btn.click(
+            fn=ClientConnector.verify_2fa,
+            inputs=[auth_state, password_2fa],
+            outputs=[auth_state],
+        ).then(
+            fn=ComponentsFn.get_dynamic_visible_components,
+            inputs=[auth_state],
+            outputs=dynamic_visible_components,
+        ).then(
+            fn=ComponentsFn.update_status,
+            inputs=[auth_state],
+            outputs=[auth_status],
+        )
+
+        delete_session_btn.click(
+            fn=ComponentsFn.delete_session,
+            inputs=[auth_state],
+            outputs=None,
+        ).then(
+            fn=ComponentsFn.get_dynamic_visible_components,
+            inputs=[auth_state],
+            outputs=dynamic_visible_components,
+        ).then(
+            fn=ComponentsFn.update_status,
+            inputs=[auth_state],
+            outputs=[auth_status],
+        )
+
+        session_type.change(
+            fn=ComponentsFn.update_auth_state_session_type,
+            inputs=[auth_state, session_type],
+            outputs=None,
+        )
+
+
+        with gr.Group():
+            gr.Markdown('Парсинг')
+            with gr.Row():
+                with gr.Column():
+                    with gr.Group():
+                        gr.Markdown('Чаты для парсинга')
+                        chats_usernames = Components.chats_usernames()
+                        add_chat_btn = Components.add_chat_btn()
+                        chats_list_status = Components.chats_list_status()
+                with gr.Column():
+                    with gr.Group():
+                        gr.Markdown('Параметры парсинга')
+                        parse_args = Components.get_parse_args()
+                with gr.Column():
+                    with gr.Group():
+                        gr.Markdown('Результаты парсинга')
+                        start_parse_btn = Components.start_parse_btn()
+                        parse_status = Components.parse_status()
+                        download_btn = Components.download_btn()
+
+        add_chat_btn.click(
+            fn=Parser.add_chat_to_chats_list,
+            inputs=[auth_state, chats_usernames, chats_list, api_id, api_hash],
+            outputs=[chats_list_status],
+        )
+
+        start_parse_btn.click(
+            fn=Parser.parse_chats,
+            inputs=[auth_state, chats_list, api_id, api_hash, *parse_args],
+            outputs=[parse_status, csv_names],
+        ).then(
+            fn=ComponentsFn.update_download_btn,
+            inputs=[csv_names],
+            outputs=[download_btn],
+        )
+
+    return interface

utils/parser.py

@@ -0,0 +1,240 @@
+import asyncio
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Collection
+
+import pandas as pd
+import gradio as gr
+from telethon import TelegramClient, types, errors
+
+from utils.auth import AuthState, ClientConnector
+from utils.validation import Validator
+
+
+MESSAGE_DICT = dict[str, str | int | datetime | None]
+DEFAULT_PARSE_KWARGS = dict(
+    limit=None,
+    offset_date=None,
+    reverse=False,
+)
+
+
+@dataclass
+class Chat:
+    chat: types.TLObject
+    chat_name: str | None
+    chat_username: str
+    chat_type: str
+    chat_id: int
+
+    @classmethod
+    def from_telethon_chat(cls, chat: types.TLObject, chat_username: str):
+        chat_id = chat.id
+        if isinstance(chat, types.User):
+            chat_type = 'Chat'
+            chat_name = f'{chat.first_name} {chat.last_name}'
+        else:
+            chat_type = 'Channel/Group'
+            chat_name = chat.title
+        return cls(chat, chat_name, chat_username, chat_type, chat_id)
+
+    def get_chat_info(self) -> str:
+        chat_info = f'Chat name: {self.chat_name}, Chat type: {self.chat_type}, Chat ID: {self.chat_id}'
+        return chat_info
+
+
+class Parser:
+    parse_results_dir = Path('parse_results_dir')
+
+    @staticmethod
+    def message_to_dict(message: types.Message) -> MESSAGE_DICT:
+        text = message.text if message.text else message.message
+        if not text:
+            return None
+
+        date = message.date
+        sender = message.sender
+        sender_type = type(sender).__name__
+        chat = message._chat
+        chat_id = chat.id
+        chat_type = type(chat).__name__
+        chat_name = chat.title if isinstance(chat, types.Channel) else f'{chat.first_name} {chat.last_name}'
+
+        if isinstance(sender, types.User):
+            sender_id = message.sender.id
+            username = sender.username
+            first_name = sender.first_name
+            last_name = sender.last_name
+        else:
+            sender_id = message._sender_id
+            username = getattr(message.sender, 'username', None)
+            first_name = None
+            last_name = None
+
+        message_dict = {
+            'date': date,
+            'chat_type': chat_type,
+            'chat_name': chat_name,
+            'chat_id': chat_id,
+            'sender_type': sender_type,
+            'sender_username': username,
+            'sender_first_name': first_name,
+            'sender_last_name': last_name,
+            'sender_id': sender_id,
+            'text': text,
+        }
+        return message_dict
+
+    @classmethod
+    async def get_messages_from_chat(
+        cls,
+        client: TelegramClient,
+        chat: types.TLObject,
+        parse_chats_pb_info: str,
+        **parse_kwargs,
+        ) -> list[MESSAGE_DICT]:
+
+        async with client:
+            progress = gr.Progress()
+            messages = client.iter_messages(entity=chat, **parse_kwargs)
+            message_dicts = []
+            message_count = 0
+            async for message in messages:
+                message_count += 1
+                if message_count % 1000 == 0:
+                    await asyncio.sleep(1)
+                message_dict = cls.message_to_dict(message)
+                if message_dict is not None:
+                    message_dicts.append(message_dict)
+
+                if message_count % 1000 == 0:
+                    await asyncio.sleep(1)
+
+                if parse_kwargs['limit'] is not None:
+                    total = parse_kwargs['limit']
+                    progress(message_count / total, desc=f'{parse_chats_pb_info}, Parsing messages {message_count}/{total}')
+                else:
+                    progress(message_count, desc=f'{parse_chats_pb_info}, Parsing messages {message_count}/?')
+
+        if not parse_kwargs['reverse']:
+            message_dicts = message_dicts[::-1]
+        return message_dicts
+
+    @classmethod
+    async def parse_chats(
+        cls, 
+        auth_state: AuthState,
+        chats_list: list[Chat],
+        api_id: str,
+        api_hash: str,
+        *parse_args,
+        ) -> tuple[str, list[Path]]:
+
+        cvs_paths = []
+        parse_result = ''
+
+        if len(chats_list) == 0:
+            return 'Список чатов для парсинга пустой', cvs_paths
+
+        client = ClientConnector.get_client(auth_state.get_session(), api_id, api_hash)
+        validation_result = await Validator.validate_auth(client)
+        if not validation_result.is_valid:
+            return 'Клиент не авторизован', cvs_paths
+
+        parse_kwargs = dict(zip(DEFAULT_PARSE_KWARGS.keys(), parse_args))
+        progress = gr.Progress()
+
+        for i, chat in enumerate(chats_list, start=1):
+            try:
+                parse_chats_pb_info = f'Parsing chats {i}/{len(chats_list)}'
+                message_dicts = await cls.get_messages_from_chat(client, chat.chat, parse_chats_pb_info, **parse_kwargs)
+                if len(message_dicts) == 0:
+                    log_msg = f'Из чата {chat.chat_username} не было извлечено ни одного сообщения'
+                    parse_result += log_msg + '\n'
+                else:
+                    cvs_path = cls.messages_to_csv(message_dicts)
+                    cvs_paths.append(cvs_path)
+                    log_msg = f'Успешный парсинг чата {chat.chat_username}, кол-во сообщений: {len(message_dicts)}'
+                    parse_result += log_msg + '\n'
+            except Exception as ex:
+                log_msg = f'Ошибка при парсинге чата {chat.chat_username}, код ошибки: {ex}'
+                parse_result += log_msg + '\n'
+
+            progress(i / len(chats_list), desc=parse_chats_pb_info)
+        return parse_result, cvs_paths
+
+    @classmethod
+    def messages_to_csv(cls, message_dicts: Collection[MESSAGE_DICT]) -> Path:
+        df = pd.DataFrame.from_dict(message_dicts)
+        chat_name = message_dicts[0].get('chat_name', '')
+        cvs_path = cls.parse_results_dir / f'telegram_history_{chat_name}.csv'
+        df.to_csv(cvs_path, index=False)
+        return cvs_path
+
+    @classmethod
+    def zip_files(cls, file_paths: Collection[Path]) -> Path:
+        zip_filepath = cls.parse_results_dir / 'parse_results_csv.zip'
+        with zipfile.ZipFile(zip_filepath, 'w') as zipf:
+            for file_path in file_paths:
+                zipf.write(file_path, arcname=file_path)
+        return zip_filepath
+
+    @staticmethod
+    def get_chats_info(chats_list: list[Chat]) -> str:
+        chats_info = ''
+        for i, chat in enumerate(chats_list, start=1):
+            chats_info += f'{i}: ' + chat.get_chat_info() + '\n'
+        return chats_info
+
+    @staticmethod
+    async def get_chat(client: TelegramClient, chat_username: str) -> types.TLObject:
+        try:
+            if client.is_connected():
+                chat = await client.get_entity(chat_username)
+            else:
+                async with client:
+                    chat = await client.get_entity(chat_username)
+        except (errors.UsernameNotOccupiedError, errors.UsernameInvalidError) as ex:
+            log_msg = f'Чат или канал {chat_username} не найден или введен неверно'
+            raise errors.UsernameInvalidError(log_msg)
+        except Exception as ex:
+            log_msg = f'Ошибка при получении объекта чата, код ошибки: {ex}'
+            raise Exception(log_msg)
+        return chat
+
+    @classmethod
+    async def add_chat_to_chats_list(
+        cls, 
+        auth_state: AuthState,
+        chats_usernames,
+        chats_list: list[Chat],
+        api_id: str,
+        api_hash: str,
+        ) -> str:
+
+        if chats_usernames.strip() == '':
+            return 'Не заданы адрес/адреса чатов для добавления'
+
+        client = ClientConnector.get_client(auth_state.get_session(), api_id, api_hash)
+        validation_result = await Validator.validate_auth(client)
+        if not validation_result.is_valid:
+            return 'Клиент не авторизован'
+
+        for chat_username in chats_usernames.split():
+            try:
+                telethon_chat = await cls.get_chat(client, chat_username.strip())
+                if not telethon_chat in chats_list:
+                    chat = Chat.from_telethon_chat(telethon_chat, chat_username)
+                    chats_list.append(chat)
+                else:
+                    log_msg = f'Чат {chat_username} уже есть в списке'
+                    gr.Info(log_msg)
+            except Exception as ex:
+                log_msg = str(ex)
+                gr.Info(log_msg)
+        return cls.get_chats_info(chats_list)
+
+
+Parser.parse_results_dir.mkdir(exist_ok=True)

utils/setup_logging.py

@@ -0,0 +1,31 @@
+import sys
+import logging
+import pytz
+from pytz.tzinfo import BaseTzInfo
+from datetime import datetime
+
+
+
+def setup_logging(log_to_file: bool, level: int, timezone: BaseTzInfo) -> None:
+    '''Настройка логгирования под конкретный часовой пояс'''
+    logging.Formatter.converter = lambda *args: datetime.now(tz=timezone).timetuple()
+    handlers = [logging.StreamHandler(sys.stdout)]
+    
+    if log_to_file:
+        log_file_name = 'bot_log.log'
+        handlers.append(logging.FileHandler(log_file_name))
+    
+    format = '%(asctime)s - %(filename)s:%(lineno)d - %(levelname)s - %(funcName)s: %(message)s'
+    logging.basicConfig(
+        level=level,
+        format=format,
+        datefmt='%Y-%m-%d %H:%M:%S',
+        handlers=handlers,
+        force=True,
+    )
+
+LOG_TO_FILE = False
+LEVEL = logging.INFO
+TIMEZONE: BaseTzInfo = pytz.timezone('Europe/Moscow')
+
+setup_logging(log_to_file=LOG_TO_FILE, level=LEVEL, timezone=TIMEZONE)

utils/validation.py

@@ -0,0 +1,64 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from telethon import TelegramClient
+
+
+@dataclass
+class ValidationResult:
+    is_valid: bool
+    is_error: bool = False
+    message: str = ''
+
+
+class Validator:
+    env_vars = ['API_ID', 'API_HASH', 'PHONE_NUMBER']
+
+    @staticmethod
+    def validate_env_file(env_filename: str = '.env') -> ValidationResult:
+        if not Path(env_filename).is_file():
+            log_msg = 'Отсутвует файл .env'
+            return ValidationResult(is_valid=False, message=log_msg)
+        return ValidationResult(is_valid=True)
+
+    @staticmethod
+    def validate_env_var(var_name: str, error_msg: str | None = None) -> ValidationResult:
+        if os.getenv(var_name) is None:
+            return ValidationResult(is_valid=False, message=error_msg)
+        return ValidationResult(is_valid=True)
+
+    @classmethod
+    def validate_env_id(cls) -> ValidationResult:
+        return cls.validate_env_var('API_ID', 'Отсутствует переменная API_ID')
+
+    @classmethod
+    def validate_env_hash(cls) -> ValidationResult:
+        return cls.validate_env_var('API_HASH', 'Отсутствует переменная API_HASH')
+
+    @classmethod
+    def validate_env_phone_number(cls) -> ValidationResult:
+        return cls.validate_env_var('PHONE_NUMBER', 'Отсутствует переменная PHONE_NUMBER')
+
+    @classmethod
+    def validate_env_vars(cls) -> ValidationResult:
+        if all([cls.validate_env_var(var).is_valid for var in cls.env_vars]):
+            return ValidationResult(is_valid=True)
+        return ValidationResult(is_valid=False)
+
+    @staticmethod
+    async def validate_auth(client: TelegramClient) -> ValidationResult:
+        try:
+            if not client.is_connected():
+                await client.connect()
+            is_user_authorized = await client.is_user_authorized()
+            if not is_user_authorized:
+                log_msg = 'Клиент не авторизован'
+                return ValidationResult(is_valid=False, message=log_msg)
+            return ValidationResult(is_valid=True)
+        except Exception as ex:
+            log_msg = f'Ошибка при подключении клиента, код ошибки: {ex}'
+            return ValidationResult(is_valid=False, is_error=True, message=log_msg)
+        finally:
+            if client.is_connected():
+                await client.disconnect()