smolkik_adm/DownloadSM
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update v0.1.1.1 bed

Browse Source
This commit is contained in:
smolkik-code
2025-12-27 02:16:41 +07:00
parent bc42c8775d
commit 7feac6f3a4
4 changed files with 123 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -1,3 +1,4 @@
.env
cookies.txt
.venv
.venv
session-smolkikadm

3
bot/config.py
View File

@@ -3,7 +3,8 @@ from pathlib import Path
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "REPLACE_ME")
BOT_BASE_URL = os.getenv("BOT_BASE_URL", "http://127.0.0.1:8081/bot")
INSTA_SESSION_FILE = Path("session-smolkikadm")
INSTA_USERNAME = "smolkikadm"
DOWNLOAD_DIR = Path(os.getenv("DOWNLOAD_DIR", "downloads"))
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)

77
bot/handlers.py
View File

@@ -8,6 +8,7 @@ from .utils import is_instagram_story_url, VIDEO_SUFFIXES, IMAGE_SUFFIXES, extra
from .ytdlp_client import build_opts, run
from .telegram_send import send_video, send_audio, send_document, send_photo
from .transcode import transcode_to_mobile_mp4
from .instaloader_client import download_user_stories
# Профили качества для видео
QUALITY_MAP = {
@@ -96,39 +97,37 @@ class Handlers:
# -------- Stories: одиночная --------
async def download_story(self, url: str, message, chat_id: int):
try:
opts = build_opts(for_video=True, outdir=DOWNLOAD_DIR, cookiefile=COOKIES_FILE, browser=BROWSER, message=message, out_for_playlist=True)
# Для видео принудительно mp4, изображения не пострадают, т.к. merge только для видео
opts['format'] = 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]'
filepath, info = await run(url, opts)
username = extract_instagram_username(url)
if not username:
await message.edit_text("⚠️ Не удалось определить username из ссылки.")
return
count = await self._send_playlist_mixed_with_images(info, message, chat_id)
# fallback одиночного файла
if count == 0 and filepath and filepath.exists():
suffix = (filepath.suffix or '').lower()
await message.edit_text("⏳ Загрузка Stories через Instaloader...")
files = download_user_stories(username)
if not files:
await message.edit_text("ℹ️ Stories не найдены или доступ ограничен.")
return
count = 0
for fp in files:
suffix = fp.suffix.lower()
if suffix in IMAGE_SUFFIXES:
await self._send_photo_safe(message, filepath, chat_id)
count = 1
await self._send_photo_safe(message, fp, chat_id)
elif suffix in VIDEO_SUFFIXES:
out_mp4 = await transcode_to_mobile_mp4(filepath)
out_mp4 = await transcode_to_mobile_mp4(fp)
try:
await self._send_video_safe(message, out_mp4, chat_id)
count = 1
finally:
out_mp4.unlink(missing_ok=True)
else:
await self._send_document_safe(message, filepath, chat_id)
count = 1
filepath.unlink(missing_ok=True)
await self._send_document_safe(message, fp, chat_id)
count += 1
await message.edit_text(f"✅ Stories отправлены! ({count} шт.)")
if count > 0:
await message.edit_text(f"✅ Stories отправлены! ({count} шт.)")
else:
await message.edit_text("ℹ️ Stories не найдены или доступ ограничен.")
except Exception as e:
try:
await message.edit_text(f"⚠️ Ошибка Stories: {str(e)[:100]}")
except:
await message.reply_text(f"⚠️ Ошибка Stories: {str(e)[:100]}")
await message.edit_text(f"⚠️ Ошибка Stories (Instaloader): {e}")
# -------- Stories: все активные у автора --------
async def download_all_stories(self, url: str, message, chat_id: int):
try:
@@ -137,18 +136,32 @@ class Handlers:
await message.edit_text("⚠️ Не удалось определить username из ссылки.")
return
list_url = f"https://www.instagram.com/stories/{username}/"
opts = build_opts(for_video=True, outdir=DOWNLOAD_DIR, cookiefile=COOKIES_FILE, browser=BROWSER, message=message, out_for_playlist=True)
opts['format'] = 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]'
_filepath, info = await run(list_url, opts)
await message.edit_text("⏳ Загрузка всех активных Stories автора через Instaloader...")
count = await self._send_playlist_mixed_with_images(info, message, chat_id)
if count:
await message.edit_text(f"✅ Отправлено элементов: {count} (у @{username})")
else:
files = download_user_stories(username)
if not files:
await message.edit_text(f"ℹ️ У @{username} нет активных сторис или доступ ограничен.")
return
count = 0
for fp in files:
suffix = fp.suffix.lower()
if suffix in IMAGE_SUFFIXES:
await self._send_photo_safe(message, fp, chat_id)
elif suffix in VIDEO_SUFFIXES:
out_mp4 = await transcode_to_mobile_mp4(fp)
try:
await self._send_video_safe(message, out_mp4, chat_id)
finally:
out_mp4.unlink(missing_ok=True)
else:
await self._send_document_safe(message, fp, chat_id)
count += 1
await message.edit_text(f"✅ Отправлено элементов: {count} (у @{username})")
except Exception as e:
await message.edit_text(f"⚠️ Ошибка при загрузке всех Stories автора: {e}")
await message.edit_text(f"⚠️ Ошибка при загрузке всех Stories автора (Instaloader): {e}")
# -------- Видео (с выбором качества) --------
async def download_video(self, url: str, message, quality_key: str, chat_id: int):

