bot.py

#!/usr/bin/env python3
"""
AdoreBot - Instagram Automation Tool
=====================================

A Python-based Instagram bot that automatically engages with content from
a target account by liking posts, reels, and stories.

Features:
    - Automatic authentication with session persistence
    - Likes posts, reels, and stories from specified account
    - Smart filtering: only engages with content from current day
    - Human-like behavior with random delays between actions
    - Comprehensive logging and progress tracking
    - Rate limiting protection and error handling
    - Clean, modular architecture for easy maintenance

Usage:
    Configure your credentials in .env file, then run:
    $ python bot.py

Author: Kamil Supera
License: MIT
Repository: https://github.com/KamilSupera/AdoreBot

Warning:
    This tool violates Instagram's Terms of Service by using automation and
    unofficial APIs. Instagram explicitly prohibits bots and automated actions.
    
    Possible consequences:
    - Temporary action blocks (hours to days)
    - Account restrictions or temporary bans
    - Permanent account suspension
    - IP-based blocking
    
    Use at your own risk. For educational purposes only.
"""

import os
import sys
import time
import random
import logging
import warnings
from pathlib import Path
from typing import List, Dict
from datetime import datetime, timezone
from dotenv import load_dotenv
from instagrapi import Client
from instagrapi.exceptions import (
    LoginRequired,
    PleaseWaitFewMinutes,
    ChallengeRequired,
    ClientError
)

warnings.filterwarnings('ignore')

class CleanOutputHandler(logging.StreamHandler):
    def emit(self, record):
        msg = self.format(record)
        if any(x in msg for x in ['<script', '<html', 'data-sjs', '{"require"', 'data-content-len', 
                                    'qplTimings', 'ServerJS', 'data-bt']):
            return
        if len(msg) > 500:
            return
        super().emit(record)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('bot.log'),
        CleanOutputHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

logging.getLogger('instagrapi').setLevel(logging.CRITICAL)
logging.getLogger('instagrapi').propagate = False
for handler in logging.getLogger('instagrapi').handlers[:]:
    logging.getLogger('instagrapi').removeHandler(handler)


