Delete auto_responder.py

auto_responder.py

@@ -1,195 +0,0 @@
-from transformers import BlenderbotTokenizer, BlenderbotForConditionalGeneration
-import tweepy
-import json
-from typing import Dict, Optional
-import time
-import logging
-
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
-logger = logging.getLogger(__name__)
-
-class TwitterBot:
-    def __init__(self, config_path: str):
-        """
-        Initialize the Twitter bot with API credentials
-        """
-        # Load configuration
-        self.config = self._load_config(config_path)
-        if not self._validate_config():
-            raise ValueError("Invalid or missing Twitter credentials")
-        
-        # Initialize AI model
-        logger.info("Initializing AI model...")
-        self.model_name = "facebook/blenderbot-400M-distill"
-        self.tokenizer = BlenderbotTokenizer.from_pretrained(self.model_name)
-        self.model = BlenderbotForConditionalGeneration.from_pretrained(self.model_name)
-        logger.info("AI model initialized successfully")
-        
-        # Initialize Twitter client
-        self._init_twitter()
-        
-        # Store conversation history
-        self.conversation_history: Dict[str, list] = {}
-
-    def _load_config(self, config_path: str) -> dict:
-        """Load configuration from JSON file"""
-        try:
-            with open(config_path, 'r') as f:
-                config = json.load(f)
-            logger.info(f"Configuration loaded from {config_path}")
-            return config
-        except Exception as e:
-            logger.error(f"Error loading config: {e}")
-            return {}
-
-    def _validate_config(self) -> bool:
-        """Validate Twitter credentials exist"""
-        required_keys = ['consumer_key', 'consumer_secret', 'access_token', 'access_token_secret']
-        
-        if 'twitter' not in self.config:
-            logger.error("No Twitter configuration found")
-            return False
-            
-        for key in required_keys:
-            if key not in self.config['twitter']:
-                logger.error(f"Missing Twitter credential: {key}")
-                return False
-            if not self.config['twitter'][key]:
-                logger.error(f"Empty Twitter credential: {key}")
-                return False
-        
-        logger.info("Twitter configuration validated successfully")
-        return True
-
-    def _init_twitter(self):
-        """Initialize Twitter client"""
-        try:
-            logger.info("Initializing Twitter client...")
-            
-            # Initialize with both OAuth 1.0a and OAuth 2.0
-            self.twitter_client = tweepy.Client(
-                consumer_key=self.config['twitter']['consumer_key'],
-                consumer_secret=self.config['twitter']['consumer_secret'],
-                access_token=self.config['twitter']['access_token'],
-                access_token_secret=self.config['twitter']['access_token_secret'],
-                wait_on_rate_limit=True
-            )
-            
-            # Test the connection with a simple API call
-            try:
-                user = self.twitter_client.get_me()
-                logger.info(f"Twitter credentials verified successfully for user ID: {user.data.id}")
-            except Exception as e:
-                logger.error(f"Error verifying credentials: {e}")
-                raise
-            
-        except Exception as e:
-            logger.error(f"Error initializing Twitter client: {str(e)}")
-            self.twitter_client = None
-            raise
-
-    def generate_response(self, message: str, sender_id: str) -> str:
-        """
-        Generate an AI response to the incoming message
-        """
-        if sender_id not in self.conversation_history:
-            self.conversation_history[sender_id] = []
-        
-        self.conversation_history[sender_id].append(message)
-
-        try:
-            # Generate AI response
-            inputs = self.tokenizer(message, return_tensors="pt", truncation=True, max_length=512)
-            reply_ids = self.model.generate(
-                inputs["input_ids"],
-                max_length=128,
-                num_beams=4,
-                no_repeat_ngram_size=3,
-                temperature=0.7,
-                do_sample=True
-            )
-            response = self.tokenizer.decode(reply_ids[0], skip_special_tokens=True)
-            
-            # Store response in history
-            self.conversation_history[sender_id].append(response)
-            
-            # Ensure response is not too long for Twitter
-            if len(response) > 280:
-                response = response[:277] + "..."
-            
-            return response
-        except Exception as e:
-            logger.error(f"Error generating response: {e}")
-            return "عذراً، حدث خطأ في معالجة رسالتك"
-
-    def respond_to_mentions(self, since_id: Optional[int] = None) -> int:
-        """
-        Respond to Twitter mentions
-        Returns the latest tweet ID processed
-        """
-        if not self.twitter_client:
-            logger.error("Twitter client not initialized")
-            return since_id or 0
-
-        try:
-            logger.info(f"Checking for mentions since ID: {since_id}")
-            # Get mentions using the v2 API
-            mentions = self.twitter_client.get_users_mentions(
-                self.twitter_client.get_me().data.id,
-                since_id=since_id
-            )
-            latest_id = since_id or 0
-
-            if mentions.data:
-                for mention in mentions.data:
-                    if mention.id > latest_id:
-                        latest_id = mention.id
-
-                    logger.info(f"Processing mention ID: {mention.id}")
-                    response = self.generate_response(
-                        mention.text,
-                        str(mention.author_id)
-                    )
-
-                    # Reply to the tweet using v2 API
-                    self.twitter_client.create_tweet(
-                        text=response,
-                        in_reply_to_tweet_id=mention.id
-                    )
-                    logger.info(f"Replied to tweet {mention.id}")
-
-            return latest_id
-        except Exception as e:
-            logger.error(f"Error processing Twitter mentions: {e}")
-            return since_id or 0
-
-    def run(self, interval: int = 60):
-        """
-        Run the bot continuously
-        """
-        last_mention_id = None
-        logger.info("Starting Twitter bot...")
-
-        while True:
-            try:
-                # Process Twitter mentions
-                last_mention_id = self.respond_to_mentions(last_mention_id)
-                logger.info(f"Checked for new mentions. Last ID: {last_mention_id}")
-                
-                # Wait before next check
-                time.sleep(interval)
-            except Exception as e:
-                logger.error(f"Error in main loop: {e}")
-                time.sleep(interval)
-
-if __name__ == "__main__":
-    try:
-        # Create and run the Twitter bot
-        bot = TwitterBot("config.json")
-        bot.run()
-    except Exception as e:
-        logger.error(f"Fatal error: {e}")
-        raise