74
bot/instaloader_client.py Normal file
View File

@@ -0,0 +1,74 @@
from pathlib import Path
from typing import List, Optional
import instaloader
from instaloader import Profile
from .config import DOWNLOAD_DIR, INSTA_USERNAME, INSTA_SESSION_FILE
_instaloader: Optional[instaloader.Instaloader] = None
def get_instaloader() -> instaloader.Instaloader:
"""Создаёт Instaloader и грузит session-файл один раз."""
global _instaloader
if _instaloader is not None:
return _instaloader
L = instaloader.Instaloader(
dirname_pattern=str(DOWNLOAD_DIR / "{target}"),
filename_pattern="{date_utc}_story_{shortcode}",
download_video_thumbnails=False,
save_metadata=False,
download_geotags=False,
compress_json=False,
)
session_path = Path(INSTA_SESSION_FILE)
if not INSTA_USERNAME or not session_path.exists():
raise RuntimeError(
f"Instaloader session not configured: {INSTA_USERNAME=}, {session_path=}"
)
# загружаем сессию, созданную командой: instaloader -l INSTA_USERNAME
L.load_session_from_file(INSTA_USERNAME, filename=str(session_path))
_instaloader = L
return L
def download_user_stories(username: str) -> List[Path]:
"""Скачивает все актуальные stories пользователя и возвращает список файлов."""
L = get_instaloader()
try:
profile = Profile.from_username(L.context, username)
# Скачиваем stories для данного user id
L.download_stories(userids=[profile.userid])
except KeyError as e:
# Типичный кейс несовместимости Instaloader с текущим ответом Instagram:
# Instagram возвращает ответ без поля 'data' в GraphQL.
if str(e) == "'data'":
raise RuntimeError(
"Instagram вернул ответ без поля 'data' для GraphQL-запроса stories. "
"Это, как правило, несовместимость текущей версии Instaloader с API Instagram. "
"Обновите instaloader до последней версии или проверьте открытые issue проекта."
) from e
raise
except Exception as e:
raise RuntimeError(f"Ошибка при работе с @{username}: {e}") from e
user_dir = DOWNLOAD_DIR / username
if not user_dir.exists():
return []
# Возвращаем только медиа-файлы сторис
return sorted(
[
p
for p in user_dir.iterdir()
if p.is_file() and p.suffix.lower() in {".jpg", ".jpeg", ".png", ".mp4"}
]
)