class InstagramBot:    
    def __init__(self):
        load_dotenv()
        
        self.username = os.getenv('INSTAGRAM_USERNAME')
        self.password = os.getenv('INSTAGRAM_PASSWORD')
        self.target_username = os.getenv('TARGET_USERNAME')
        self.min_delay = int(os.getenv('MIN_DELAY', '3'))
        self.max_delay = int(os.getenv('MAX_DELAY', '8'))
        self.max_posts = int(os.getenv('MAX_POSTS', '20'))
        self.max_clips = int(os.getenv('MAX_CLIPS', '20'))
        self.like_stories = os.getenv('LIKE_STORIES', 'true').lower() == 'true'
        self.characters_in_separator = 50
        self.instagram_path = "https://www.instagram.com"
        
        if not all([self.username, self.password, self.target_username]):
            logger.error("Missing credentials in .env file")
            sys.exit(1)
        
        self.client = Client()
        self.session_file = Path(f"{self.username}_session.json")
        self.today_date = datetime.now(timezone.utc).date()
        
    def login(self) -> bool:
        try:
            if self.session_file.exists():
                logger.info("Loading existing session...")
                self.client.load_settings(self.session_file)
                self.client.login(self.username, self.password)
                
                try:
                    self.client.get_timeline_feed()
                    logger.info("Session is valid!")
                    return True
                except LoginRequired:
                    logger.warning("Session expired, logging in again...")
                    self.session_file.unlink()
            
            logger.info("Logging in to Instagram...")
            self.client.login(self.username, self.password)
            self.client.dump_settings(self.session_file)
            logger.info("Login successful! Session saved.")
            return True
            
        except ChallengeRequired:
            logger.error("Challenge required! Instagram needs verification.")
            logger.error("Please login manually through the app/website first.")
            return False
            
        except PleaseWaitFewMinutes:
            logger.error("Rate limited! Please wait a few minutes before trying again.")
            return False
            
        except ClientError as e:
            logger.error(f"Login failed: {e}")
            return False
            
    def human_delay(self):
        delay = random.uniform(self.min_delay, self.max_delay)
        logger.info(f"Waiting {delay:.1f} seconds...")
        time.sleep(delay)
        
    def get_target_user_id(self) -> int:
        try:
            logger.info(f"Fetching user info for @{self.target_username}...")
            user_info = self.client.user_info_by_username(self.target_username)
            return user_info.pk
        except Exception as e:
            logger.error(f"Failed to get user info: {e}")
            sys.exit(1)
    
    def like_content(self, content_id: str, content_url: str, content_type: str = "media") -> bool:
        try:
            if content_type == "story":
                self.client.story_like(content_id)
            else:
                self.client.media_like(content_id)
            return True
        except PleaseWaitFewMinutes:
            logger.error("Rate limited! Stopping bot.")
            raise
        except Exception as e:
            error_msg = str(e).lower()
            if 'already' in error_msg or 'liked' in error_msg:
                return None
            logger.error(f"Failed to like {content_url}")
            return False
    
    def process_posts(self, user_id: int) -> Dict:
        stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        
        try:
            posts = self.client.user_medias(user_id, amount=self.max_posts)
            stats['total'] = len(posts)
            
            for i, post in enumerate(posts, 1):
                try:
                    post_url = f"{self.instagram_path}/p/{post.code}/"
                    post_date = post.taken_at.date()
                    
                    if post_date != self.today_date:
                        stats['too_old'] += 1
                        continue
                    
                    if post.has_liked:
                        stats['already_liked'] += 1
                        continue
                    
                    result = self.like_content(post.id, post_url, "media")
                    
                    if result is True:
                        stats['liked'] += 1
                    elif result is None:
                        stats['already_liked'] += 1
                    else:
                        stats['failed'] += 1
                    
                    if i < len(posts):
                        self.human_delay()
                        
                except PleaseWaitFewMinutes:
                    raise
                except Exception as e:
                    logger.warning(f"Error processing post {i}")
                    stats['failed'] += 1
                    continue
                    
        except PleaseWaitFewMinutes:
            raise
        except Exception as e:
            logger.error(f"Failed to fetch posts")
        
        logger.info(f"Posts summary: {stats['liked']} liked, {stats['already_liked']} already liked, {stats['too_old']} skipped (old)")
        return stats
    
    def process_reels(self, user_id: int) -> Dict:
        stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        
        try:
            reels = self.client.user_clips(user_id, amount=self.max_clips)
            stats['total'] = len(reels)
            
            for i, reel in enumerate(reels, 1):
                try:
                    reel_url = f"{self.instagram_path}/p/{reel.code}/"
                    reel_date = reel.taken_at.date()
                    
                    if reel_date != self.today_date:
                        stats['too_old'] += 1
                        continue
                    
                    if reel.has_liked:
                        stats['already_liked'] += 1
                        continue
                    
                    result = self.like_content(reel.id, reel_url, "media")
                    
                    if result is True:
                        stats['liked'] += 1
                    elif result is None:
                        stats['already_liked'] += 1
                    else:
                        stats['failed'] += 1
                    
                    if i < len(reels):
                        self.human_delay()
                        
                except PleaseWaitFewMinutes:
                    raise
                except Exception as e:
                    logger.warning(f"Error processing reel {i}")
                    stats['failed'] += 1
                    continue
                    
        except PleaseWaitFewMinutes:
            raise
        except Exception as e:
            logger.error(f"Failed to fetch reels")
        
        logger.info(f"Reels summary: {stats['liked']} liked, {stats['already_liked']} already liked, {stats['too_old']} skipped (old)")
        return stats
    
    def process_stories(self, user_id: int) -> Dict:
        stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        
        if not self.like_stories:
            logger.info("Story liking is disabled")
            return stats
        
        try:
            stories = self.client.user_stories(user_id)
            
            if not stories:
                logger.info("No active stories found")
                return stats
            
            stats['total'] = len(stories)
            
            for i, story in enumerate(stories, 1):
                try:
                    story_url = f"{self.instagram_path}/stories/{self.target_username}/{story.id}/"
                    
                    result = self.like_content(story.id, story_url, "story")
                    
                    if result is True:
                        stats['liked'] += 1
                    elif result is None:
                        stats['already_liked'] += 1
                    else:
                        stats['failed'] += 1
                    
                    if i < len(stories):
                        self.human_delay()
                        
                except PleaseWaitFewMinutes:
                    raise
                except Exception as e:
                    logger.warning(f"Error processing story {i}")
                    stats['failed'] += 1
                    continue
                    
        except PleaseWaitFewMinutes:
            raise
        except Exception as e:
            logger.error(f"Failed to fetch stories")
        
        if stats['total'] > 0:
            logger.info(f"Stories summary: {stats['liked']} liked, {stats['already_liked']} already liked")
        return stats
    
    def print_summary(self, posts_stats: Dict, reels_stats: Dict, stories_stats: Dict):
        logger.info("=" * self.characters_in_separator)
        logger.info("SUMMARY")
        logger.info("=" * self.characters_in_separator)
        
        total_found = posts_stats['total'] + reels_stats['total'] + stories_stats['total']
        total_liked = posts_stats['liked'] + reels_stats['liked'] + stories_stats['liked']
        total_already = posts_stats['already_liked'] + reels_stats['already_liked'] + stories_stats['already_liked']
        total_old = posts_stats['too_old'] + reels_stats['too_old'] + stories_stats['too_old']
        total_failed = posts_stats['failed'] + reels_stats['failed'] + stories_stats['failed']
        
        logger.info(f"Total items found: {total_found}")
        logger.info(f"  - Stories: {stories_stats['total']}")
        logger.info(f"  - Posts: {posts_stats['total']}")
        logger.info(f"  - Reels: {reels_stats['total']}")
        logger.info("")
        
        logger.info(f"Successfully liked: {total_liked}")
        logger.info(f"  - Stories: {stories_stats['liked']}")
        logger.info(f"  - Posts: {posts_stats['liked']}")
        logger.info(f"  - Reels: {reels_stats['liked']}")
        logger.info("")
        
        logger.info(f"Already liked: {total_already}")
        logger.info(f"  - Stories: {stories_stats['already_liked']}")
        logger.info(f"  - Posts: {posts_stats['already_liked']}")
        logger.info(f"  - Reels: {reels_stats['already_liked']}")
        logger.info("")
        
        if total_old > 0:
            logger.info(f"Not from today: {total_old}")
            logger.info(f"  - Posts: {posts_stats['too_old']}")
            logger.info(f"  - Reels: {reels_stats['too_old']}")
            logger.info("")
        
        if total_failed > 0:
            logger.info(f"Failed: {total_failed}")
            logger.info(f"  - Stories: {stories_stats['failed']}")
            logger.info(f"  - Posts: {posts_stats['failed']}")
            logger.info(f"  - Reels: {reels_stats['failed']}")
        
        logger.info("=" * self.characters_in_separator)
        
    def run(self):
        logger.info("=" * self.characters_in_separator)
        logger.info("Instagram Bot Started")
        logger.info("=" * self.characters_in_separator)
        logger.info(f"Your account: @{self.username}")
        logger.info(f"Target account: @{self.target_username}")
        logger.info(f"Today's date: {self.today_date}")
        logger.info(f"Max to check: {self.max_posts} posts, {self.max_clips} reels")
        stories_status = "enabled" if self.like_stories else "disabled"
        logger.info(f"Stories: {stories_status}")
        logger.info("=" * self.characters_in_separator)
        
        if not self.login():
            logger.error("Login failed. Exiting.")
            sys.exit(1)
        
        self.human_delay()
        
        user_id = self.get_target_user_id()
        self.human_delay()
        
        posts_stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        reels_stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        stories_stats = {'total': 0, 'liked': 0, 'already_liked': 0, 'too_old': 0, 'failed': 0}
        
        try:
            if self.like_stories:
                stories_stats = self.process_stories(user_id)
                self.human_delay()
            
            posts_stats = self.process_posts(user_id)
            self.human_delay()
            
            reels_stats = self.process_reels(user_id)
            
        except PleaseWaitFewMinutes:
            logger.error("Rate limited! Stopping early.")
        
        self.print_summary(posts_stats, reels_stats, stories_stats)
        logger.info("Bot finished!")


def main():
    try:
        bot = InstagramBot()
        bot.run()
    except KeyboardInterrupt:
        logger.info("\nBot stopped by user")
        sys.exit(0)
    except Exception as e:
        logger.error(f"Unexpected error: {e}", exc_info=True)
        sys.exit(1)


if __name__ == "__main__":
    main()