- 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
32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import os
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
@dataclass
|
|
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"))
|
|
|
|
def validate(self) -> list[str]:
|
|
errors = []
|
|
if not self.telegram_bot_token:
|
|
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
|
|
|
|
|
|
config = Config()
|