1
0
Fork 0
mirror of https://github.com/EDeev/converterbot.git synced 2026-10-07 20:49:33 +03:00

Бот: защита от zip-бомбы, обработка в отдельном потоке, токен из окружения

- архив проверялся только по размеру (20 МБ) — распакованный мог занять гигабайты;
  теперь до распаковки проверяются число файлов (до 5000) и объём (до 200 МБ);
- конвертация шла прямо в обработчике и блокировала бота для всех;
- формат проверяется до скачивания, документ без имени больше не роняет обработчик;
- токен — из BOT_TOKEN;
- rep_to_txt: вместо изображений выводилась ссылка-заготовка raw.githubusercontent.com/...,
  которая никуда не вела; запуск из командной строки с путём вместо жёстко прописанного
  пути Windows.
This commit is contained in:
Деев Егор Викторович 2026-10-05 23:24:29 +00:00
parent 7b8e3ce61a
commit 911b6ce233
2 changed files with 64 additions and 46 deletions

58
bot.py
View file

@ -1,7 +1,6 @@
import asyncio import asyncio
import logging import logging
import os import os
import shutil
import zipfile import zipfile
from pathlib import Path from pathlib import Path
from tempfile import TemporaryDirectory from tempfile import TemporaryDirectory
@ -14,11 +13,15 @@ from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.client.bot import DefaultBotProperties from aiogram.client.bot import DefaultBotProperties
# Импорт наших конвертеров # Импорт наших конвертеров
from md_to_docx import MarkdownToDocxConverter, DocumentSettings from md2gost import MarkdownToDocxConverter, DocumentSettings
from rep_to_txt import generate_complete_project_structure from rep_to_txt import generate_complete_project_structure
# Конфигурация # Конфигурация
BOT_TOKEN = "**************************" # @my_convbot BOT_TOKEN = os.getenv("BOT_TOKEN", "XXXXXXXXXXXXXXXXXXXXXXXX") # @my_convbot
MAX_FILE_SIZE = 20 * 1024 * 1024 # больше Telegram-боту не скачать
MAX_UNPACKED_SIZE = 200 * 1024 * 1024 # защита от zip-бомбы
MAX_FILES_IN_ARCHIVE = 5000
# Инициализация бота # Инициализация бота
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
@ -60,14 +63,18 @@ async def help_handler(msg: Message) -> None:
async def handle_document(msg: Message) -> None: async def handle_document(msg: Message) -> None:
"""Обработка загруженных документов""" """Обработка загруженных документов"""
document = msg.document document = msg.document
file_name = document.file_name # у документа может не быть имени; путь берём только из имени файла, без каталогов
file_size = document.file_size file_name = Path(document.file_name or "file").name
file_size = document.file_size or 0
if file_size > 20 * 1024 * 1024: if file_size > MAX_FILE_SIZE:
await msg.answer("❌ Файл слишком большой! Максимум 20 МБ") await msg.answer("❌ Файл слишком большой! Максимум 20 МБ")
return return
file_ext = Path(file_name).suffix.lower() file_ext = Path(file_name).suffix.lower()
if file_ext not in (".md", ".zip"):
await msg.answer("❌ Неподдерживаемый формат файла! Пришлите .md или .zip")
return
status_msg = await msg.answer("⏳ Обрабатываю файл...") status_msg = await msg.answer("⏳ Обрабатываю файл...")
@ -78,19 +85,15 @@ async def handle_document(msg: Message) -> None:
input_path = os.path.join(temp_dir, file_name) input_path = os.path.join(temp_dir, file_name)
await bot.download_file(file_info.file_path, input_path) await bot.download_file(file_info.file_path, input_path)
# конвертация — в отдельном потоке, чтобы бот не замирал для остальных
if file_ext == '.md': if file_ext == '.md':
# Конвертация MD → DOCX # Конвертация MD → DOCX
output_path = await convert_md_to_docx(input_path, temp_dir) output_path = await asyncio.to_thread(convert_md_to_docx, input_path, temp_dir)
output_name = Path(file_name).stem + '.docx' output_name = Path(file_name).stem + '.docx'
elif file_ext in ['.zip']:
# Анализ архива → TXT
output_path = await analyze_archive(input_path, temp_dir, file_ext)
output_name = Path(file_name).stem + '_structure.txt'
else: else:
await status_msg.edit_text("❌ Неподдерживаемый формат файла!") # Анализ архива → TXT
return output_path = await asyncio.to_thread(analyze_archive, input_path, temp_dir)
output_name = Path(file_name).stem + '_structure.txt'
# Отправка результата # Отправка результата
with open(output_path, 'rb') as output_file: with open(output_path, 'rb') as output_file:
@ -102,11 +105,20 @@ async def handle_document(msg: Message) -> None:
await status_msg.edit_text("✅ Конвертация завершена!") await status_msg.edit_text("✅ Конвертация завершена!")
except ArchiveTooLarge as e:
await status_msg.edit_text(f"❌ {e}")
except zipfile.BadZipFile:
await status_msg.edit_text("❌ Архив повреждён или это не .zip")
except Exception as e: except Exception as e:
logger.error(f"Ошибка обработки файла: {e}") logger.exception("Ошибка обработки файла")
await status_msg.edit_text(f"❌ Ошибка обработки: {str(e)}") await status_msg.edit_text(f"❌ Ошибка обработки: {str(e)}")
async def convert_md_to_docx(md_path: str, temp_dir: str) -> str:
class ArchiveTooLarge(Exception):
pass
def convert_md_to_docx(md_path: str, temp_dir: str) -> str:
"""Конвертация Markdown в DOCX""" """Конвертация Markdown в DOCX"""
output_path = os.path.join(temp_dir, "output.docx") output_path = os.path.join(temp_dir, "output.docx")
@ -123,13 +135,23 @@ async def convert_md_to_docx(md_path: str, temp_dir: str) -> str:
return output_path return output_path
async def analyze_archive(archive_path: str, temp_dir: str, file_ext: str) -> str: def check_archive(zip_ref: zipfile.ZipFile) -> None:
"""Архив на 20 МБ может распаковаться в гигабайты и забить диск — проверяем до распаковки"""
infos = zip_ref.infolist()
if len(infos) > MAX_FILES_IN_ARCHIVE:
raise ArchiveTooLarge(f"В архиве больше {MAX_FILES_IN_ARCHIVE} файлов")
if sum(info.file_size for info in infos) > MAX_UNPACKED_SIZE:
raise ArchiveTooLarge(f"Распакованный архив больше {MAX_UNPACKED_SIZE // 1024 // 1024} МБ")
def analyze_archive(archive_path: str, temp_dir: str) -> str:
"""Анализ архива и создание структуры проекта""" """Анализ архива и создание структуры проекта"""
extract_dir = os.path.join(temp_dir, "extracted") extract_dir = os.path.join(temp_dir, "extracted")
os.makedirs(extract_dir, exist_ok=True) os.makedirs(extract_dir, exist_ok=True)
# Извлечение архива # Извлечение архива
with zipfile.ZipFile(archive_path, 'r') as zip_ref: with zipfile.ZipFile(archive_path, 'r') as zip_ref:
check_archive(zip_ref)
zip_ref.extractall(extract_dir) zip_ref.extractall(extract_dir)
# Поиск основной папки проекта # Поиск основной папки проекта

