IRIS Legacy

#!/usr/bin/env python3 """ IRIS Bot Status Check Script Shows current bot status and health """ import sys import os import asyncio import subprocess from pathlib import Path from datetime import datetime # Add project root to path PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT)) try: import redis.asyncio as redis from dotenv import load_dotenv except ImportError: print("❌ Missing dependencies. Installing...") subprocess.run([sys.executable, "-m", "pip", "install", "-q", "redis", "python-dotenv"]) import redis.asyncio as redis from dotenv import load_dotenv # Colors class Colors: RED = '\033[0;31m' GREEN = '\033[0;32m' YELLOW = '\033[1;33m' BLUE = '\033[0;34m' BOLD = '\033[1m' NC = '\033[0m' def print_header(text: str): print("━" * 60) print(f"{Colors.BOLD}{text}{Colors.NC}") print("━" * 60) print() def print_section(text: str): print(f"{Colors.BLUE}{text}{Colors.NC}") def get_pid_file() -> Path: return PROJECT_ROOT / ".bot.pid" def get_log_file() -> Path: return PROJECT_ROOT / "bot.log" def check_process_status() -> dict: """Check bot process status""" pid_file = get_pid_file() if not pid_file.exists(): return {'running': False, 'pid': None} try: pid = int(pid_file.read_text().strip()) # Check if process exists os.kill(pid, 0) # Get process info try: uptime = subprocess.check_output( ["ps", "-p", str(pid), "-o", "etime="], text=True ).strip() except: uptime = "unknown" try: mem_kb = int(subprocess.check_output( ["ps", "-p", str(pid), "-o", "rss="], text=True ).strip()) mem_mb = mem_kb / 1024 except: mem_mb = 0 return { 'running': True, 'pid': pid, 'uptime': uptime, 'memory_mb': mem_mb } except (ProcessLookupError, ValueError): # Process doesn't exist pid_file.unlink(missing_ok=True) return {'running': False, 'pid': None} async def check_redis_status(host: str, port: int, db: int) -> dict: """Check Redis connection and info""" try: client = redis.Redis( host=host, port=port, db=db, socket_connect_timeout=3, decode_responses=True ) await client.ping() info = await client.info("server") keys_count = await client.dbsize() await client.aclose() return { 'connected': True, 'version': info.get('redis_version', 'unknown'), 'keys': keys_count, 'uptime_days': info.get('uptime_in_days', 0) } except Exception as e: return { 'connected': False, 'error': str(e) } def check_log_file() -> dict: """Check log file status""" log_file = get_log_file() if not log_file.exists(): return {'exists': False} size = log_file.stat().st_size size_mb = size / (1024 * 1024) # Count lines try: with open(log_file, 'r') as f: lines = sum(1 for _ in f) except: lines = 0 # Find last error last_error = None try: with open(log_file, 'r') as f: for line in f: if any(word in line.lower() for word in ['error', 'exception', 'failed']): last_error = line.strip() except: pass return { 'exists': True, 'path': str(log_file), 'size_mb': size_mb, 'lines': lines, 'last_error': last_error } async def main(): print_header("📊 IRIS BOT - STATUS CHECK") # Load environment env_file = PROJECT_ROOT / ".env" if env_file.exists(): load_dotenv(env_file) # Bot Process print_section("🤖 Bot Process:") process_info = check_process_status() if process_info['running']: print(f" Status: {Colors.GREEN}✓ RUNNING{Colors.NC}") print(f" PID: {process_info['pid']}") print(f" Uptime: {process_info['uptime']}") print(f" Memory: {process_info['memory_mb']:.1f} MB") else: print(f" Status: {Colors.RED}✗ NOT RUNNING{Colors.NC}") print() # Redis Connection print_section("💾 Redis Connection:") redis_host = os.getenv('REDIS_HOST', 'localhost') redis_port = int(os.getenv('REDIS_PORT', '6379')) redis_db = int(os.getenv('REDIS_DB', '0')) redis_info = await check_redis_status(redis_host, redis_port, redis_db) if redis_info['connected']: print(f" Status: {Colors.GREEN}✓ CONNECTED{Colors.NC}") print(f" Host: {redis_host}:{redis_port}/{redis_db}") print(f" Version: {redis_info['version']}") print(f" Keys: {redis_info['keys']}") print(f" Uptime: {redis_info['uptime_days']} days") else: print(f" Status: {Colors.RED}✗ DISCONNECTED{Colors.NC}") print(f" Error: {redis_info.get('error', 'unknown')}") print() # API Keys print_section("🔑 API Keys:") api_keys = { 'Telegram': os.getenv('TELEGRAM_BOT_TOKEN'), 'Anthropic': os.getenv('ANTHROPIC_API_KEY'), 'OpenAI': os.getenv('OPENAI_API_KEY') } for name, key in api_keys.items(): if key: print(f" {name}: {Colors.GREEN}✓{Colors.NC}") else: print(f" {name}: {Colors.RED}✗{Colors.NC}") print() # Log File print_section("📝 Log File:") log_info = check_log_file() if log_info['exists']: print(f" Path: {log_info['path']}") print(f" Size: {log_info['size_mb']:.2f} MB ({log_info['lines']} lines)") if log_info['last_error']: print(f" {Colors.YELLOW}Last Error:{Colors.NC}") error_line = log_info['last_error'][:100] print(f" {error_line}") else: print(" No log file found") print() # Configuration print_section("⚙️ Configuration:") print(f" LLM Provider: {os.getenv('LLM_PROVIDER', 'anthropic')}") print(f" Environment: {os.getenv('ENVIRONMENT', 'development')}") print(f" Log Level: {os.getenv('LOG_LEVEL', 'INFO')}") print() print("━" * 60) # Return exit code return 0 if process_info['running'] else 1 if __name__ == "__main__": try: sys.exit(asyncio.run(main())) except KeyboardInterrupt: print("\n\n❌ Interrupted by user") sys.exit(1) except Exception as e: print(f"\n❌ Unexpected error: {e}") import traceback traceback.print_exc() sys.exit(1)