feat: Telegram bot for Instagram monitoring via instaloader
- aiogram 3 bot with commands /start /check /status /stop - InstagramMonitor: periodic check of followees stories and posts - Docker deployment with docker-compose - State tracking to avoid duplicate sends - Auto-cleanup of old media files
This commit is contained in:
11
.env.example
Normal file
11
.env.example
Normal file
@@ -0,0 +1,11 @@
|
||||
# Telegram
|
||||
TELEGRAM_BOT_TOKEN=your_bot_token_here
|
||||
TELEGRAM_CHAT_ID=your_telegram_user_id_here
|
||||
|
||||
# Instagram
|
||||
INSTAGRAM_USERNAME=your_instagram_username
|
||||
INSTAGRAM_PASSWORD=your_instagram_password
|
||||
|
||||
# Settings
|
||||
CHECK_INTERVAL_MINUTES=15
|
||||
MAX_MEDIA_SIZE_MB=50
|
||||
7
.gitignore
vendored
Normal file
7
.gitignore
vendored
Normal file
@@ -0,0 +1,7 @@
|
||||
data/
|
||||
.env
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
.venv/
|
||||
venv/
|
||||
16
Dockerfile
Normal file
16
Dockerfile
Normal file
@@ -0,0 +1,16 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
RUN apt-get update && \
|
||||
apt-get install -y --no-install-recommends ffmpeg && \
|
||||
rm -rf /var/lib/apt/lists/*
|
||||
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
RUN mkdir -p /data/downloads /data/session
|
||||
|
||||
CMD ["python", "bot.py"]
|
||||
133
bot.py
Normal file
133
bot.py
Normal file
@@ -0,0 +1,133 @@
|
||||
import asyncio
|
||||
import os
|
||||
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
|
||||
|
||||
from config import config
|
||||
from instagram_monitor import InstagramMonitor
|
||||
|
||||
router = Router()
|
||||
bot = Bot(token=config.telegram_bot_token)
|
||||
monitor = InstagramMonitor()
|
||||
|
||||
|
||||
@router.message(CommandStart())
|
||||
async def cmd_start(message: Message) -> None:
|
||||
if message.from_user and message.from_user.id == config.telegram_chat_id:
|
||||
await message.answer(
|
||||
"Instagram Monitor Bot запущен.\n"
|
||||
f"Проверяю подписки каждые {config.check_interval_minutes} минут.\n"
|
||||
"Команды:\n"
|
||||
"/check — проверить сейчас\n"
|
||||
"/status — статус бота\n"
|
||||
"/stop — остановить мониторинг"
|
||||
)
|
||||
else:
|
||||
await message.answer("Этот бот не для тебя.")
|
||||
|
||||
|
||||
@router.message(F.text == "/status")
|
||||
async def cmd_status(message: Message) -> None:
|
||||
if message.from_user and message.from_user.id == config.telegram_chat_id:
|
||||
state_file = Path(config.download_dir) / "state.json"
|
||||
tracked = 0
|
||||
if state_file.exists():
|
||||
import json
|
||||
with open(state_file) as f:
|
||||
tracked = len(json.load(f))
|
||||
await message.answer(
|
||||
f"Отслежено медиа: {tracked}\n"
|
||||
f"Интервал: {config.check_interval_minutes} мин\n"
|
||||
f"Instagram: {config.instagram_username}"
|
||||
)
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
@router.message(F.text == "/check")
|
||||
async def cmd_check(message: Message) -> None:
|
||||
if message.from_user and message.from_user.id == config.telegram_chat_id:
|
||||
await message.answer("Проверяю обновления...")
|
||||
count = 0
|
||||
async for update in monitor.check_updates():
|
||||
await send_media(update)
|
||||
count += 1
|
||||
await message.answer(f"Готово. Найдено: {count}")
|
||||
|
||||
|
||||
async def send_media(update: dict) -> None:
|
||||
chat_id = config.telegram_chat_id
|
||||
media_path = update["media_path"]
|
||||
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
|
||||
|
||||
path = Path(media_path)
|
||||
if not path.exists():
|
||||
logger.warning(f"Media file not found: {media_path}")
|
||||
return
|
||||
|
||||
try:
|
||||
if path.suffix == ".mp4":
|
||||
video = FSInputFile(str(path))
|
||||
await bot.send_video(chat_id=chat_id, video=video, caption=text, parse_mode=None)
|
||||
elif path.suffix in (".jpg", ".jpeg", ".png", ".webp"):
|
||||
photo = FSInputFile(str(path))
|
||||
await bot.send_photo(chat_id=chat_id, photo=photo, caption=text, parse_mode=None)
|
||||
else:
|
||||
document = FSInputFile(str(path))
|
||||
await bot.send_document(chat_id=chat_id, document=document, caption=text, parse_mode=None)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send media: {e}")
|
||||
await bot.send_message(chat_id=chat_id, text=f"{header}\n\n{caption}\n\n[Файл не удалось отправить]")
|
||||
|
||||
|
||||
async def monitor_loop() -> None:
|
||||
logger.info("Starting monitor loop")
|
||||
while True:
|
||||
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)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
errors = config.validate()
|
||||
if errors:
|
||||
for e in errors:
|
||||
logger.error(e)
|
||||
return
|
||||
|
||||
dp = Dispatcher()
|
||||
dp.include_router(router)
|
||||
|
||||
asyncio.create_task(monitor_loop())
|
||||
|
||||
logger.info("Bot started")
|
||||
await dp.start_polling(bot)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
asyncio.run(main())
|
||||
29
config.py
Normal file
29
config.py
Normal file
@@ -0,0 +1,29 @@
|
||||
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"))
|
||||
|
||||
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()
|
||||
16
docker-compose.yml
Normal file
16
docker-compose.yml
Normal file
@@ -0,0 +1,16 @@
|
||||
services:
|
||||
ig-bot:
|
||||
build: .
|
||||
container_name: ig-telegram-bot
|
||||
restart: unless-stopped
|
||||
env_file: .env
|
||||
volumes:
|
||||
- ./data:/data
|
||||
environment:
|
||||
- DOWNLOAD_DIR=/data/downloads
|
||||
- SESSION_FILE=/data/session/session-ig
|
||||
logging:
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
172
instagram_monitor.py
Normal file
172
instagram_monitor.py
Normal file
@@ -0,0 +1,172 @@
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import AsyncIterator
|
||||
|
||||
import instaloader
|
||||
from loguru import logger
|
||||
|
||||
from config import config
|
||||
|
||||
|
||||
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=30,
|
||||
)
|
||||
self._session_path = Path(config.session_file)
|
||||
self._download_dir = Path(config.download_dir)
|
||||
self._state_file = self._download_dir / "state.json"
|
||||
self._download_dir.mkdir(parents=True, exist_ok=True)
|
||||
self._state: dict[str, str] = self._load_state()
|
||||
|
||||
def _load_state(self) -> dict[str, str]:
|
||||
if self._state_file.exists():
|
||||
with open(self._state_file) as f:
|
||||
return json.load(f)
|
||||
return {}
|
||||
|
||||
def _save_state(self) -> None:
|
||||
with open(self._state_file, "w") as f:
|
||||
json.dump(self._state, f, indent=2)
|
||||
|
||||
def login(self) -> bool:
|
||||
try:
|
||||
if self._session_path.exists():
|
||||
self.loader.load_session_from_file(config.instagram_username, str(self._session_path))
|
||||
logger.info("Session loaded from file")
|
||||
return True
|
||||
|
||||
self.loader.login(config.instagram_username, config.instagram_password)
|
||||
self.loader.save_session_to_file(str(self._session_path))
|
||||
logger.info("Logged in and session saved")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"Login failed: {e}")
|
||||
return False
|
||||
|
||||
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 []
|
||||
|
||||
def _get_profile_stories(self, username: str) -> list[instaloader.StoryItem]:
|
||||
try:
|
||||
profile = instaloader.Profile.from_username(self.loader.context, username)
|
||||
stories = []
|
||||
for story in self.loader.get_stories(userids=[profile.userid]):
|
||||
if str(story.mediaid) in self._state:
|
||||
continue
|
||||
stories.append(story)
|
||||
return stories
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching stories for {username}: {e}")
|
||||
return []
|
||||
|
||||
def _download_media(self, item: instaloader.Post | instaloader.StoryItem) -> str | None:
|
||||
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"
|
||||
|
||||
if typename == "StoryItem":
|
||||
target_dir = self._download_dir / "stories" / item.owner_username
|
||||
else:
|
||||
target_dir = self._download_dir / "posts" / item.owner_username
|
||||
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if typename == "Post":
|
||||
self.loader.download_post(item, target=str(target_dir))
|
||||
else:
|
||||
self.loader.download_storyitem(item, target=str(target_dir))
|
||||
|
||||
for ext in [".jpg", ".mp4", ".webp"]:
|
||||
media_file = target_dir / f"{shortcode}{ext}"
|
||||
if media_file.exists():
|
||||
return str(media_file)
|
||||
|
||||
for f in target_dir.iterdir():
|
||||
if f.suffix in (".jpg", ".mp4", ".webp") and shortcode in f.name:
|
||||
return str(f)
|
||||
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error downloading media {getattr(item, 'shortcode', '?')}: {e}")
|
||||
return None
|
||||
|
||||
async def check_updates(self) -> AsyncIterator[dict]:
|
||||
if not self.login():
|
||||
logger.error("Cannot check updates without login")
|
||||
return
|
||||
|
||||
try:
|
||||
profile = instaloader.Profile.from_username(self.loader.context, config.instagram_username)
|
||||
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 followee in followees:
|
||||
username = followee.username
|
||||
|
||||
for story in self._get_profile_stories(username):
|
||||
media_path = self._download_media(story)
|
||||
if media_path:
|
||||
self._state[str(story.mediaid)] = datetime.utcnow().isoformat()
|
||||
self._save_state()
|
||||
yield {
|
||||
"type": "story",
|
||||
"username": username,
|
||||
"media_path": media_path,
|
||||
"caption": "",
|
||||
"shortcode": str(story.mediaid),
|
||||
}
|
||||
|
||||
for post in self._get_profile_posts(username):
|
||||
caption = post.caption or ""
|
||||
media_path = self._download_media(post)
|
||||
if media_path:
|
||||
self._state[str(post.mediaid)] = datetime.utcnow().isoformat()
|
||||
self._save_state()
|
||||
yield {
|
||||
"type": "post",
|
||||
"username": username,
|
||||
"media_path": media_path,
|
||||
"caption": caption[:500],
|
||||
"shortcode": post.shortcode,
|
||||
}
|
||||
|
||||
def cleanup_downloads(self, max_age_hours: int = 24) -> None:
|
||||
now = datetime.utcnow()
|
||||
for subdir in ["stories", "posts"]:
|
||||
dir_path = self._download_dir / subdir
|
||||
if not dir_path.exists():
|
||||
continue
|
||||
for profile_dir in dir_path.iterdir():
|
||||
if not profile_dir.is_dir():
|
||||
continue
|
||||
for f in profile_dir.iterdir():
|
||||
if f.is_file():
|
||||
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}")
|
||||
5
requirements.txt
Normal file
5
requirements.txt
Normal file
@@ -0,0 +1,5 @@
|
||||
aiogram>=3.12,<4.0
|
||||
instaloader>=4.15,<5.0
|
||||
loguru>=0.7,<1.0
|
||||
aiofiles>=24.1,<25.0
|
||||
python-dotenv>=1.0,<2.0
|
||||
Reference in New Issue
Block a user