""" Bot Engine - Background Processing Loop Handles automatic DM and Comment reply cycles """ import time import random import logging import threading from datetime import datetime, timedelta from typing import Optional, Dict, List, Callable from config import BOT_CONFIG, DEFAULT_SYSTEM_PROMPT from instagram_handler import InstagramHandler from ai_engine import AIEngine logger = logging.getLogger("BotEngine") class BotEngine: """ Background bot that: - Checks for new DMs every 2-3 minutes - Checks for new comments on posts - Generates AI replies using Llama 3.1 - Sends replies with anti-ban delays - Tracks replied messages to avoid duplicates """ def __init__( self, ig_handler: InstagramHandler, ai_engine: AIEngine, log_callback: Optional[Callable] = None, ): self.ig = ig_handler self.ai = ai_engine self.log_callback = log_callback # Bot state self.is_running = False self.thread: Optional[threading.Thread] = None self.stop_event = threading.Event() # Settings self.auto_dm_enabled = False self.auto_comment_enabled = False self.system_prompt = "" # Tracking self.replied_dm_ids = set() self.replied_comment_ids = set() self.dm_conversation_context: Dict[str, List[Dict]] = {} # Stats self.total_dms_replied = 0 self.total_comments_replied = 0 self.last_check_time = None self.errors_count = 0 self.start_time = None # Rate limiting self.actions_this_hour = 0 self.hour_start = datetime.now() def _log(self, message: str, log_type: str = "info"): """Log message and call callback if set""" logger.info(f"[{log_type.upper()}] {message}") if self.log_callback: try: self.log_callback(message, log_type) except Exception: pass def _check_rate_limit(self) -> bool: """Check if we're within rate limits""" now = datetime.now() # Reset counter every hour if (now - self.hour_start).total_seconds() > 3600: self.actions_this_hour = 0 self.hour_start = now if self.actions_this_hour >= BOT_CONFIG["max_actions_per_hour"]: self._log( f"Rate limit reached ({self.actions_this_hour} actions this hour). Waiting...", "warning" ) return False return True def _smart_delay(self): """Wait between actions to avoid Instagram detection""" delay = random.uniform( BOT_CONFIG["action_delay_min"], BOT_CONFIG["action_delay_max"], ) self._log(f"Waiting {delay:.0f}s before next action...", "info") # Use stop_event to make delay interruptible self.stop_event.wait(timeout=delay) def _get_context_for_thread(self, thread_id: str) -> List[Dict]: """Get conversation context for a DM thread""" return self.dm_conversation_context.get(thread_id, []) def _save_context(self, thread_id: str, role: str, text: str): """Save message to conversation context""" if thread_id not in self.dm_conversation_context: self.dm_conversation_context[thread_id] = [] self.dm_conversation_context[thread_id].append({ "role": role, "text": text, }) # Keep only last N messages max_ctx = BOT_CONFIG["max_context_messages"] if len(self.dm_conversation_context[thread_id]) > max_ctx: self.dm_conversation_context[thread_id] = \ self.dm_conversation_context[thread_id][-max_ctx:] def _process_dms(self): """Check and reply to new DMs""" if not self.auto_dm_enabled: return if not self._check_rate_limit(): return try: self._log("Checking for new DMs... 💌", "info") # Fetch unread DMs unread = self.ig.get_unread_dms(count=BOT_CONFIG["max_dm_fetch"]) if not unread: self._log("No new DMs found", "info") return self._log(f"Found {len(unread)} new DMs!", "dm") for dm in unread: # Check if already replied msg_id = dm.get("message_id", "") if msg_id in self.replied_dm_ids: continue # Check stop event if self.stop_event.is_set(): return # Check rate limit if not self._check_rate_limit(): return thread_id = dm["thread_id"] sender = dm["sender_username"] sender_name = dm.get("sender_full_name", sender) message_text = dm["message_text"] self._log( f"📩 DM from @{sender}: \"{message_text[:50]}...\"", "dm" ) # Get conversation context context = self._get_context_for_thread(thread_id) # Save incoming message to context self._save_context(thread_id, "user", message_text) # Generate AI reply result = self.ai.generate_reply( message=message_text, system_prompt=self.system_prompt, context=context, message_type="dm", sender_name=sender_name, ) reply = result.get("reply", "") method = result.get("method", "unknown") if not reply: self._log(f"Skipped DM from @{sender} (no reply generated)", "warning") self.replied_dm_ids.add(msg_id) continue # Send reply self._log(f"Sending reply to @{sender}: \"{reply[:50]}...\" [{method}]", "dm") success = self.ig.send_dm(thread_id, reply) if success: self.replied_dm_ids.add(msg_id) self._save_context(thread_id, "assistant", reply) self.total_dms_replied += 1 self.actions_this_hour += 1 self._log( f"✅ Replied to @{sender} successfully! [{method}]", "success" ) else: self.errors_count += 1 self._log(f"❌ Failed to send DM to @{sender}", "error") # Delay between DM replies if not self.stop_event.is_set(): self._smart_delay() except Exception as e: self.errors_count += 1 self._log(f"Error processing DMs: {str(e)[:100]}", "error") def _process_comments(self): """Check and reply to new comments on posts""" if not self.auto_comment_enabled: return if not self._check_rate_limit(): return try: self._log("Checking for new comments... đŸ’Ŧ", "info") # Fetch unreplied comments comments = self.ig.get_recent_comments( count=BOT_CONFIG["max_comment_fetch"] ) if not comments: self._log("No new comments found", "info") return self._log(f"Found {len(comments)} unreplied comments!", "comment") for comment in comments: comment_id = comment.get("comment_id", "") # Check if already replied if comment_id in self.replied_comment_ids: continue # Check stop event if self.stop_event.is_set(): return # Check rate limit if not self._check_rate_limit(): return commenter = comment["commenter_username"] commenter_name = comment.get("commenter_full_name", commenter) comment_text = comment["comment_text"] media_id = comment["media_id"] self._log( f"đŸ’Ŧ Comment from @{commenter}: \"{comment_text[:50]}...\"", "comment" ) # Generate AI reply result = self.ai.generate_reply( message=comment_text, system_prompt=self.system_prompt, context=None, # No context for comments message_type="comment", sender_name=commenter_name, ) reply = result.get("reply", "") method = result.get("method", "unknown") if not reply: self._log( f"Skipped comment from @{commenter} (no reply generated)", "warning" ) self.replied_comment_ids.add(comment_id) continue # Add @mention to reply reply_with_mention = f"@{commenter} {reply}" # Truncate if too long for comment if len(reply_with_mention) > 200: reply_with_mention = reply_with_mention[:197] + "..." self._log( f"Replying to @{commenter}: \"{reply[:50]}...\" [{method}]", "comment" ) # Send reply success = self.ig.reply_to_comment(media_id, comment_id, reply_with_mention) if success: self.replied_comment_ids.add(comment_id) self.total_comments_replied += 1 self.actions_this_hour += 1 self._log( f"✅ Replied to @{commenter}'s comment! [{method}]", "success" ) else: self.errors_count += 1 self._log( f"❌ Failed to reply to @{commenter}'s comment", "error" ) # Delay between comment replies if not self.stop_event.is_set(): self._smart_delay() except Exception as e: self.errors_count += 1 self._log(f"Error processing comments: {str(e)[:100]}", "error") def _bot_loop(self): """Main bot loop - runs in background thread""" self._log("🚀 Bot loop started!", "bot") while not self.stop_event.is_set(): try: # Check if Instagram session is still valid if not self.ig.is_logged_in: self._log("❌ Instagram session expired! Stopping bot.", "error") self.is_running = False break # Process DMs if self.auto_dm_enabled and not self.stop_event.is_set(): self._process_dms() # Small delay between DM and Comment processing if not self.stop_event.is_set(): self.stop_event.wait(timeout=random.uniform(10, 20)) # Process Comments if self.auto_comment_enabled and not self.stop_event.is_set(): self._process_comments() # Update last check time self.last_check_time = datetime.now() # Wait before next check cycle (2-3 minutes) if not self.stop_event.is_set(): wait_time = random.uniform( BOT_CONFIG["check_interval_min"], BOT_CONFIG["check_interval_max"], ) self._log( f"💤 Next check in {wait_time:.0f} seconds...", "info" ) self.stop_event.wait(timeout=wait_time) except Exception as e: self.errors_count += 1 self._log(f"❌ Bot loop error: {str(e)[:150]}", "error") # Wait before retrying if not self.stop_event.is_set(): self.stop_event.wait(timeout=60) self._log("⏚ī¸ Bot loop stopped.", "bot") def start( self, auto_dm: bool = True, auto_comment: bool = True, system_prompt: str = "", ) -> Dict: """Start the bot in a background thread""" if self.is_running: return {"success": False, "message": "Bot is already running!"} if not self.ig.is_logged_in: return {"success": False, "message": "Not logged in to Instagram!"} if not auto_dm and not auto_comment: return { "success": False, "message": "Enable at least one feature (DM or Comment reply)!", } # Update settings self.auto_dm_enabled = auto_dm self.auto_comment_enabled = auto_comment self.system_prompt = system_prompt.strip() if system_prompt.strip() else "" # Reset stop event self.stop_event.clear() # Start background thread self.thread = threading.Thread(target=self._bot_loop, daemon=True) self.thread.start() self.is_running = True self.start_time = datetime.now() self.errors_count = 0 features = [] if auto_dm: features.append("DM Reply") if auto_comment: features.append("Comment Reply") prompt_info = "Custom prompt" if self.system_prompt else "Default prompt" self._log( f"Bot started! Features: {', '.join(features)} | {prompt_info}", "bot" ) return { "success": True, "message": f"Bot started! 🚀 ({', '.join(features)})", } def stop(self) -> Dict: """Stop the bot""" if not self.is_running: return {"success": False, "message": "Bot is not running!"} self._log("Stopping bot...", "bot") # Signal the thread to stop self.stop_event.set() # Wait for thread to finish (max 10 seconds) if self.thread and self.thread.is_alive(): self.thread.join(timeout=10) self.is_running = False self.start_time = None self._log("Bot stopped successfully! ⏚ī¸", "bot") return { "success": True, "message": "Bot stopped! ⏚ī¸", } def get_stats(self) -> Dict: """Get bot statistics""" uptime = None if self.start_time: uptime = datetime.now() - self.start_time return { "is_running": self.is_running, "auto_dm_enabled": self.auto_dm_enabled, "auto_comment_enabled": self.auto_comment_enabled, "total_dms_replied": self.total_dms_replied, "total_comments_replied": self.total_comments_replied, "errors_count": self.errors_count, "last_check_time": str(self.last_check_time) if self.last_check_time else None, "uptime_seconds": int(uptime.total_seconds()) if uptime else 0, "actions_this_hour": self.actions_this_hour, "replied_dm_count": len(self.replied_dm_ids), "replied_comment_count": len(self.replied_comment_ids), } def update_settings( self, auto_dm: Optional[bool] = None, auto_comment: Optional[bool] = None, system_prompt: Optional[str] = None, ): """Update bot settings while running""" if auto_dm is not None: self.auto_dm_enabled = auto_dm if auto_comment is not None: self.auto_comment_enabled = auto_comment if system_prompt is not None: self.system_prompt = system_prompt.strip() if system_prompt.strip() else "" self._log("Settings updated! ⚙ī¸", "info")