Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # Сервер розвернув на vps, ось таким чином:
- # https://github.com/aiogram/telegram-bot-api/tree/master/example
- # Мій код:
- import os
- from pathlib import Path
- from datetime import datetime
- import asyncio
- from loguru import logger
- from aiogram import Bot, Dispatcher, types, F, Router
- from aiogram.types import FSInputFile, ErrorEvent
- from aiogram.filters import Command
- from aiogram.utils.chat_action import ChatActionSender
- from aiogram.utils.keyboard import InlineKeyboardBuilder
- from aiogram.client.session.aiohttp import AiohttpSession
- from aiogram.client.telegram import TelegramAPIServer
- from corrupt import corrupt_file
- def create_dated_log_dir(base_dir="logs"):
- """Creates a directory structure for daily logs within the base directory.
- Args:
- base_dir (str, optional): The base directory for logs. Defaults to "logs".
- """
- today_str = datetime.utcnow().strftime("%Y-%m-%d")
- dated_dir = f"{base_dir}/{today_str}"
- if not os.path.exists(dated_dir):
- os.makedirs(dated_dir)
- return dated_dir
- Path("logs").mkdir(parents=True, exist_ok=True)
- logger.add(f"{create_dated_log_dir()}/{{time:HH-mm-ss}}.log",)
- bot_token = os.environ.get('BOT_TOKEN')
- if not bot_token:
- raise ValueError("BOT_TOKEN environment variable is not set")
- api_server = os.environ.get('API_SERVER')
- if not api_server:
- raise ValueError("API_SERVER environment variable is not set")
- session = AiohttpSession(api=TelegramAPIServer.from_base(api_server))
- bot = Bot(token=bot_token, session=session)
- dp = Dispatcher()
- router = Router()
- dp.startup.register(lambda: logger.info("Bot have been started"))
- dp.shutdown.register(lambda: logger.info("Bot have been stopped"))
- files_dir = Path("files")
- files_dir.mkdir(parents=True, exist_ok=True)
- @router.message(F.document)
- async def file_handler(message: types.Message):
- status = await message.answer("📥 Downloading file...")
- file_path = Path(files_dir / message.document.file_name)
- await bot.download(message.document, file_path)
- await status.edit_text("👾 Corrupting file...")
- corrupted_file_path = await corrupt_file(str(file_path))
- if file_path.exists():
- file_path.unlink()
- Path(corrupted_file_path).rename(file_path)
- await status.edit_text("📤 Sending file...")
- async with ChatActionSender.upload_document(bot=bot, chat_id=message.chat.id):
- await message.reply_document(FSInputFile(file_path))
- file_path.unlink(missing_ok=True)
- await status.delete()
- @router.message(F.photo)
- async def image_handler(message: types.Message):
- photo = await bot.get_file(message.photo[-1].file_id)
- file_ext = photo.file_path.split(".")[-1]
- file_path = Path(files_dir / f"image.{file_ext}")
- await bot.download(message.photo[-1], file_path)
- corrupted_file_path = await corrupt_file(str(file_path))
- file_path.unlink(missing_ok=True)
- Path(corrupted_file_path).rename(file_path)
- await message.reply_document(FSInputFile(file_path))
- file_path.unlink(missing_ok=True)
- @router.errors(F.update.message.as_("message"))
- async def errors_handler(error: ErrorEvent, message: types.Message):
- logger.exception(f"An error occurred while handling request from {message.from_user.id}: {error.exception}")
- await message.reply("❌Error happened. Please try again later.")
- async def main():
- await bot.delete_webhook(drop_pending_updates=True)
- dp.include_router(router)
- await dp.start_polling(bot)
- if __name__ == "__main__":
- try:
- asyncio.run(main())
- except KeyboardInterrupt:
- exit()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement