|
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,
|
|
)
|
|
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
|
|
|
