330 lines
15 KiB
Python
330 lines
15 KiB
Python
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() |