View file

@ -1,3 +1,4 @@
import argparse
import os import os
IGNORE_PATTERNS = { IGNORE_PATTERNS = {
@ -100,13 +101,11 @@ def process_single_file(relative_path, file_path):
file_ext = os.path.splitext(relative_path)[1].lower() file_ext = os.path.splitext(relative_path)[1].lower()
# Обнаружение двоичных файлов и генерация URL-адресов # Обнаружение двоичных файлов (вместо изображений раньше выводилась ссылка-заготовка
# https://raw.githubusercontent.com/.../ — она никуда не вела)
if file_ext in BINARY_EXTENSIONS or is_likely_binary(file_path): if file_ext in BINARY_EXTENSIONS or is_likely_binary(file_path):
# GitHub raw URL if file_ext in {'.png', '.jpg', '.jpeg', '.gif', '.svg', '.ico', '.webp', '.bmp'}:
if file_ext in {'.png', '.jpg', '.jpeg', '.gif', '.svg', '.ico'}: content_lines.append("[Image - content not displayed]")
# Структура URL - настраивается на основе фактического хранилища
github_url = f"https://raw.githubusercontent.com/.../{relative_path.replace(os.sep, '/')}"
content_lines.append(github_url)
else: else:
content_lines.append("[Binary file - content not displayed]") content_lines.append("[Binary file - content not displayed]")
else: else:
@ -141,25 +140,22 @@ def is_likely_binary(file_path):
chunk = f.read(8192) chunk = f.read(8192)
# Обнаружение нулевого байта - надежный бинарный индикатор # Обнаружение нулевого байта - надежный бинарный индикатор
return b'\x00' in chunk return b'\x00' in chunk
except: except OSError:
return True return True
def main(argv=None):
parser = argparse.ArgumentParser(description="Дерево каталогов проекта и содержимое текстовых файлов в один .txt")
parser.add_argument("path", help="папка проекта")
parser.add_argument("-o", "--output", help="куда сохранить (по умолчанию <папка>_rep.txt)")
args = parser.parse_args(argv)
project_path = os.path.abspath(args.path)
output = args.output or os.path.basename(project_path.rstrip(os.sep)) + "_rep.txt"
with open(output, "w", encoding="utf-8") as f:
f.write(generate_complete_project_structure(project_path))
print(output)
if __name__ == "__main__": if __name__ == "__main__":
# Конфигурация: измените путь к целевому каталогу проекта main()
project_path = r"D:\Programs\GitHub\deev.space\static"
# project_path = r"D:/Programs/GitHub/openoffice"
# project_path = "."
print("Приступаем к формированию комплексной структуры проекта...")
tree_output = generate_complete_project_structure(project_path)
output_filename = project_path.split('\\')[-1] + "_rep.txt"
try:
with open(output_filename, "w", encoding="utf-8") as f:
f.write(tree_output)
print(f"\nПолная проектная документация, сохраненная в: {output_filename}")
except Exception as e:
print(f"Предупреждение: Не удалось сохранить файл - {e}")
print("Формирование структуры проекта успешно завершено!")