808 lines
34 KiB
Python
808 lines
34 KiB
Python
import os
|
|
import time
|
|
import json
|
|
import logging
|
|
import requests
|
|
import schedule
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
import colorama
|
|
from colorama import Fore, Back, Style
|
|
|
|
# Initialize colorama for Windows
|
|
colorama.init()
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Configure logging with colors and terminal output
|
|
class ColoredFormatter(logging.Formatter):
|
|
"""Custom formatter with colors for different log levels"""
|
|
|
|
COLORS = {
|
|
'DEBUG': Fore.CYAN,
|
|
'INFO': Fore.GREEN,
|
|
'WARNING': Fore.YELLOW,
|
|
'ERROR': Fore.RED,
|
|
'CRITICAL': Fore.RED + Back.WHITE,
|
|
}
|
|
|
|
def format(self, record):
|
|
# Add color to the level name
|
|
levelname = record.levelname
|
|
if levelname in self.COLORS:
|
|
record.levelname = f"{self.COLORS[levelname]}{levelname}{Style.RESET_ALL}"
|
|
|
|
# Add color to the message based on content
|
|
if 'SENT' in record.getMessage():
|
|
record.msg = f"{Fore.BLUE}{record.msg}{Style.RESET_ALL}"
|
|
elif 'RECEIVED' in record.getMessage():
|
|
record.msg = f"{Fore.MAGENTA}{record.msg}{Style.RESET_ALL}"
|
|
elif 'AI GENERATED' in record.getMessage():
|
|
record.msg = f"{Fore.CYAN}{record.msg}{Style.RESET_ALL}"
|
|
elif 'ERROR' in record.getMessage():
|
|
record.msg = f"{Fore.RED}{record.msg}{Style.RESET_ALL}"
|
|
elif 'SUCCESS' in record.getMessage():
|
|
record.msg = f"{Fore.GREEN}{record.msg}{Style.RESET_ALL}"
|
|
|
|
return super().format(record)
|
|
|
|
# Set up logging
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
# Create console handler with colored formatter
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.INFO)
|
|
formatter = ColoredFormatter(
|
|
'%(asctime)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%dT%H:%M:%S'
|
|
)
|
|
console_handler.setFormatter(formatter)
|
|
logger.addHandler(console_handler)
|
|
|
|
# Create file handler for persistent logs
|
|
file_handler = logging.FileHandler('responder_bot.log')
|
|
file_handler.setLevel(logging.INFO)
|
|
file_formatter = logging.Formatter(
|
|
'%(asctime)s - %(levelname)s - %(message)s',
|
|
datefmt='%Y-%m-%dT%H:%M:%S'
|
|
)
|
|
file_handler.setFormatter(file_formatter)
|
|
logger.addHandler(file_handler)
|
|
|
|
class EnhancedMeetMeClient:
|
|
"""Enhanced client for interacting with MeetMe API (local or production)"""
|
|
|
|
def __init__(self):
|
|
self.session = requests.Session()
|
|
self.api_base_url = os.getenv('API_BASE_URL', 'https://api.meetme.com')
|
|
self.use_local_api = os.getenv('USE_LOCAL_API', 'false').lower() == 'true'
|
|
|
|
# Validate required environment variables
|
|
self._validate_environment()
|
|
|
|
# Configure headers based on API type
|
|
if self.use_local_api:
|
|
self.headers = {
|
|
'User-Agent': 'MeetMe-Bot/1.0',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'
|
|
}
|
|
else:
|
|
self.headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'
|
|
}
|
|
|
|
self.session.headers.update(self.headers)
|
|
self.auth_token = None
|
|
self.my_user_id = None
|
|
|
|
# Statistics tracking
|
|
self.stats = {
|
|
'messages_sent': 0,
|
|
'messages_received': 0,
|
|
'conversations_active': 0,
|
|
'errors': 0,
|
|
'start_time': datetime.now()
|
|
}
|
|
|
|
def _validate_environment(self):
|
|
"""Validate required environment variables"""
|
|
if self.use_local_api:
|
|
if not os.getenv('API_USERNAME'):
|
|
raise ValueError("API_USERNAME must be set for local API")
|
|
if not os.getenv('API_PASSWORD'):
|
|
raise ValueError("API_PASSWORD must be set for local API")
|
|
else:
|
|
if not os.getenv('MM_USERNAME'):
|
|
raise ValueError("MM_USERNAME must be set for production API")
|
|
if not os.getenv('MM_PASSWORD'):
|
|
raise ValueError("MM_PASSWORD must be set for production API")
|
|
|
|
if not os.getenv('OLLAMA_ENDPOINT'):
|
|
raise ValueError("OLLAMA_ENDPOINT must be set")
|
|
|
|
def get_stats(self):
|
|
"""Get current statistics"""
|
|
runtime = datetime.now() - self.stats['start_time']
|
|
return {
|
|
**self.stats,
|
|
'runtime': str(runtime),
|
|
'messages_per_hour': self.stats['messages_sent'] / max(runtime.total_seconds() / 3600, 1)
|
|
}
|
|
|
|
def login(self):
|
|
"""Authenticate with MeetMe API (local or production)"""
|
|
try:
|
|
logger.info(f"{Fore.YELLOW}[LOGIN] ATTEMPTING LOGIN to {'local' if self.use_local_api else 'production'} API...{Style.RESET_ALL}")
|
|
|
|
if self.use_local_api:
|
|
return self._login_local()
|
|
else:
|
|
return self._login_production()
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] LOGIN ERROR: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def _login_local(self):
|
|
"""Login to local Node.js backend"""
|
|
try:
|
|
username = os.getenv('API_USERNAME')
|
|
password = os.getenv('API_PASSWORD')
|
|
|
|
login_data = {
|
|
'email': username,
|
|
'password': password
|
|
}
|
|
|
|
response = self.session.post(
|
|
f"{self.api_base_url}/auth/login",
|
|
json=login_data,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.auth_token = data.get('token')
|
|
self.my_user_id = data.get('user_id') or data.get('user', {}).get('_id')
|
|
if self.auth_token:
|
|
self.session.headers.update({'Authorization': f'Bearer {self.auth_token}'})
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] LOGIN SUCCESS - Local API as user {self.my_user_id}{Style.RESET_ALL}")
|
|
return True
|
|
else:
|
|
logger.error(f"[ERROR] LOGIN FAILED - Status {response.status_code}: {response.text}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] LOGIN TIMEOUT - Server not responding")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] LOGIN ERROR: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def _login_production(self):
|
|
"""Login to production MeetMe API"""
|
|
try:
|
|
username = os.getenv('MM_USERNAME')
|
|
password = os.getenv('MM_PASSWORD')
|
|
|
|
login_data = {
|
|
'email': username,
|
|
'password': password
|
|
}
|
|
|
|
response = self.session.post(
|
|
f"{self.api_base_url}/auth/login",
|
|
json=login_data,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.my_user_id = data.get('user_id') or data.get('user', {}).get('id')
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] LOGIN SUCCESS - Production API as user {self.my_user_id}{Style.RESET_ALL}")
|
|
return True
|
|
else:
|
|
logger.error(f"[ERROR] LOGIN FAILED - Status {response.status_code}: {response.text}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] LOGIN TIMEOUT - Server not responding")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] LOGIN ERROR: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def get_conversations(self):
|
|
"""Get list of active conversations"""
|
|
try:
|
|
logger.info(f"{Fore.CYAN}[SEARCHING] CHECKING for active conversations...{Style.RESET_ALL}")
|
|
|
|
if self.use_local_api:
|
|
return self._get_conversations_local()
|
|
else:
|
|
return self._get_conversations_production()
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting conversations: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def _get_conversations_local(self):
|
|
"""Get conversations from local API"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{self.api_base_url}/conversations",
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
conversations = []
|
|
for conv in data.get('conversations', []):
|
|
conv_id = conv.get('_id') or conv.get('id')
|
|
if conv_id: # Only add valid conversations
|
|
conversations.append({
|
|
'id': conv_id,
|
|
'display_name': conv.get('display_name') or conv.get('name', 'Unknown'),
|
|
'my_user_id': self.my_user_id
|
|
})
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] FOUND {len(conversations)} active conversations{Style.RESET_ALL}")
|
|
return conversations
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to get conversations from local API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT getting conversations from local API")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting conversations from local API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def _get_conversations_production(self):
|
|
"""Get conversations from production API"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{self.api_base_url}/conversations",
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
conversations = []
|
|
for conv in data.get('conversations', []):
|
|
conv_id = conv.get('id') or conv.get('_id')
|
|
if conv_id: # Only add valid conversations
|
|
conversations.append({
|
|
'id': conv_id,
|
|
'display_name': conv.get('display_name') or conv.get('name', 'Unknown'),
|
|
'my_user_id': self.my_user_id
|
|
})
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] FOUND {len(conversations)} active conversations{Style.RESET_ALL}")
|
|
return conversations
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to get conversations from production API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT getting conversations from production API")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting conversations from production API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def get_messages(self, conversation_id, page=1):
|
|
"""Get messages for a specific conversation"""
|
|
try:
|
|
if self.use_local_api:
|
|
return self._get_messages_local(conversation_id, page)
|
|
else:
|
|
return self._get_messages_production(conversation_id, page)
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting messages: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def _get_messages_local(self, conversation_id, page=1):
|
|
"""Get messages from local API"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{self.api_base_url}/conversations/{conversation_id}/messages",
|
|
params={'page': page, 'limit': 10},
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
messages = []
|
|
for msg in data.get('messages', []):
|
|
# Handle different field names between APIs
|
|
msg_id = msg.get('_id') or msg.get('id')
|
|
sender_id = msg.get('sender_id')
|
|
text = msg.get('message') or msg.get('text') or msg.get('content', '')
|
|
timestamp = msg.get('timestamp') or msg.get('created_at') or msg.get('date')
|
|
|
|
if msg_id and sender_id and text and timestamp: # Only add valid messages
|
|
messages.append({
|
|
'id': msg_id,
|
|
'sender_id': sender_id,
|
|
'text': text,
|
|
'timestamp': timestamp
|
|
})
|
|
logger.debug(f"Retrieved {len(messages)} messages for conversation {conversation_id} from local API")
|
|
return messages
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to get messages from local API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT getting messages from local API")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting messages from local API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def _get_messages_production(self, conversation_id, page=1):
|
|
"""Get messages from production API"""
|
|
try:
|
|
response = self.session.get(
|
|
f"{self.api_base_url}/conversations/{conversation_id}/messages",
|
|
params={'page': page, 'limit': 10},
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
messages = []
|
|
for msg in data.get('messages', []):
|
|
# Handle different field names between APIs
|
|
msg_id = msg.get('id') or msg.get('_id')
|
|
sender_id = msg.get('sender_id')
|
|
text = msg.get('text') or msg.get('message') or msg.get('content', '')
|
|
timestamp = msg.get('timestamp') or msg.get('created_at') or msg.get('date')
|
|
|
|
if msg_id and sender_id and text and timestamp: # Only add valid messages
|
|
messages.append({
|
|
'id': msg_id,
|
|
'sender_id': sender_id,
|
|
'text': text,
|
|
'timestamp': timestamp
|
|
})
|
|
logger.debug(f"Retrieved {len(messages)} messages for conversation {conversation_id} from production API")
|
|
return messages
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to get messages from production API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT getting messages from production API")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR getting messages from production API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return []
|
|
|
|
def send_message(self, conversation_id, text):
|
|
"""Send message to a specific conversation"""
|
|
try:
|
|
logger.info(f"{Fore.BLUE}[SENDING] SENDING REPLY to conversation {conversation_id}...{Style.RESET_ALL}")
|
|
logger.info(f"{Fore.CYAN}[MESSAGE] REPLY CONTENT: {text[:100]}{'...' if len(text) > 100 else ''}{Style.RESET_ALL}")
|
|
|
|
if self.use_local_api:
|
|
return self._send_message_local(conversation_id, text)
|
|
else:
|
|
return self._send_message_production(conversation_id, text)
|
|
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR sending message: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def _send_message_local(self, conversation_id, text):
|
|
"""Send message via local API"""
|
|
try:
|
|
message_data = {
|
|
'conversation_id': conversation_id,
|
|
'message': text,
|
|
'sender_id': self.my_user_id,
|
|
'timestamp': datetime.now().isoformat()
|
|
}
|
|
|
|
response = self.session.post(
|
|
f"{self.api_base_url}/messages",
|
|
json=message_data,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] REPLY SENT SUCCESSFULLY to conversation {conversation_id} via local API{Style.RESET_ALL}")
|
|
self.stats['messages_sent'] += 1
|
|
return True
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to send message via local API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT sending message via local API")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR sending message via local API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def _send_message_production(self, conversation_id, text):
|
|
"""Send message via production API"""
|
|
try:
|
|
message_data = {
|
|
'conversation_id': conversation_id,
|
|
'message': text
|
|
}
|
|
|
|
response = self.session.post(
|
|
f"{self.api_base_url}/messages/send",
|
|
json=message_data,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] REPLY SENT SUCCESSFULLY to conversation {conversation_id} via production API{Style.RESET_ALL}")
|
|
self.stats['messages_sent'] += 1
|
|
return True
|
|
else:
|
|
logger.error(f"[ERROR] FAILED to send message via production API: {response.status_code}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT sending message via production API")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR sending message via production API: {str(e)}")
|
|
self.stats['errors'] += 1
|
|
return False
|
|
|
|
def generate_response(incoming_text):
|
|
"""Generate AI response using Ollama"""
|
|
try:
|
|
my_name = os.getenv('MY_NAME', 'DrJones')
|
|
my_profile = os.getenv('MY_PROFILE', 'water-tech geek, coffee addict')
|
|
ollama_endpoint = os.getenv('OLLAMA_ENDPOINT')
|
|
ollama_model = os.getenv('OLLAMA_MODEL', 'quen3')
|
|
|
|
if not ollama_endpoint:
|
|
logger.error("[ERROR] OLLAMA_ENDPOINT not configured")
|
|
return None
|
|
|
|
logger.info(f"{Fore.YELLOW}[AUTO RESPONDER] AI GENERATING response to: \"{incoming_text[:50]}{'...' if len(incoming_text) > 50 else ''}\"{Style.RESET_ALL}")
|
|
|
|
# Build prompt with context
|
|
prompt = f"""You are {my_name}, a {my_profile}.
|
|
|
|
Someone just sent you this message: "{incoming_text}"
|
|
|
|
Generate a friendly, conversational response that:
|
|
- Acknowledges their message appropriately
|
|
- Shows genuine interest in the conversation
|
|
- Matches your personality as a {my_profile}
|
|
- Is natural and not overly formal
|
|
- Encourages continued conversation
|
|
- Keeps it under 100 words
|
|
|
|
Respond as if you're having a casual chat with a friend."""
|
|
|
|
# Prepare request to Ollama
|
|
ollama_request = {
|
|
"model": ollama_model,
|
|
"messages": [
|
|
{"role": "system", "content": "You are a charismatic, thoughtful conversationalist."},
|
|
{"role": "user", "content": prompt}
|
|
]
|
|
}
|
|
|
|
logger.debug(f"Sending prompt to Ollama: {prompt}")
|
|
|
|
response = requests.post(
|
|
ollama_endpoint,
|
|
json=ollama_request,
|
|
headers={'Content-Type': 'application/json'},
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assistant_reply = data.get('choices', [{}])[0].get('message', {}).get('content', '').strip()
|
|
|
|
if assistant_reply:
|
|
logger.info(f"{Fore.CYAN}[AUTO RESPONDER] AI GENERATED REPLY: {assistant_reply}{Style.RESET_ALL}")
|
|
return assistant_reply
|
|
else:
|
|
logger.warning("[WARNING] Empty response from Ollama")
|
|
return None
|
|
else:
|
|
logger.error(f"[ERROR] Ollama API error: {response.status_code}")
|
|
return None
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.error("[ERROR] TIMEOUT calling Ollama API")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"[ERROR] ERROR generating response: {str(e)}")
|
|
return None
|
|
|
|
def parse_timestamp(timestamp_str):
|
|
"""Safely parse timestamp string to datetime object"""
|
|
try:
|
|
if isinstance(timestamp_str, str):
|
|
# Try to parse as Unix timestamp first
|
|
try:
|
|
if timestamp_str.isdigit():
|
|
return datetime.fromtimestamp(int(timestamp_str))
|
|
except (ValueError, OSError):
|
|
pass
|
|
|
|
# Try different timestamp formats
|
|
for fmt in ['%Y-%m-%dT%H:%M:%S.%fZ', '%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%d %H:%M:%S']:
|
|
try:
|
|
return datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
|
|
except ValueError:
|
|
continue
|
|
# If all formats fail, try to parse as ISO format
|
|
return datetime.fromisoformat(timestamp_str)
|
|
elif isinstance(timestamp_str, (int, float)):
|
|
# Handle Unix timestamp
|
|
return datetime.fromtimestamp(timestamp_str)
|
|
else:
|
|
logger.warning(f"Unknown timestamp format: {timestamp_str}")
|
|
return datetime.now()
|
|
except Exception as e:
|
|
logger.error(f"Error parsing timestamp {timestamp_str}: {str(e)}")
|
|
return datetime.now()
|
|
|
|
class EnhancedResponderBot:
|
|
"""Enhanced auto responder bot for MeetMe conversations"""
|
|
|
|
def __init__(self):
|
|
self.client = EnhancedMeetMeClient()
|
|
self.last_seen = {} # conversation_id -> timestamp mapping
|
|
self.conversation_stats = {} # Track conversation statistics
|
|
self.max_conversations = 100 # Prevent memory leaks
|
|
|
|
# Bot statistics
|
|
self.bot_stats = {
|
|
'total_responses': 0,
|
|
'conversations_responded': 0,
|
|
'last_activity': datetime.now()
|
|
}
|
|
|
|
def initialize(self):
|
|
"""Initialize bot by logging in and seeding last_seen"""
|
|
try:
|
|
logger.info(f"{Fore.YELLOW}[INITIALIZING] INITIALIZING responder bot...{Style.RESET_ALL}")
|
|
|
|
if not self.client.login():
|
|
logger.error(f"{Fore.RED}[ERROR] FAILED to login. Exiting.{Style.RESET_ALL}")
|
|
return False
|
|
|
|
# Seed last_seen with current conversations
|
|
conversations = self.client.get_conversations()
|
|
for conv in conversations:
|
|
messages = self.client.get_messages(conv['id'], page=1)
|
|
if messages:
|
|
# Get the latest message timestamp
|
|
newest_message = max(messages, key=lambda x: parse_timestamp(x['timestamp']))
|
|
latest_timestamp = newest_message['timestamp']
|
|
self.last_seen[conv['id']] = latest_timestamp
|
|
self.conversation_stats[conv['id']] = {
|
|
'message_count': 0,
|
|
'last_activity': latest_timestamp,
|
|
'display_name': conv.get('display_name', 'Unknown')
|
|
}
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] SEEDED conversation {conv['id']} with timestamp {latest_timestamp}{Style.RESET_ALL}")
|
|
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] INITIALIZED with {len(self.last_seen)} conversations{Style.RESET_ALL}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"{Fore.RED}[ERROR] INITIALIZATION ERROR: {str(e)}{Style.RESET_ALL}")
|
|
return False
|
|
|
|
def _cleanup_old_conversations(self):
|
|
"""Remove old conversations to prevent memory leaks"""
|
|
if len(self.conversation_stats) > self.max_conversations:
|
|
# Remove conversations with no recent activity
|
|
current_time = datetime.now()
|
|
old_conversations = []
|
|
|
|
for conv_id, stats in self.conversation_stats.items():
|
|
last_activity = parse_timestamp(stats['last_activity'])
|
|
if (current_time - last_activity).days > 7: # Remove conversations older than 7 days
|
|
old_conversations.append(conv_id)
|
|
|
|
for conv_id in old_conversations:
|
|
del self.conversation_stats[conv_id]
|
|
if conv_id in self.last_seen:
|
|
del self.last_seen[conv_id]
|
|
|
|
if old_conversations:
|
|
logger.info(f"{Fore.YELLOW}🧹 CLEANED UP {len(old_conversations)} old conversations{Style.RESET_ALL}")
|
|
|
|
def poll_and_respond(self):
|
|
"""Poll for new messages and respond"""
|
|
try:
|
|
# Cleanup old conversations
|
|
self._cleanup_old_conversations()
|
|
|
|
conversations = self.client.get_conversations()
|
|
|
|
for conv in conversations:
|
|
try:
|
|
conv_id = conv['id']
|
|
messages = self.client.get_messages(conv_id, page=1)
|
|
|
|
if not messages:
|
|
continue
|
|
|
|
# Get the newest message
|
|
newest_message = max(messages, key=lambda x: parse_timestamp(x['timestamp']))
|
|
newest_timestamp = newest_message['timestamp']
|
|
sender_id = newest_message['sender_id']
|
|
|
|
# Check if this is a new message from someone else
|
|
if (conv_id not in self.last_seen or
|
|
parse_timestamp(newest_timestamp) > parse_timestamp(self.last_seen[conv_id])) and \
|
|
sender_id != self.client.my_user_id:
|
|
|
|
logger.info(f"{Fore.MAGENTA}[RECEIVED] NEW MESSAGE in conversation {conv_id}: {newest_message['text']}{Style.RESET_ALL}")
|
|
self.client.stats['messages_received'] += 1
|
|
|
|
# Update conversation statistics
|
|
if conv_id not in self.conversation_stats:
|
|
self.conversation_stats[conv_id] = {
|
|
'message_count': 0,
|
|
'last_activity': newest_timestamp,
|
|
'display_name': conv.get('display_name', 'Unknown')
|
|
}
|
|
self.conversation_stats[conv_id]['message_count'] += 1
|
|
self.conversation_stats[conv_id]['last_activity'] = newest_timestamp
|
|
|
|
# Generate AI response
|
|
ai_response = generate_response(newest_message['text'])
|
|
|
|
if ai_response:
|
|
# Send the response
|
|
if self.client.send_message(conv_id, ai_response):
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] SUCCESS: Sent response to conversation {conv_id}{Style.RESET_ALL}")
|
|
# Update last_seen timestamp
|
|
self.last_seen[conv_id] = newest_timestamp
|
|
self.bot_stats['total_responses'] += 1
|
|
self.bot_stats['conversations_responded'] += 1
|
|
self.bot_stats['last_activity'] = datetime.now()
|
|
else:
|
|
logger.error(f"{Fore.RED}[ERROR] FAILED: Could not send response to conversation {conv_id}{Style.RESET_ALL}")
|
|
else:
|
|
logger.warning(f"{Fore.YELLOW}[WARNING] SKIPPING conversation {conv_id} - no AI response generated{Style.RESET_ALL}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"{Fore.RED}[ERROR] ERROR processing conversation {conv['id']}: {str(e)}{Style.RESET_ALL}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
logger.error(f"{Fore.RED}[ERROR] ERROR in poll_and_respond: {str(e)}{Style.RESET_ALL}")
|
|
|
|
def get_stats(self):
|
|
"""Get conversation statistics"""
|
|
total_conversations = len(self.conversation_stats)
|
|
total_messages = sum(stats['message_count'] for stats in self.conversation_stats.values())
|
|
active_conversations = len([conv_id for conv_id, stats in self.conversation_stats.items()
|
|
if stats['message_count'] > 0])
|
|
|
|
return {
|
|
'total_conversations': total_conversations,
|
|
'total_messages': total_messages,
|
|
'active_conversations': active_conversations,
|
|
'conversation_details': self.conversation_stats,
|
|
'bot_responses': self.bot_stats['total_responses'],
|
|
'conversations_responded': self.bot_stats['conversations_responded'],
|
|
'last_activity': self.bot_stats['last_activity']
|
|
}
|
|
|
|
def print_banner():
|
|
"""Print startup banner"""
|
|
banner = f"""
|
|
{Fore.MAGENTA}============================================================
|
|
AUTO RESPONDER BOT v2.0
|
|
Enhanced Edition
|
|
============================================================{Style.RESET_ALL}
|
|
"""
|
|
print(banner)
|
|
|
|
def print_stats(stats):
|
|
"""Print formatted statistics"""
|
|
print(f"\n{Fore.YELLOW}[STATS] BOT STATISTICS:{Style.RESET_ALL}")
|
|
print(f" Messages Sent: {Fore.GREEN}{stats['messages_sent']}{Style.RESET_ALL}")
|
|
print(f" Messages Received: {Fore.MAGENTA}{stats['messages_received']}{Style.RESET_ALL}")
|
|
print(f" Active Conversations: {Fore.CYAN}{stats['active_conversations']}{Style.RESET_ALL}")
|
|
print(f" Total Responses: {Fore.GREEN}{stats['bot_responses']}{Style.RESET_ALL}")
|
|
print(f" Conversations Responded: {Fore.GREEN}{stats['conversations_responded']}{Style.RESET_ALL}")
|
|
print(f" Errors: {Fore.RED}{stats['errors']}{Style.RESET_ALL}")
|
|
print(f" Runtime: {Fore.CYAN}{stats['runtime']}{Style.RESET_ALL}")
|
|
print(f" Messages/Hour: {Fore.CYAN}{stats['messages_per_hour']:.1f}{Style.RESET_ALL}")
|
|
|
|
def main():
|
|
"""Main function for enhanced responder bot"""
|
|
print_banner()
|
|
logger.info(f"{Fore.MAGENTA}[STARTING] Enhanced MeetMe Auto Responder Bot{Style.RESET_ALL}")
|
|
|
|
# Log API configuration
|
|
use_local = os.getenv('USE_LOCAL_API', 'false').lower() == 'true'
|
|
api_url = os.getenv('API_BASE_URL', 'https://api.meetme.com')
|
|
logger.info(f"{Fore.YELLOW}[CONFIG] CONFIG: Using {'local' if use_local else 'production'} API: {api_url}{Style.RESET_ALL}")
|
|
|
|
# Initialize bot
|
|
bot = EnhancedResponderBot()
|
|
if not bot.initialize():
|
|
logger.error(f"{Fore.RED}[ERROR] FAILED to initialize bot. Exiting.{Style.RESET_ALL}")
|
|
return
|
|
|
|
# Get polling interval from environment
|
|
poll_interval = int(os.getenv('POLL_INTERVAL', 30))
|
|
logger.info(f"{Fore.YELLOW}[SETTING UP] SETTING UP polling every {poll_interval} seconds{Style.RESET_ALL}")
|
|
|
|
# Schedule the polling task
|
|
schedule.every(poll_interval).seconds.do(bot.poll_and_respond)
|
|
|
|
# Schedule stats logging every 5 minutes
|
|
def log_stats():
|
|
stats = bot.get_stats()
|
|
client_stats = bot.client.get_stats()
|
|
combined_stats = {**client_stats, **stats}
|
|
print_stats(combined_stats)
|
|
|
|
schedule.every(5).minutes.do(log_stats)
|
|
|
|
logger.info(f"{Fore.GREEN}[SUCCESS] BOT IS RUNNING. Press Ctrl+C to stop.{Style.RESET_ALL}")
|
|
|
|
# Main loop
|
|
try:
|
|
while True:
|
|
schedule.run_pending()
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
logger.info(f"{Fore.YELLOW}[STOP] BOT STOPPED by user{Style.RESET_ALL}")
|
|
# Log final stats
|
|
final_stats = bot.get_stats()
|
|
client_stats = bot.client.get_stats()
|
|
combined_stats = {**client_stats, **final_stats}
|
|
print_stats(combined_stats)
|
|
except Exception as e:
|
|
logger.error(f"{Fore.RED}[ERROR] UNEXPECTED ERROR: {str(e)}{Style.RESET_ALL}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |