75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
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"}
|
||
]
|
||
)
|