diff --git a/.env.example b/.env.example index f8a633d..01bfbd4 100644 --- a/.env.example +++ b/.env.example @@ -1,10 +1,9 @@ # Telegram -TELEGRAM_BOT_TOKEN=7949149242:AAEwFjIEAocJfVpnolqZHD95qrhb1kvkndo -TELEGRAM_CHAT_ID=146466816 +TELEGRAM_BOT_TOKEN=your_bot_token_from_botfather +TELEGRAM_CHAT_ID=your_telegram_user_id -# Instagram -INSTAGRAM_USERNAME=smolkik_adm -INSTAGRAM_PASSWORD=lL1@^7$rg8a8 +# Instagram (используется только для логирования, авторизация через cookies.txt) +INSTAGRAM_USERNAME=your_instagram_login # Settings CHECK_INTERVAL_MINUTES=15 diff --git a/README.md b/README.md index 6731215..5a66516 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,7 @@ docker compose up -d | `/start` | Информация о боте | | `/check` | Проверить обновления сейчас | | `/status` | Статистика базы | +| `/health` | Статус подключений | | `/stop` | Остановить мониторинг | ## API diff --git a/bot.py b/bot.py index 5a0afa9..e8c788d 100644 --- a/bot.py +++ b/bot.py @@ -1,22 +1,24 @@ import asyncio -import os +import signal +import sys +from datetime import datetime, timezone from pathlib import Path from aiogram import Bot, Dispatcher, F, Router from aiogram.filters import CommandStart from aiogram.types import FSInputFile, Message from loguru import logger -import sys logger.remove() logger.add(sys.stderr, level="DEBUG") from config import config -from instagram_monitor import InstagramMonitor, API_BASE +from instagram_monitor import InstagramMonitor router = Router() bot = Bot(token=config.telegram_bot_token) monitor = InstagramMonitor() +_shutdown_event = asyncio.Event() @router.message(CommandStart()) @@ -28,6 +30,7 @@ async def cmd_start(message: Message) -> None: "Команды:\n" "/check — проверить сейчас\n" "/status — статус бота\n" + "/health — здоровье бота\n" "/stop — остановить мониторинг" ) else: @@ -46,12 +49,23 @@ async def cmd_status(message: Message) -> None: ) +@router.message(F.text == "/health") +async def cmd_health(message: Message) -> None: + if message.from_user and message.from_user.id == config.telegram_chat_id: + now = datetime.now(timezone.utc).isoformat(timespec="seconds") + await message.answer( + f"Бот: жив\n" + f"Время: {now} UTC\n" + f"Instagram: {'авторизован' if monitor._logged_in else 'не авторизован'}\n" + f"Подписки: {len(monitor._followees_cache) if monitor._followees_cache else 'не загружены'}" + ) + + @router.message(F.text == "/stop") async def cmd_stop(message: Message) -> None: if message.from_user and message.from_user.id == config.telegram_chat_id: await message.answer("Останавливаю мониторинг...") - import sys - sys.exit(0) + _shutdown_event.set() @router.message(F.text == "/check") @@ -71,14 +85,9 @@ async def send_media(update: dict) -> None: username = update["username"] media_type = update["type"] caption = update["caption"] - shortcode = update["shortcode"] header = f"{'Сторис' if media_type == 'story' else 'Пост'} от @{username}" - - if caption: - text = f"{header}\n\n{caption}" - else: - text = header + text = f"{header}\n\n{caption}" if caption else header path = Path(media_path) if not path.exists(): @@ -110,7 +119,7 @@ async def cmd_debug(message: Message) -> None: username = parts[1].lstrip("@") await message.answer(f"Ищу @{username}...") - followees = monitor._get_followees() + followees = await monitor._get_followees() user = next((u for u in followees if u["username"] == username), None) if not user: await message.answer(f"@{username} не найден в подписках") @@ -119,24 +128,48 @@ async def cmd_debug(message: Message) -> None: user_pk = user["pk"] await message.answer(f"user_pk: {user_pk}\nПроверяю сториз...") - resp = monitor.session.get( - f"{API_BASE}/feed/user/{user_pk}/story/", + resp = await monitor.client.get( + "https://www.instagram.com/api/v1/feed/user/" + f"{user_pk}/story/", params={"user_id": user_pk}, - timeout=30, + headers={ + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/120.0.0.0 Safari/537.36", + "X-IG-App-ID": "936619743392459", + "Accept": "*/*", + }, ) await message.answer(f"HTTP {resp.status_code}\nResponse: {resp.text[:800]}") async def monitor_loop() -> None: logger.info("Starting monitor loop") - while True: + while not _shutdown_event.is_set(): try: async for update in monitor.check_updates(): await send_media(update) monitor.cleanup_downloads(max_age_hours=48) except Exception as e: logger.error(f"Monitor loop error: {e}") - await asyncio.sleep(config.check_interval_minutes * 60) + + try: + await asyncio.wait_for( + _shutdown_event.wait(), + timeout=config.check_interval_minutes * 60, + ) + except asyncio.TimeoutError: + pass + + logger.info("Monitor loop stopped") + + +async def shutdown() -> None: + logger.info("Shutting down...") + _shutdown_event.set() + await bot.session.close() + await monitor.close() + logger.info("Shutdown complete") async def main() -> None: @@ -146,6 +179,10 @@ async def main() -> None: logger.error(e) return + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, lambda: asyncio.create_task(shutdown())) + dp = Dispatcher() dp.include_router(router) @@ -153,8 +190,8 @@ async def main() -> None: logger.info("Bot started") await dp.start_polling(bot) + await shutdown() if __name__ == "__main__": - import asyncio asyncio.run(main()) diff --git a/config.py b/config.py index 02c31ce..8c16b3c 100644 --- a/config.py +++ b/config.py @@ -1,5 +1,5 @@ import os -from dataclasses import dataclass, field +from dataclasses import dataclass @dataclass @@ -7,10 +7,8 @@ class Config: telegram_bot_token: str = os.getenv("TELEGRAM_BOT_TOKEN", "") telegram_chat_id: int = int(os.getenv("TELEGRAM_CHAT_ID", "0")) instagram_username: str = os.getenv("INSTAGRAM_USERNAME", "") - instagram_password: str = os.getenv("INSTAGRAM_PASSWORD", "") check_interval_minutes: int = int(os.getenv("CHECK_INTERVAL_MINUTES", "15")) download_dir: str = os.getenv("DOWNLOAD_DIR", "/data/downloads") - session_file: str = os.getenv("SESSION_FILE", "/data/session") max_media_size_mb: int = int(os.getenv("MAX_MEDIA_SIZE_MB", "50")) max_post_age_hours: int = int(os.getenv("MAX_POST_AGE_HOURS", "24")) concurrent_checks: int = int(os.getenv("CONCURRENT_CHECKS", "5")) @@ -21,10 +19,6 @@ class Config: errors.append("TELEGRAM_BOT_TOKEN is required") if not self.telegram_chat_id: errors.append("TELEGRAM_CHAT_ID is required") - if not self.instagram_username: - errors.append("INSTAGRAM_USERNAME is required") - if not self.instagram_password: - errors.append("INSTAGRAM_PASSWORD is required") return errors diff --git a/instagram_monitor.py b/instagram_monitor.py index 01a1441..7ca6436 100644 --- a/instagram_monitor.py +++ b/instagram_monitor.py @@ -6,28 +6,38 @@ from datetime import datetime, timedelta, timezone from pathlib import Path from typing import AsyncIterator -import requests +import aiofiles +import httpx from loguru import logger from config import config -IG_APP_VERSION = "269.0.0.18.75" -IG_USER_AGENT = f"Instagram {IG_APP_VERSION} Android (30/11; 420dpi; 1080x2400; samsung; SM-G991B; o1s; exynos2100; en_US; 458229258)" -IG_HEADERS = { - "User-Agent": IG_USER_AGENT, +API_BASE = "https://i.instagram.com/api/v1" +WEB_API_BASE = "https://www.instagram.com/api/v1" + +WEB_HEADERS = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + "X-IG-App-ID": "936619743392459", + "X-Requested-With": "XMLHttpRequest", + "Accept": "*/*", + "Referer": "https://www.instagram.com/", +} + +MOBILE_HEADERS = { + "User-Agent": "Instagram 269.0.0.18.75 Android (30/11; 420dpi; 1080x2400; " + "samsung; SM-G991B; o1s; exynos2100; en_US; 458229258)", "X-IG-App-ID": "567067343352127", "X-IG-Android-ID": "android-8f6d1cb2fe5f0a0b", "Accept-Language": "en-US", "Accept-Encoding": "gzip", - "Connection": "keep-alive", } -API_BASE = "https://i.instagram.com/api/v1" +FOLLOWEES_CACHE_TTL = 300 # 5 minutes class Database: def __init__(self, db_path: Path) -> None: - self._db_path = db_path self._conn = sqlite3.connect(str(db_path), check_same_thread=False) self._conn.execute("PRAGMA journal_mode=WAL") self._init_db() @@ -40,6 +50,7 @@ class Database: media_type TEXT NOT NULL, shortcode TEXT, file_path TEXT, + expiring_at TEXT, downloaded_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_username ON media(username); @@ -52,11 +63,14 @@ class Database: return cur.fetchone() is not None def mark_downloaded(self, media_id: str, username: str, media_type: str, - shortcode: str = "", file_path: str = "") -> None: + shortcode: str = "", file_path: str = "", + expiring_at: str = "") -> None: self._conn.execute( - "INSERT OR IGNORE INTO media (media_id, username, media_type, shortcode, file_path, downloaded_at) " - "VALUES (?, ?, ?, ?, ?, ?)", - (media_id, username, media_type, shortcode, file_path, datetime.utcnow().isoformat()), + "INSERT OR IGNORE INTO media " + "(media_id, username, media_type, shortcode, file_path, expiring_at, downloaded_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (media_id, username, media_type, shortcode, file_path, + expiring_at, datetime.utcnow().isoformat()), ) self._conn.commit() @@ -77,26 +91,67 @@ class Database: self._conn.commit() return deleted + def cleanup_expired(self) -> int: + now = datetime.utcnow().isoformat() + cur = self._conn.execute( + "SELECT file_path FROM media WHERE expiring_at IS NOT NULL AND expiring_at < ?", + (now,), + ) + files = cur.fetchall() + deleted = 0 + for (fp,) in files: + if fp and Path(fp).exists(): + Path(fp).unlink() + deleted += 1 + self._conn.execute( + "DELETE FROM media WHERE expiring_at IS NOT NULL AND expiring_at < ?", + (now,), + ) + self._conn.commit() + return deleted + def close(self) -> None: self._conn.close() class InstagramMonitor: def __init__(self) -> None: - self.session = requests.Session() - self.session.headers.update(IG_HEADERS) + self.client = httpx.AsyncClient( + headers={**MOBILE_HEADERS}, + timeout=httpx.Timeout(30.0, connect=15.0), + follow_redirects=True, + ) self._download_dir = Path(config.download_dir) self._db_path = self._download_dir / "media.db" - self._cookies_path = self._download_dir / "cookies.txt" self._download_dir.mkdir(parents=True, exist_ok=True) self.db = Database(self._db_path) self._logged_in = False self._own_user_id: int = 0 + self._followees_cache: list[dict] | None = None + self._followees_cache_ts: float = 0 - def _random_delay(self, min_sec: float = 2.0, max_sec: float = 5.0) -> None: - time.sleep(random.uniform(min_sec, max_sec)) + async def _request(self, method: str, url: str, **kwargs) -> httpx.Response: + for attempt in range(3): + resp = await self.client.request(method, url, **kwargs) + if resp.status_code == 429: + wait = 60 * (attempt + 1) + logger.warning(f"Rate limited (429), waiting {wait}s...") + await asyncio.sleep(wait) + continue + if resp.status_code == 401 and attempt < 2: + logger.warning(f"Session expired (401), waiting {30}s...") + self._logged_in = False + await asyncio.sleep(30) + if await self.login(): + continue + break + return resp - def login(self) -> bool: + async def _maybe_renew_session(self) -> None: + if not self._logged_in: + await self.login() + + async def login(self) -> bool: if self._logged_in: return True @@ -104,7 +159,7 @@ class InstagramMonitor: for candidate in [ Path(config.download_dir) / "cookies.txt", Path(config.download_dir).parent / "cookies.txt", - self._cookies_path, + self._download_dir / "cookies.txt", ]: if candidate.exists(): cookies_path = candidate @@ -123,10 +178,10 @@ class InstagramMonitor: parts = line.split("\t") if len(parts) >= 7: name, value = parts[5], parts[6] - self.session.cookies.set(name, value, domain=".instagram.com") + self.client.cookies.set(name, value, domain=".instagram.com") - sessionid = self.session.cookies.get("sessionid") - ds_user_id = self.session.cookies.get("ds_user_id") + sessionid = self.client.cookies.get("sessionid") + ds_user_id = self.client.cookies.get("ds_user_id") if not sessionid or not ds_user_id: logger.error("cookies.txt missing sessionid or ds_user_id") return False @@ -134,48 +189,45 @@ class InstagramMonitor: logger.info(f"Cookies loaded: ds_user_id={ds_user_id}") self._own_user_id = int(ds_user_id) - # Set CSRF token from cookies for POST requests - csrf = self.session.cookies.get("csrftoken", "") + csrf = self.client.cookies.get("csrftoken", "") if csrf: - self.session.headers["X-CSRFToken"] = csrf - logger.info(f"CSRF token set") + self.client.headers["X-CSRFToken"] = csrf - resp = self.session.get( + resp = await self.client.get( f"{API_BASE}/friendships/{ds_user_id}/following/", params={"count": 1}, - timeout=30, ) if resp.status_code == 200: logger.info("Login verified via followees endpoint") self._logged_in = True return True - elif resp.status_code == 401: - logger.warning("Rate limited by Instagram, waiting 60s...") - time.sleep(60) - # Retry once - resp = self.session.get( + + if resp.status_code == 401: + logger.warning("Session expired, waiting 60s...") + await asyncio.sleep(60) + resp = await self.client.get( f"{API_BASE}/friendships/{ds_user_id}/following/", params={"count": 1}, - timeout=30, ) if resp.status_code == 200: logger.info("Login verified after retry") self._logged_in = True return True - logger.error(f"Login check still failed after retry: {resp.status_code}") - return False - else: - logger.error(f"Login check failed: {resp.status_code} {resp.text[:200]}") - return False + + logger.error(f"Login check failed: {resp.status_code} {resp.text[:200]}") + return False except Exception as e: logger.error(f"Login error: {e}") return False - def _get_followees(self) -> list[dict]: + async def _get_followees(self) -> list[dict]: + now = time.time() + if self._followees_cache and (now - self._followees_cache_ts) < FOLLOWEES_CACHE_TTL: + return self._followees_cache + followees = [] max_id = None - own_user_id = self._own_user_id while True: try: @@ -183,12 +235,11 @@ class InstagramMonitor: if max_id: params["max_id"] = max_id - resp = self.session.get( - f"{API_BASE}/friendships/{own_user_id}/following/", + resp = await self._request( + "GET", + f"{API_BASE}/friendships/{self._own_user_id}/following/", params=params, - timeout=30, ) - if resp.status_code != 200: logger.error(f"Followees fetch failed: {resp.status_code} {resp.text[:200]}") break @@ -206,31 +257,29 @@ class InstagramMonitor: if not max_id or not users: break - self._random_delay(1, 2) + await asyncio.sleep(random.uniform(1, 2)) except Exception as e: logger.error(f"Error fetching followees: {e}") break + self._followees_cache = followees + self._followees_cache_ts = now return followees - def _get_user_stories(self, user_pk: int) -> list[dict]: - """Fetch stories via web API profile endpoint.""" + async def _get_user_stories(self, user_pk: int) -> list[dict]: try: - resp = self.session.get( - f"https://www.instagram.com/api/v1/feed/user/{user_pk}/story/", + resp = await self._request( + "GET", + f"{WEB_API_BASE}/feed/user/{user_pk}/story/", params={"user_id": user_pk}, - headers={ - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", - "X-IG-App-ID": "936619743392459", - "X-Requested-With": "XMLHttpRequest", - "Accept": "*/*", - "Referer": f"https://www.instagram.com/", - }, - timeout=30, + headers=WEB_HEADERS, ) if resp.status_code != 200: - logger.debug(f"Stories {user_pk}: HTTP {resp.status_code} {resp.text[:300]}") + return [] + + text = resp.text.strip() + if not text: return [] data = resp.json() @@ -246,58 +295,19 @@ class InstagramMonitor: media_id = str(item.get("pk", item.get("id", ""))) if self.db.is_downloaded(media_id): continue + item["_expiring_at"] = reel.get("expiring_at", 0) stories.append(item) return stories except Exception as e: - logger.debug(f"Stories {user_pk} error: {e}") + logger.debug(f"Stories {user_pk}: {e}") return [] - def _get_all_stories(self) -> dict[int, list[dict]]: - """Fetch all stories from reels_tray.""" - user_stories: dict[int, list[dict]] = {} + async def _get_user_posts(self, user_pk: int) -> list[dict]: try: - resp = self.session.get( - f"{API_BASE}/feed/reels_tray/", - params={"include_feed_video": "true"}, - timeout=30, - ) - if resp.status_code != 200: - logger.warning(f"reels_tray: HTTP {resp.status_code} - {resp.text[:200]}") - return user_stories - - data = resp.json() - trays = data.get("tray", []) - logger.info(f"reels_tray: found {len(trays)} users with stories") - - for tray in trays: - user_info = tray.get("user", {}) - user_pk = user_info.get("pk", 0) - username = user_info.get("username", "") - if not user_pk: - continue - - items = tray.get("items", []) - stories = [] - for item in items: - media_id = str(item.get("pk", item.get("id", ""))) - if self.db.is_downloaded(media_id): - continue - stories.append(item) - - if stories: - user_stories[user_pk] = stories - - return user_stories - except Exception as e: - logger.error(f"Error fetching reels_tray: {e}") - return user_stories - - def _get_user_posts(self, user_pk: int) -> list[dict]: - try: - resp = self.session.get( + resp = await self._request( + "GET", f"{API_BASE}/feed/user/{user_pk}/", params={"count": 12}, - timeout=30, ) if resp.status_code != 200: return [] @@ -325,7 +335,8 @@ class InstagramMonitor: logger.error(f"Error fetching posts for {user_pk}: {e}") return [] - def _download_media(self, item: dict, username: str, is_story: bool = False) -> str | None: + async def _download_media(self, item: dict, username: str, + is_story: bool = False) -> str | None: try: media_type = item.get("media_type", 1) pk = str(item.get("pk", item.get("id", ""))) @@ -365,32 +376,30 @@ class InstagramMonitor: if filepath.exists(): return str(filepath) - resp = self.session.get(url, timeout=60, stream=True) - if resp.status_code == 200: - with open(filepath, "wb") as f: - for chunk in resp.iter_content(chunk_size=8192): - f.write(chunk) + async with self.client.stream("GET", url) as resp: + if resp.status_code != 200: + logger.error(f"Download failed ({resp.status_code}): {url[:100]}") + return None + async with aiofiles.open(filepath, "wb") as f: + async for chunk in resp.aiter_bytes(chunk_size=65536): + await f.write(chunk) logger.info(f"Downloaded: {filepath.name}") return str(filepath) - else: - logger.error(f"Download failed ({resp.status_code}): {url[:100]}") - return None except Exception as e: logger.error(f"Error downloading media: {e}") return None async def check_updates(self) -> AsyncIterator[dict]: - if not self.login(): + if not await self.login(): logger.error("Cannot check updates without login") return - self._random_delay(3, 7) + await asyncio.sleep(random.uniform(3, 7)) - followees = self._get_followees() + followees = await self._get_followees() logger.info(f"Checking {len(followees)} followees") - # Fetch stories per-user via web API (mobile API endpoints are blocked) logger.info("Fetching stories via web API for each followee...") semaphore = asyncio.Semaphore(config.concurrent_checks) @@ -400,19 +409,25 @@ class InstagramMonitor: results = [] async with semaphore: - stories = await asyncio.to_thread(self._get_user_stories, user_pk) + stories = await self._get_user_stories(user_pk) if stories: logger.info(f"@{username}: {len(stories)} new stories") for story in stories: - media_path = await asyncio.to_thread( - self._download_media, story, username, True - ) + media_path = await self._download_media(story, username, True) if media_path: media_id = str(story.get("pk", story.get("id", ""))) shortcode = story.get("code", media_id) - self.db.mark_downloaded(media_id, username, "story", shortcode, media_path) + expiring_ts = story.get("_expiring_at", 0) + expiring_at = ( + datetime.fromtimestamp(expiring_ts, tz=timezone.utc).isoformat() + if expiring_ts else "" + ) + self.db.mark_downloaded( + media_id, username, "story", shortcode, + media_path, expiring_at, + ) results.append({ "type": "story", "username": username, @@ -420,7 +435,7 @@ class InstagramMonitor: "caption": "", "shortcode": shortcode, }) - await asyncio.to_thread(self._random_delay, 0.5, 1.5) + await asyncio.sleep(random.uniform(0.5, 1.5)) return results @@ -434,7 +449,7 @@ class InstagramMonitor: for item in result: yield item - # Fetch posts in parallel + logger.info("Fetching posts...") semaphore = asyncio.Semaphore(config.concurrent_checks) async def process_followee_posts(user: dict) -> list[dict]: @@ -443,7 +458,7 @@ class InstagramMonitor: results = [] async with semaphore: - posts = await asyncio.to_thread(self._get_user_posts, user_pk) + posts = await self._get_user_posts(user_pk) if posts: logger.info(f"@{username}: {len(posts)} new posts") @@ -451,13 +466,13 @@ class InstagramMonitor: for post in posts: caption = post.get("caption", {}) caption_text = caption.get("text", "") if isinstance(caption, dict) else "" - media_path = await asyncio.to_thread( - self._download_media, post, username - ) + media_path = await self._download_media(post, username) if media_path: media_id = str(post.get("pk", post.get("id", ""))) shortcode = post.get("code", media_id) - self.db.mark_downloaded(media_id, username, "post", shortcode, media_path) + self.db.mark_downloaded( + media_id, username, "post", shortcode, media_path, + ) results.append({ "type": "post", "username": username, @@ -465,7 +480,7 @@ class InstagramMonitor: "caption": caption_text[:500], "shortcode": shortcode, }) - await asyncio.to_thread(self._random_delay, 1, 2) + await asyncio.sleep(random.uniform(1, 2)) return results @@ -483,3 +498,10 @@ class InstagramMonitor: deleted = self.db.cleanup_old(max_age_hours) if deleted: logger.info(f"Cleaned up {deleted} old files") + expired = self.db.cleanup_expired() + if expired: + logger.info(f"Cleaned up {expired} expired stories") + + async def close(self) -> None: + await self.client.aclose() + self.db.close() diff --git a/requirements.txt b/requirements.txt index d139378..408a54b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ aiogram>=3.12,<4.0 -requests>=2.31,<3.0 +httpx>=0.28,<1.0 loguru>=0.7,<1.0 aiofiles>=24.1,<25.0