From 1bd68c9a0ec86d4f185a3e942e95ea57e27269d0 Mon Sep 17 00:00:00 2001 From: smolkik-code Date: Wed, 17 Jun 2026 12:02:51 +0700 Subject: [PATCH] rewrite: switch from instaloader to Instagram Private API instaloader blocked by Instagram GraphQL 403. Using mobile API instead: - i.instagram.com/api/v1 endpoints (same as Android app) - Direct cookie loading from cookies.txt - Stories, posts, albums, videos - Random delays between requests - Removed instaloader dependency, using requests directly --- .DS_Store | Bin 0 -> 8196 bytes .env.example | 8 +- Dockerfile | 4 +- instagram_monitor.py | 424 +++++++++++++++++++++---------------------- requirements.txt | 3 +- 5 files changed, 217 insertions(+), 222 deletions(-) create mode 100644 .DS_Store diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..f62d9dee7c00d1cd7a6d456e5af6ea3b5cdb8228 GIT binary patch literal 8196 zcmeHMTWl0n7(U;$&>1_xv{ER@vMUQnptVc66^gjMaH+I{Y)iMLSZ8;Jc4RtJcV@R- zQY|qWZ;5vkpNNV+5;Ynz27*4BC^4FbL=uV72l2%xUeJjCGiSE6ToO%;0iBb~`Of*z z|DQSkKi@xR&n#mM&3SDlW7Ui?kuDFFa;mOVym9ZkrU-##8X?G@Wk#IgjA>`6f8rYL zP!L8Sj6fKHFalu&!U)_95uiQWE8-3IeLfo2VFbbm{4XQmy&qz9d6*1vjHCbRpvpf2 z5SEhIFVt4N24O6~WPoEFX@m;6DIqsSaKr#NCwVN;O9nW`k()CFhYtjIMsPxbe|MTc z=BqO#Ifiu@fiMEsM!-9}mokm1%w_82^Lu94G#odPI18bqblUW?GO4V*Y;S7N9!a^X zn|GSk+#avjY1?L|*j6*U4Sl3vncZSru4-6%&NnbLomBd|4NJ2}n(e%$IlgU*4@iGDqh#PtAD>aWZBPf7ZAJ#*o>=OdUmou znv*p%v_4OH!b#I9wD6A9%rtMx;C4(<}F-jjU~@j6-_AGRq-P+o-!U zrx(b>?yPO*hb*UGsmQ8kHgBq~-ej7dGRbB+*_>^fJ8j2s4cqETyZQmQIMI}$bgyIQ z+!lYbenrVU`YxU3G75St%}7&KdUx9~>`mkB)^iSFJelA=OUGx4Y-C@zW0hkY6?vcuNA88v?-5?OSYDDLs(EjgWP zE|w!JYREwCbSOZ?r$yon)uj8FrSPXBXHn>=OHx{l+e%1TreI z0EG1=gZ(fu1RE|!@hBd{5j=_K@d94N%Xl4c;w>D*J2;My zZ~`CWGkk$BaRy)EJbuCj{EUnE9hdQkP$A4077I&+r9z#M5bA|h!WLnxuv6#}GQxl` z$bArs`L}n%C%+*aB$%9P!S_Jn`1MmQx^-LIeRBIvpK9OE^pflL*;RAq#+EFvTitL^ z(*y-MV~Zr6On#s5Q?kV8yU&TMx$0-3yfE5D385djl7>7ji^-^zd5J5Id)jx(@u)J5 zQigYz%FCk4bV?WAT@kH~DgtFG?6T~wLQ2hygv3Dl` literal 0 HcmV?d00001 diff --git a/.env.example b/.env.example index 3a7bff4..f8a633d 100644 --- a/.env.example +++ b/.env.example @@ -1,10 +1,10 @@ # Telegram -TELEGRAM_BOT_TOKEN=your_bot_token_here -TELEGRAM_CHAT_ID=your_telegram_user_id_here +TELEGRAM_BOT_TOKEN=7949149242:AAEwFjIEAocJfVpnolqZHD95qrhb1kvkndo +TELEGRAM_CHAT_ID=146466816 # Instagram -INSTAGRAM_USERNAME=your_instagram_username -INSTAGRAM_PASSWORD=your_instagram_password +INSTAGRAM_USERNAME=smolkik_adm +INSTAGRAM_PASSWORD=lL1@^7$rg8a8 # Settings CHECK_INTERVAL_MINUTES=15 diff --git a/Dockerfile b/Dockerfile index 4efbd12..0b7af69 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.12-slim WORKDIR /app RUN apt-get update && \ - apt-get install -y --no-install-recommends ffmpeg && \ + apt-get install -y --no-install-recommends curl && \ rm -rf /var/lib/apt/lists/* COPY requirements.txt . @@ -11,6 +11,6 @@ RUN pip install --no-cache-dir -r requirements.txt COPY . . -RUN mkdir -p /data/downloads /data/session +RUN mkdir -p /data/downloads CMD ["python", "bot.py"] diff --git a/instagram_monitor.py b/instagram_monitor.py index 71cacfc..867b7e3 100644 --- a/instagram_monitor.py +++ b/instagram_monitor.py @@ -1,43 +1,39 @@ -import http.cookiejar import json -import os import random -import shutil import time from datetime import datetime from pathlib import Path from typing import AsyncIterator -import instaloader +import requests 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, + "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" + class InstagramMonitor: def __init__(self) -> None: - self.loader = instaloader.Instaloader( - download_video_thumbnails=False, - download_geotags=False, - download_comments=False, - save_metadata=False, - compress_json=False, - post_metadata_txt_pattern="", - max_connection_attempts=3, - request_timeout=60, - ) - self.loader.context._session.headers.update({ - "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Mobile/15E148 Safari/604.1", - "X-IG-App-ID": "936619743392459", - "X-IG-WWW-Claim": "0", - "X-Requested-With": "XMLHttpRequest", - "Accept-Language": "en-US,en;q=0.9", - }) - self._session_path = Path(config.session_file) + self.session = requests.Session() + self.session.headers.update(IG_HEADERS) self._download_dir = Path(config.download_dir) self._state_file = self._download_dir / "state.json" + self._cookies_path = self._download_dir / "cookies.txt" self._download_dir.mkdir(parents=True, exist_ok=True) self._state: dict[str, str] = self._load_state() + self._logged_in = False def _load_state(self) -> dict[str, str]: if self._state_file.exists(): @@ -49,192 +45,205 @@ class InstagramMonitor: with open(self._state_file, "w") as f: json.dump(self._state, f, indent=2) - def login(self) -> bool: - # Method 1: load existing instaloader session file - # Check configured path and also common alternatives - session_candidates = [ - self._session_path, - Path(config.download_dir).parent / "session-ig", - Path(config.download_dir) / "session-ig", - ] - for session_path in session_candidates: - if session_path.exists(): - try: - self.loader.load_session_from_file(config.instagram_username, str(session_path)) - # Verify session belongs to the right account - if self._verify_session(): - logger.info(f"Session loaded from {session_path}") - return True - else: - logger.warning(f"Session from {session_path} belongs to wrong account, skipping") - self._reset_session() - except Exception as e: - logger.warning(f"Session file {session_path} invalid: {e}") + def _random_delay(self, min_sec: float = 2.0, max_sec: float = 5.0) -> None: + time.sleep(random.uniform(min_sec, max_sec)) - # Method 2: load cookies.txt exported from browser - # Check both download_dir and parent (/data/) since user may place file in either - cookies_candidates = [ + def login(self) -> bool: + if self._logged_in: + return True + + # Try cookies.txt + cookies_path = None + for candidate in [ Path(config.download_dir) / "cookies.txt", Path(config.download_dir).parent / "cookies.txt", - ] - cookies_path = None - for candidate in cookies_candidates: + self._cookies_path, + ]: if candidate.exists(): cookies_path = candidate break - if cookies_path: - try: - self._load_cookies_from_txt(cookies_path) - logger.info(f"Session loaded from {cookies_path}") - return True - except Exception as e: - logger.warning(f"cookies.txt invalid or missing sessionid cookie: {e}") - # Check what cookies are actually in the file - try: - import http.cookiejar - cj = http.cookiejar.MozillaCookieJar(str(cookies_path)) - cj.load(ignore_discard=True, ignore_expires=True) - cookie_names = [c.name for c in cj] - logger.info(f"Cookies found in file: {cookie_names}") - if "sessionid" not in cookie_names: - logger.error( - "cookies.txt is missing 'sessionid' cookie. " - "You must be logged into instagram.com in the browser before exporting." - ) - except Exception: - pass + if not cookies_path: + logger.error("No cookies.txt found") + return False - # Method 3: load cookies.json exported from browser extension - cookies_json_path = Path(config.download_dir) / "cookies.json" - if cookies_json_path.exists(): - try: - self._load_cookies_from_json(cookies_json_path) - logger.info("Session loaded from cookies.json") - return True - except Exception as e: - logger.warning(f"cookies.json invalid: {e}") + try: + with open(cookies_path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t") + if len(parts) >= 7: + name, value = parts[5], parts[6] + self.session.cookies.set(name, value, domain=".instagram.com") - # Method 4: direct login (often blocked by Instagram now) - if config.instagram_password: - try: - self.loader.login(config.instagram_username, config.instagram_password) - self.loader.save_session_to_file(str(self._session_path)) - logger.info("Logged in via direct login and session saved") - return True - except Exception as e: - logger.error(f"Direct login failed: {e}") - logger.info( - "Instagram blocks most automated logins. " - "Use one of these methods instead:\n" - " 1. Run 'instaloader --login=USERNAME' on host, copy session to data/\n" - " 2. Export cookies from browser (must have sessionid cookie)" - ) + sessionid = self.session.cookies.get("sessionid") + ds_user_id = self.session.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 - logger.error("No valid session or credentials found") - return False + logger.info(f"Cookies loaded: ds_user_id={ds_user_id}") - def _load_cookies_from_txt(self, path: Path) -> None: - cj = http.cookiejar.MozillaCookieJar(str(path)) - cj.load(ignore_discard=True, ignore_expires=True) - self.loader.context._session.cookies.update(cj) - # Show which account the cookies belong to - for cookie in cj: - if cookie.name == "ds_user_id": - logger.info(f"Cookie ds_user_id = {cookie.value} (expected username: {config.instagram_username})") - if cookie.name == "sessionid": - logger.info(f"Cookie sessionid found, expires: {cookie.expires}") - # Verify session works - instaloader.Profile.from_username(self.loader.context, config.instagram_username) + resp = self.session.get(f"{API_BASE}/accounts/current_user/", headers=IG_HEADERS, timeout=30) + if resp.status_code == 200: + data = resp.json() + username = data.get("user", {}).get("username", "unknown") + logger.info(f"Logged in as: {username}") + self._logged_in = True + return True + else: + logger.error(f"Session validation failed: {resp.status_code} {resp.text[:200]}") + return False - def _verify_session(self) -> bool: - """Check if loaded session belongs to the expected account.""" - try: - profile = instaloader.Profile.from_username(self.loader.context, config.instagram_username) - return True - except instaloader.exceptions.ProfileNotExistsException: - return False - except Exception: - return False - - def _reset_session(self) -> None: - """Clear session cookies to start fresh.""" - self.loader.context._session.cookies.clear() - - def _random_delay(self, min_sec: float = 2.0, max_sec: float = 5.0) -> None: - """Random delay to avoid detection.""" - delay = random.uniform(min_sec, max_sec) - time.sleep(delay) - - def _load_cookies_from_json(self, path: Path) -> None: - with open(path) as f: - cookies = json.load(f) - for cookie in cookies: - self.loader.context._session.cookies.set( - cookie["name"], - cookie["value"], - domain=cookie.get("domain", ".instagram.com"), - path=cookie.get("path", "/"), - ) - instaloader.Profile.from_username(self.loader.context, config.instagram_username) - - def _get_profile_posts(self, username: str) -> list[instaloader.Post]: - try: - profile = instaloader.Profile.from_username(self.loader.context, username) - posts = [] - for post in profile.get_posts(): - if str(post.mediaid) in self._state: - break - posts.append(post) - return list(reversed(posts)) except Exception as e: - logger.error(f"Error fetching posts for {username}: {e}") - return [] + logger.error(f"Login error: {e}") + return False - def _get_profile_stories(self, username: str) -> list[instaloader.StoryItem]: + def _get_followees(self) -> list[dict]: + followees = [] + max_id = None + + while True: + try: + params = {"count": 50} + if max_id: + params["max_id"] = max_id + + resp = self.session.get( + f"{API_BASE}/users/self/following/", + params=params, + timeout=30, + ) + + if resp.status_code != 200: + logger.error(f"Followees fetch failed: {resp.status_code}") + break + + data = resp.json() + users = data.get("users", []) + for user in users: + followees.append({ + "pk": user["pk"], + "username": user["username"], + "full_name": user.get("full_name", ""), + }) + + max_id = data.get("next_max_id") + if not max_id or not users: + break + + self._random_delay(1, 2) + + except Exception as e: + logger.error(f"Error fetching followees: {e}") + break + + return followees + + def _get_user_stories(self, user_pk: int) -> list[dict]: try: - profile = instaloader.Profile.from_username(self.loader.context, username) + resp = self.session.get( + f"{API_BASE}/feed/user/{user_pk}/story/", + timeout=30, + ) + if resp.status_code != 200: + return [] + + data = resp.json() + items = data.get("items", []) stories = [] - for story in self.loader.get_stories(userids=[profile.userid]): - if str(story.mediaid) in self._state: + for item in items: + media_id = str(item.get("pk", item.get("id", ""))) + if media_id in self._state: continue - stories.append(story) + stories.append(item) return stories except Exception as e: - logger.error(f"Error fetching stories for {username}: {e}") + logger.error(f"Error fetching stories for {user_pk}: {e}") return [] - def _download_media(self, item: instaloader.Post | instaloader.StoryItem) -> str | None: + def _get_user_posts(self, user_pk: int) -> list[dict]: try: - typename = type(item).__name__ - shortcode = getattr(item, "shortcode", None) or str(item.mediaid) - media_type = "reel" if getattr(item, "is_video", False) else "photo" + resp = self.session.get( + f"{API_BASE}/feed/user/{user_pk}/", + params={"count": 12}, + timeout=30, + ) + if resp.status_code != 200: + return [] - if typename == "StoryItem": - target_dir = self._download_dir / "stories" / item.owner_username + data = resp.json() + items = data.get("items", []) + posts = [] + for item in items: + media_id = str(item.get("pk", item.get("id", ""))) + if media_id in self._state: + break + posts.append(item) + return list(reversed(posts)) + except Exception as e: + 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: + try: + media_type = item.get("media_type", 1) + pk = str(item.get("pk", item.get("id", ""))) + + if is_story: + target_dir = self._download_dir / "stories" / username else: - target_dir = self._download_dir / "posts" / item.owner_username - + target_dir = self._download_dir / "posts" / username target_dir.mkdir(parents=True, exist_ok=True) - if typename == "Post": - self.loader.download_post(item, target=str(target_dir)) + # photo=1, video=2, album=8 + if media_type == 2: + # Video + video_versions = item.get("video_versions", []) + if not video_versions: + return None + url = video_versions[0]["url"] + ext = "mp4" + elif media_type == 8: + # Album + carousel = item.get("carousel_media", []) + if not carousel: + return None + first = carousel[0] + if first.get("media_type") == 2: + url = first["video_versions"][0]["url"] + ext = "mp4" + else: + url = first["image_versions2"]["candidates"][0]["url"] + ext = "jpg" else: - self.loader.download_storyitem(item, target=str(target_dir)) + # Photo + images = item.get("image_versions2", {}) + candidates = images.get("candidates", []) + if not candidates: + return None + url = candidates[0]["url"] + ext = "jpg" - for ext in [".jpg", ".mp4", ".webp"]: - media_file = target_dir / f"{shortcode}{ext}" - if media_file.exists(): - return str(media_file) + filepath = target_dir / f"{pk}.{ext}" + if filepath.exists(): + return str(filepath) - for f in target_dir.iterdir(): - if f.suffix in (".jpg", ".mp4", ".webp") and shortcode in f.name: - return str(f) + 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) + logger.info(f"Downloaded: {filepath.name}") + return str(filepath) + else: + logger.error(f"Download failed ({resp.status_code}): {url[:100]}") + return None - return None except Exception as e: - logger.error(f"Error downloading media {getattr(item, 'shortcode', '?')}: {e}") + logger.error(f"Error downloading media: {e}") return None async def check_updates(self) -> AsyncIterator[dict]: @@ -244,67 +253,54 @@ class InstagramMonitor: self._random_delay(3, 7) - try: - # First verify who we are logged in as - logger.info(f"Attempting to fetch profile: {config.instagram_username}") - profile = instaloader.Profile.from_username(self.loader.context, config.instagram_username) - logger.info(f"Logged in as: {profile.username} (id: {profile.userid}, followers: {profile.followers})") - except instaloader.exceptions.ProfileNotExistsException: - logger.error( - f"Profile '{config.instagram_username}' does not exist on Instagram. " - "Check the username in .env and make sure cookies match the same account." - ) - return - except Exception as e: - logger.error(f"Error fetching profile '{config.instagram_username}': {e}") - return + followees = self._get_followees() + logger.info(f"Checking {len(followees)} followees") - try: - followees = list(profile.get_followees()) - logger.info(f"Checking {len(followees)} followees") - except Exception as e: - logger.error(f"Error fetching followees: {e}") - return + for user in followees: + username = user["username"] + user_pk = user["pk"] - for followee in followees: - username = followee.username - - self._random_delay(3, 6) + self._random_delay(2, 5) try: - for story in self._get_profile_stories(username): - media_path = self._download_media(story) + stories = self._get_user_stories(user_pk) + for story in stories: + media_path = self._download_media(story, username, is_story=True) if media_path: - self._state[str(story.mediaid)] = datetime.utcnow().isoformat() + media_id = str(story.get("pk", story.get("id", ""))) + self._state[media_id] = datetime.utcnow().isoformat() self._save_state() yield { "type": "story", "username": username, "media_path": media_path, "caption": "", - "shortcode": str(story.mediaid), + "shortcode": media_id, } - self._random_delay(1, 3) + self._random_delay(1, 2) except Exception as e: logger.error(f"Error processing stories for {username}: {e}") - self._random_delay(2, 4) + self._random_delay(2, 3) try: - for post in self._get_profile_posts(username): - caption = post.caption or "" - media_path = self._download_media(post) + posts = self._get_user_posts(user_pk) + for post in posts: + caption = post.get("caption", {}) + caption_text = caption.get("text", "") if isinstance(caption, dict) else "" + media_path = self._download_media(post, username) if media_path: - self._state[str(post.mediaid)] = datetime.utcnow().isoformat() + media_id = str(post.get("pk", post.get("id", ""))) + self._state[media_id] = datetime.utcnow().isoformat() self._save_state() yield { "type": "post", "username": username, "media_path": media_path, - "caption": caption[:500], - "shortcode": post.shortcode, + "caption": caption_text[:500], + "shortcode": media_id, } - self._random_delay(2, 5) + self._random_delay(2, 4) except Exception as e: logger.error(f"Error processing posts for {username}: {e}") @@ -322,4 +318,4 @@ class InstagramMonitor: age_hours = (now - datetime.fromtimestamp(f.stat().st_mtime)).total_seconds() / 3600 if age_hours > max_age_hours: f.unlink() - logger.debug(f"Cleaned up old file: {f.name}") + logger.debug(f"Cleaned up: {f.name}") diff --git a/requirements.txt b/requirements.txt index d314c43..d139378 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,4 @@ aiogram>=3.12,<4.0 -instaloader>=4.15,<5.0 +requests>=2.31,<3.0 loguru>=0.7,<1.0 aiofiles>=24.1,<25.0 -python-dotenv>=1.0,<2.0