File size: 8,664 Bytes
c7c4ce5 2a039c9 c7c4ce5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 |
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='memory',
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
|