import os import logging from shutil import move from pathlib import Path from countryflag import getflag from modules.checker import Result, Checker, SessionInvalid logger = logging.getLogger(__name__) class Worker: def __init__( self, input_path: str = "tdata", output_path: str = "data", temp_path: str = "temp", lang_code: str = "en", ): logger.debug( f"Инициализация Worker с путями: input={input_path}, output={output_path}, temp={temp_path}, lang={lang_code}" ) self.input_path = Path(input_path).absolute() self.output_path = Path(output_path).absolute() self.temp_path = Path(temp_path).absolute() self.lang_code = lang_code self.valid_path = self.output_path.joinpath("valid") self.invalid_path = self.output_path.joinpath("invalid") self.all_path = self.valid_path.joinpath("all.txt") logger.debug( f"Пути настроены: valid={self.valid_path}, invalid={self.invalid_path}" ) async def __aenter__(self): logger.debug("Вход в контекстный менеджер Worker") logger.debug("Создание необходимых директорий") if not self.input_path.exists(): self.input_path.mkdir() logger.debug(f"Создана директория: {self.input_path}") if not self.output_path.exists(): self.output_path.mkdir() logger.debug(f"Создана директория: {self.output_path}") if not self.temp_path.exists(): self.temp_path.mkdir() logger.debug(f"Создана директория: {self.temp_path}") else: logger.debug("Очистка существующей временной директории") self.__cleanup() if not self.valid_path.exists(): self.valid_path.mkdir() logger.debug(f"Создана директория: {self.valid_path}") if not self.invalid_path.exists(): self.invalid_path.mkdir() logger.debug(f"Создана директория: {self.invalid_path}") logger.debug("Все директории готовы") return self async def __aexit__(self, exc_type, exc_val, exc_tb): logger.debug("Выход из контекстного менеджера Worker") self.__cleanup() def __cleanup(self): logger.debug("Начало очистки временных файлов") if self.temp_path.exists(): deleted_count = 0 for file in os.listdir(path=self.temp_path): file_path = self.temp_path.joinpath(str(file)) try: if file_path.is_file() and ".gitinclude" not in str(file_path): file_path.unlink(missing_ok=True) deleted_count += 1 elif file_path.is_dir(): file_path.rmdir() deleted_count += 1 except Exception as e: logger.debug(f"Не удалось удалить {file_path}: {e}") if deleted_count > 0: logger.debug(f"Удалено временных файлов: {deleted_count}") else: logger.debug("Нет временных файлов для удаления") def __list_tdata(self): logger.debug(f"Поиск TDATA папок в {self.input_path}") tdatas = [] if self.input_path.exists(): for content in os.listdir(path=self.input_path): joined = self.input_path.joinpath(content) if joined.is_dir(): tdatas.append(joined.absolute()) logger.debug(f"Найдена TDATA папка: {joined}") logger.debug(f"Всего найдено TDATA папок: {len(tdatas)}") return tdatas def __template(self, result: Result): logger.debug(f"Создание шаблона для пользователя {result.me.id}") try: flag = getflag(countries=result.country) except Exception as e: logger.debug(f"Не удалось получить флаг для страны '{result.country}': {e}") flag = "🌐" spamblock_status = "unknown" if result.spamblock is not None: spamblock_status = "true" if result.spamblock else "false" return ( f"👤 [{flag} {result.country}] +{result.me.phone} — {result.me.id}\n" f"⭐️ {result.stars}\n" f"💎 Premium: {'true' if result.me.premium else 'false'}\n" f"🚫 Spamblock: {spamblock_status}\n" f"👥 {len(result.chats)}/{len(result.contacts)}\n" f"🍀 {len(result.admin_channels)}" ) def __append(self, content: str, path: Path): logger.debug(f"Добавление данных в файл: {path}") with open(file=path, mode="a", encoding="utf-8") as file: file.write(content + "\n\n") def __write(self, content: str, path: Path): logger.debug(f"Запись данных в файл: {path}") with open(file=path, mode="w", encoding="utf-8") as file: file.write(content) def __save_valid(self, user_id: int, result: Result, tdata: Path): logger.debug(f"Сохранение валидной TDATA для пользователя {user_id}") id_path = self.valid_path.joinpath(str(user_id)) if not id_path.absolute().exists(): id_path.mkdir() logger.debug(f"Создана директория для пользователя: {id_path}") content = self.__template(result=result) id_txt_path = id_path.joinpath(f"{user_id}.txt") logger.debug("Добавление в общий файл all.txt") self.__append(content=content, path=self.all_path) logger.debug("Запись в индивидуальный файл") self.__write(content=content, path=id_txt_path) tdata_old_path = tdata tdata_relative = tdata.relative_to(self.input_path) tdata_new_path = id_path.joinpath(tdata_relative.name) logger.debug(f"Перемещение TDATA: {tdata_old_path} -> {tdata_new_path}") move(src=str(tdata_old_path), dst=str(tdata_new_path)) logger.info(f"Валидная TDATA сохранена для пользователя {user_id}") def __save_invalid(self, tdata: Path): logger.debug(f"Сохранение невалидной TDATA: {tdata}") tdata_old_path = tdata tdata_relative = tdata.relative_to(self.input_path) tdata_new_path = self.invalid_path.joinpath(tdata_relative.name) logger.debug(f"Перемещение TDATA: {tdata_old_path} -> {tdata_new_path}") move(src=str(tdata_old_path), dst=str(tdata_new_path)) logger.info(f"Невалидная TDATA перемещена: {tdata_relative.name}") async def check_one(self, tdata: Path): tdata_absolute = tdata.absolute() logger.info(f"Проверка TDATA: {tdata.name}") try: async with Checker( path=tdata_absolute, temp_path=self.temp_path, lang_code=self.lang_code ) as checker: result = await checker.check_all() logger.info( f"Проверка завершена: {result.me.first_name} (+{result.me.phone})" ) self.__save_valid( user_id=result.me.id, result=result, tdata=tdata_absolute ) except SessionInvalid: logger.warning(f"TDATA невалидна: {tdata.name}") self.__save_invalid(tdata=tdata_absolute) except Exception as e: logger.error(f"Ошибка при проверке TDATA {tdata.name}: {e}") self.__save_invalid(tdata=tdata_absolute) async def check_all(self): logger.info("Начало массовой проверки TDATA") tdatas = self.__list_tdata() if not tdatas: logger.warning("TDATA папки не найдены") return logger.info(f"Найдено TDATA для проверки: {len(tdatas)}") for index, tdata in enumerate(tdatas, 1): logger.info(f"[{index}/{len(tdatas)}] Проверка: {tdata.name}") await self.check_one(tdata=tdata) logger.info("Массовая проверка завершена")