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

rewrite: switch from instaloader to Instagram Private API

Browse Source 
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
This commit is contained in:
smolkik-code
2026-06-17 12:02:51 +07:00
parent bc1649d0ac
commit 1bd68c9a0e
5 changed files with 217 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored Normal file
View File

Binary file not shown.

8
.env.example
View File

@@ -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

4
Dockerfile
View File

@@ -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"]

424
instagram_monitor.py
View File

@@ -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}")

3
requirements.txt
View File

@@ -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