From f49db9adf68b38efb227a9723de3d28a22387a6a Mon Sep 17 00:00:00 2001 From: smolkik-code Date: Fri, 26 Dec 2025 22:56:36 +0700 Subject: [PATCH] Initial commit --- bot1.py | 330 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 330 insertions(+) create mode 100644 bot1.py diff --git a/bot1.py b/bot1.py new file mode 100644 index 0000000..12fb42b --- /dev/null +++ b/bot1.py @@ -0,0 +1,330 @@ +import os +import logging +import asyncio +import json +import time +from functools import partial +from pathlib import Path +from typing import Optional, Tuple, Dict + +from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup +from telegram.ext import ( + ApplicationBuilder, + CommandHandler, + MessageHandler, + CallbackQueryHandler, + ContextTypes, + filters, +) +import yt_dlp +import instaloader + +# Настройки из переменных окружения с дефолтами +BASE_URL = os.getenv("BOT_BASE_URL", "http://127.0.0.1:8081/bot") + +DOWNLOAD_DIR = Path(os.getenv("DOWNLOAD_DIR", "downloads")) +COOKIES_FILE = Path(os.getenv("COOKIES_FILE", "cookies.txt")) +INSTALOADER_SESSION = Path(os.getenv("INSTALOADER_SESSION", "session-smolkik_adm")) +INSTALOADER_USER = os.getenv("INSTALOADER_USER", "smolkik_adm") + +CONNECT_TIMEOUT = int(os.getenv("CONNECT_TIMEOUT", 600)) +READ_TIMEOUT = int(os.getenv("READ_TIMEOUT", 3600)) + +DOWNLOAD_DIR.mkdir(exist_ok=True) + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s" +) + +class Downloader: + def __init__(self): + self.user_data = {} + self.cookies_enabled = COOKIES_FILE.exists() + self.session_enabled = INSTALOADER_SESSION.exists() + self._last_progress_update = 0 + + if self.cookies_enabled: + logging.info(f"Обнаружен файл cookies: {COOKIES_FILE}") + if self.session_enabled: + logging.info(f"Обнаружен файл сессии InstaLoader: {INSTALOADER_SESSION}") + + async def download_media(self, url: str, options: dict, message: Optional[object] = None) -> Tuple[str, dict]: + ydl_opts = { + 'outtmpl': str(DOWNLOAD_DIR / '%(title)s.%(ext)s'), + 'quiet': True, + **options + } + if self.cookies_enabled: + ydl_opts['cookiefile'] = str(COOKIES_FILE) + if message: + ydl_opts['progress_hooks'] = [partial(self.progress_hook, message=message)] + + logging.info(f"Начинаю загрузку: {url}") + try: + with yt_dlp.YoutubeDL(ydl_opts) as ydl: + loop = asyncio.get_running_loop() + info = await loop.run_in_executor(None, ydl.extract_info, url, True) + filepath = ydl.prepare_filename(info) + if options.get('extract_audio'): + filepath = os.path.splitext(filepath)[0] + '.mp3' + logging.info(f"Загрузка завершена: {filepath}") + return filepath, info + except yt_dlp.utils.DownloadError as e: + if "Private video" in str(e) and self.cookies_enabled: + logging.error("Не удалось получить доступ к приватному видео даже с cookies") + raise Exception("Не удалось получить доступ к видео. Проверьте cookies.") + raise + + async def download_instagram_story(self, url: str, message: object): + if not self.session_enabled: + await message.edit_text("⚠️ Нет Instagram session файла. Загрузите session-smolkik_adm в папку с ботом.") + return + + try: + username = self.extract_username_from_url(url) + loop = asyncio.get_running_loop() + + await message.edit_text("⏳ Проверяю сессию Instagram...") + + valid = await loop.run_in_executor(None, self.check_instagram_session) + if not valid: + await message.edit_text("⚠️ Сессия Instagram недействительна. Пожалуйста, обновите session-smolkik_adm.") + return + + await message.edit_text(f"⏳ Ищу сторис пользователя @{username}...") + await loop.run_in_executor(None, self._instaloader_download, username) + + story_files = [f for f in DOWNLOAD_DIR.glob(f"{username}*")] + if not story_files: + await message.edit_text("❌ Нет доступных сторис у пользователя.") + return + + for file in story_files: + with file.open('rb') as story_file: + await message.bot.send_document( + chat_id=message.chat_id, + document=story_file, + filename=file.name, + connect_timeout=CONNECT_TIMEOUT, + read_timeout=READ_TIMEOUT + ) + file.unlink() + await message.edit_text("✅ Истории успешно отправлены!") + + except Exception as e: + logging.error(f"Ошибка скачивания сторис: {str(e)}") + await message.edit_text(f"⚠️ Ошибка скачивания Stories: {str(e)}") + + def extract_username_from_url(self, url: str) -> Optional[str]: + parts = url.split('/') + username = None + if 'stories' in parts: + idx = parts.index('stories') + 1 + if idx < len(parts): + username = parts[idx] + else: + for i, part in enumerate(parts): + if 'instagram.com' in part and i + 1 < len(parts): + username = parts[i + 1] + break + if username: + username = username.split('?')[0] + return username + + def check_instagram_session(self) -> bool: + """Проверяет, что сессия Instaloader действительна""" + try: + L = instaloader.Instaloader() + L.load_session_from_file(INSTALOADER_USER, str(INSTALOADER_SESSION)) + user = L.test_login() + logging.info(f"✅ Instagram-сессия активна: {user}") + return True + except Exception as e: + logging.warning(f"❌ Сессия Instagram недействительна: {e}") + return False + + def _instaloader_download(self, username: str): + L = instaloader.Instaloader( + dirname_pattern=str(DOWNLOAD_DIR), + download_videos=True, + download_video_thumbnails=False + ) + if not INSTALOADER_SESSION.exists(): + raise FileNotFoundError(f"Session file {INSTALOADER_SESSION} не найден.") + L.load_session_from_file(INSTALOADER_USER, str(INSTALOADER_SESSION)) + profile = instaloader.Profile.from_username(L.context, username) + + for story in L.get_stories(userids=[profile.userid]): + for item in story.get_items(): + try: + L.download_storyitem(item, target=username) + except Exception as e: + logging.error(f"Ошибка при обработке элемента сторис: {e}") + + async def progress_hook(self, d: dict, message: Optional[object] = None): + now = time.time() + if d.get('status') == 'downloading' and message: + # Обновлять не чаще чем раз в секунду + if now - self._last_progress_update > 1: + percent = d.get('_percent_str', '?') + speed = d.get('_speed_str', '?') + eta = d.get('_eta_str', '?') + text = f"⬇️ Загрузка: {percent}\n🚀 Скорость: {speed}\n⏱ Осталось: {eta}" + try: + await message.edit_text(text) + except Exception: + pass + self._last_progress_update = now + + # Все методы ниже получили аннотации типов + async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + text = "🎬 Пришлите ссылку на видео или профиль Instagram (для Stories)\n" + if self.cookies_enabled: + text += "\n🔐 Используется cookies для скачивания приватного контента." + if self.session_enabled: + text += "\n🔐 Используется Instagram-сессия для сторис." + await update.message.reply_text(text) + + async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + url = update.message.text.strip() + if not url.startswith(('http://', 'https://')): + await update.message.reply_text("⚠️ Пожалуйста, отправьте корректную ссылку") + return + self.user_data[update.message.from_user.id] = {'url': url} + keyboard = [ + [InlineKeyboardButton("🎥 Видео", callback_data='video')], + [InlineKeyboardButton("🎧 Аудио", callback_data='audio')], + [InlineKeyboardButton("📁 Оригинал", callback_data='original')], + [InlineKeyboardButton("🌟 Сторис Instagram", callback_data='story')], + ] + reply_markup = InlineKeyboardMarkup(keyboard) + await update.message.reply_text("Выберите тип загрузки:", reply_markup=reply_markup) + + async def handle_quality_choice(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + query = update.callback_query + await query.answer() + user_id = query.from_user.id + data = query.data + if data == 'story': + url = self.user_data[user_id]['url'] + msg = await query.edit_message_text("⏳ Начинаю загрузку Instagram Stories...") + await self.download_instagram_story(url, msg) + elif data == 'audio': + await self.download_audio(query) + elif data == 'original': + await self.download_original(query) + elif data == 'video': + await self.select_video_quality(query) + elif data.startswith('quality_'): + await self.download_video(query, data.split('_')[1]) + + async def select_video_quality(self, query: object) -> None: + keyboard = [ + [InlineKeyboardButton("🔥 Максимальное", callback_data='quality_max')], + [InlineKeyboardButton("🖥 1080p", callback_data='quality_1080')], + [InlineKeyboardButton("📺 720p", callback_data='quality_720')], + [InlineKeyboardButton("📱 480p", callback_data='quality_480')], + ] + reply_markup = InlineKeyboardMarkup(keyboard) + await query.edit_message_text("Выберите качество видео:", reply_markup=reply_markup) + + async def download_audio(self, query: object) -> None: + user_id = query.from_user.id + url = self.user_data[user_id]['url'] + msg = await query.edit_message_text("⏳ Начинаю загрузку аудио...") + try: + options = { + 'format': 'bestaudio/best', + 'extract_audio': True, + 'audio_format': 'mp3', + 'postprocessors': [{ + 'key': 'FFmpegExtractAudio', + 'preferredcodec': 'mp3', + 'preferredquality': '192', + }], + } + filepath, info = await self.download_media(url, options, msg) + title = info.get('title', 'Аудио') + artist = info.get('uploader', 'Неизвестен') + with open(filepath, 'rb') as audio_file: + await query.message.reply_audio( + audio=audio_file, + title=title, + performer=artist, + connect_timeout=CONNECT_TIMEOUT, + read_timeout=READ_TIMEOUT + ) + await msg.edit_text("✅ Аудио успешно отправлено!") + os.remove(filepath) + except Exception as e: + logging.error(f"Ошибка: {str(e)}") + await msg.edit_text(f"⚠️ Ошибка: {str(e)}") + + async def download_original(self, query: object) -> None: + user_id = query.from_user.id + url = self.user_data[user_id]['url'] + msg = await query.edit_message_text("⏳ Начинаю загрузку оригинального файла...") + try: + options = {'format': 'best'} + filepath, info = await self.download_media(url, options, msg) + with open(filepath, 'rb') as file: + await query.message.reply_document( + document=file, + filename=os.path.basename(filepath), + connect_timeout=CONNECT_TIMEOUT, + read_timeout=READ_TIMEOUT + ) + await msg.edit_text("✅ Файл успешно отправлен!") + os.remove(filepath) + except Exception as e: + logging.error(f"Ошибка: {str(e)}") + await msg.edit_text(f"⚠️ Ошибка: {str(e)}") + + async def download_video(self, query: object, quality: str) -> None: + user_id = query.from_user.id + url = self.user_data[user_id]['url'] + quality_map = { + 'max': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best', + '1080': 'bestvideo[height<=1080][ext=mp4]+bestaudio[ext=m4a]/best[height<=1080][ext=mp4]/best', + '720': 'bestvideo[height<=720][ext=mp4]+bestaudio[ext=m4a]/best[height<=720][ext=mp4]/best', + '480': 'bestvideo[height<=480][ext=mp4]+bestaudio[ext=m4a]/best[height<=480][ext=mp4]/best', + } + msg = await query.edit_message_text(f"⏳ Начинаю загрузку видео ({quality})...") + try: + options = { + 'format': quality_map.get(quality, 'best'), + 'merge_output_format': 'mp4', + } + filepath, info = await self.download_media(url, options, msg) + with open(filepath, 'rb') as video_file: + await query.message.reply_video( + video=video_file, + supports_streaming=True, + connect_timeout=CONNECT_TIMEOUT, + read_timeout=READ_TIMEOUT + ) + await msg.edit_text("✅ Видео успешно отправлено!") + os.remove(filepath) + except Exception as e: + logging.error(f"Ошибка: {str(e)}") + await msg.edit_text(f"⚠️ Ошибка: {str(e)}") + +def main(): + downloader = Downloader() + app = ( + ApplicationBuilder() + .token(TOKEN) + .base_url(BASE_URL) + .connect_timeout(CONNECT_TIMEOUT) + .read_timeout(READ_TIMEOUT) + .build() + ) + app.add_handler(CommandHandler("start", downloader.start)) + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, downloader.handle_message)) + app.add_handler(CallbackQueryHandler(downloader.handle_quality_choice)) + logging.info("✅ Бот запущен и готов к работе (используется локальный Bot API)") + app.run_polling() + +if __name__ == "__main__": + main() \ No newline at end of file