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

refactor: full async, rate limit backoff, graceful shutdown, health endpoint

Browse Source 
- Switch from requests to httpx.AsyncClient (full async HTTP)
- Add _request() with exponential backoff on 429/401
- Cache followees with 5-min TTL to reduce API calls
- Remove dead code (_get_all_stories, instagram_password, session_file)
- Graceful shutdown via asyncio.Event + SIGTERM/SIGINT handlers
- Add /health command with connection status
- Stream media downloads with httpx + aiofiles
- Track story expiring_at for smart cleanup
- Sanitize .env.example (remove real tokens)
This commit is contained in:
smolkik-code
2026-06-17 16:03:47 +07:00
parent da64550426
commit 330e48beaa
6 changed files with 213 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.env.example
View File

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

1
README.md
View File

@@ -74,6 +74,7 @@ docker compose up -d
| `/start` | Информация о боте |
| `/check` | Проверить обновления сейчас |
| `/status` | Статистика базы |
| `/health` | Статус подключений |
| `/stop` | Остановить мониторинг |
## API

73
bot.py
View File

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

8
config.py
View File

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

280
instagram_monitor.py
View File

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

2
requirements.txt
View File

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