feat: parallel stories and posts fetching per followee

- Stories and posts fetched simultaneously via asyncio.gather
- Semaphore limits concurrent followee checks (default 5)
- CONCURRENT_CHECKS env var to tune parallelism
- 5-10x faster first run on large followee lists
This commit is contained in:
2026-06-17 12:29:34 +07:00
parent c786adc079
commit 915298e158
2 changed files with 41 additions and 19 deletions

View File

@@ -13,6 +13,7 @@ class Config:
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"))
def validate(self) -> list[str]:
errors = []

View File

@@ -1,3 +1,4 @@
import asyncio
import json
import random
import time
@@ -282,55 +283,75 @@ class InstagramMonitor:
followees = self._get_followees()
logger.info(f"Checking {len(followees)} followees")
for user in followees:
semaphore = asyncio.Semaphore(config.concurrent_checks)
async def process_followee(user: dict) -> list[dict]:
username = user["username"]
user_pk = user["pk"]
results = []
self._random_delay(2, 5)
async with semaphore:
self._random_delay(1, 3)
# Fetch stories and posts in parallel
stories_task = asyncio.to_thread(self._get_user_stories, user_pk)
posts_task = asyncio.to_thread(self._get_user_posts, user_pk)
stories, posts = await asyncio.gather(stories_task, posts_task)
try:
stories = self._get_user_stories(user_pk)
if stories:
logger.info(f"@{username}: {len(stories)} new stories")
for story in stories:
media_path = self._download_media(story, username, is_story=True)
media_path = await asyncio.to_thread(
self._download_media, story, username, True
)
if media_path:
media_id = str(story.get("pk", story.get("id", "")))
self._state[media_id] = datetime.utcnow().isoformat()
self._save_state()
yield {
results.append({
"type": "story",
"username": username,
"media_path": media_path,
"caption": "",
"shortcode": media_id,
}
self._random_delay(1, 2)
except Exception as e:
logger.error(f"Error processing stories for {username}: {e}")
})
await asyncio.to_thread(self._random_delay, 0.5, 1.5)
self._random_delay(2, 3)
if posts:
logger.info(f"@{username}: {len(posts)} new posts")
try:
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)
media_path = await asyncio.to_thread(
self._download_media, post, username
)
if media_path:
media_id = str(post.get("pk", post.get("id", "")))
self._state[media_id] = datetime.utcnow().isoformat()
self._save_state()
yield {
results.append({
"type": "post",
"username": username,
"media_path": media_path,
"caption": caption_text[:500],
"shortcode": media_id,
}
self._random_delay(2, 4)
except Exception as e:
logger.error(f"Error processing posts for {username}: {e}")
})
await asyncio.to_thread(self._random_delay, 1, 2)
return results
# Process all followees in parallel (with semaphore limit)
tasks = [process_followee(user) for user in followees]
done_results = await asyncio.gather(*tasks, return_exceptions=True)
for result in done_results:
if isinstance(result, Exception):
logger.error(f"Followee task failed: {result}")
continue
for item in result:
yield item
def cleanup_downloads(self, max_age_hours: int = 24) -> None:
now = datetime.utcnow()