# app.py - ULTIMATE FINAL VERSION - NEVER TOUCH AGAIN
import os
import sys
import uvicorn
import asyncio
import importlib.util
import json
from pathlib import Path
from fastapi import FastAPI, Form, APIRouter
from fastapi.responses import HTMLResponse, JSONResponse
from groq import Groq
# ============================================
# 1. GLOBAL CONFIGURATION & CONSTANTS
# ============================================
class AumCoreConfig:
"""Central configuration for AumCore AI"""
VERSION = "3.0.0-Final"
USERNAME = "AumCore AI"
PORT = 7860
HOST = "0.0.0.0"
# Paths
BASE_DIR = Path(__file__).parent
MODULES_DIR = BASE_DIR / "modules"
CONFIG_DIR = BASE_DIR / "config"
LOGS_DIR = BASE_DIR / "logs"
DATA_DIR = BASE_DIR / "data"
# Create directories if they don't exist
for dir_path in [MODULES_DIR, CONFIG_DIR, LOGS_DIR, DATA_DIR]:
dir_path.mkdir(exist_ok=True)
# ============================================
# 2. MODULE LOADER SYSTEM (CORE INNOVATION)
# ============================================
class ModuleManager:
"""Dynamic module loading system - FUTURE PROOF"""
def __init__(self, app, client):
self.app = app
self.client = client
self.config = AumCoreConfig()
self.loaded_modules = {}
self.module_config = self._load_module_config()
def _load_module_config(self) -> dict:
"""Load module configuration from JSON"""
config_file = self.config.CONFIG_DIR / "modules.json"
default_config = {
"enabled_modules": ["diagnostics", "testing", "orchestrator"],
"auto_start": True,
"module_settings": {
"diagnostics": {"auto_run": True, "interval_minutes": 60},
"testing": {"auto_test": False, "test_on_startup": True},
"orchestrator": {"enabled": True, "background_tasks": True}
}
}
if not config_file.exists():
config_file.write_text(json.dumps(default_config, indent=4))
return default_config
try:
return json.loads(config_file.read_text())
except:
return default_config
def load_all_modules(self):
"""Load all enabled modules dynamically"""
print("=" * 60)
print("๐ AUMCORE AI - MODULAR SYSTEM INITIALIZING")
print("=" * 60)
for module_name in self.module_config["enabled_modules"]:
self.load_module(module_name)
print(f"๐ฆ Modules Loaded: {len(self.loaded_modules)}")
print(f"๐ง Active: {list(self.loaded_modules.keys())}")
print("=" * 60)
def load_module(self, module_name: str):
"""Load a single module by name"""
module_path = self.config.MODULES_DIR / f"{module_name}.py"
if not module_path.exists():
print(f"โ ๏ธ Module '{module_name}' not found at {module_path}")
return False
try:
# Dynamic module loading
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# Register module with app
if hasattr(module, 'register_module'):
module.register_module(self.app, self.client, AumCoreConfig.USERNAME)
self.loaded_modules[module_name] = {
"module": module,
"path": module_path,
"status": "loaded"
}
print(f"โ
Module '{module_name}' loaded successfully")
return True
else:
print(f"โ ๏ธ Module '{module_name}' missing register_module() function")
return False
except Exception as e:
print(f"โ Failed to load module '{module_name}': {str(e)}")
return False
def get_module(self, module_name: str):
"""Get loaded module instance"""
return self.loaded_modules.get(module_name, {}).get("module")
def get_module_status(self) -> dict:
"""Get status of all modules"""
return {
"total_modules": len(self.loaded_modules),
"loaded_modules": list(self.loaded_modules.keys()),
"config": self.module_config,
"module_details": {
name: info["status"]
for name, info in self.loaded_modules.items()
}
}
# ============================================
# 3. CORE FASTAPI APPLICATION
# ============================================
app = FastAPI(
title="AumCore AI",
description="Advanced Modular AI Assistant System",
version=AumCoreConfig.VERSION
)
# Initialize Groq client
try:
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
GROQ_AVAILABLE = True
except Exception as e:
print(f"โ ๏ธ Groq client initialization failed: {e}")
client = None
GROQ_AVAILABLE = False
# Initialize Module Manager
module_manager = ModuleManager(app, client)
# ============================================
# 4. CORE UI (NEVER CHANGES)
# ============================================
HTML_UI = '''
AumCore AI - Ultimate Version
'''
# ============================================
# 5. CORE ENDPOINTS (NEVER CHANGES)
# ============================================
@app.get("/", response_class=HTMLResponse)
async def get_ui():
"""Main UI endpoint"""
return HTML_UI
@app.post("/reset")
async def reset():
"""Reset system memory"""
try:
# Check if memory_db module exists
try:
from core.memory_db import tidb_memory
return {"success": True, "message": "Memory clear ho gayi hai!"}
except ImportError:
return {"success": True, "message": "Reset command accepted (no TiDB configured)"}
except Exception as e:
return {"success": False, "message": f"Reset error: {str(e)}"}
@app.post("/chat")
async def chat(message: str = Form(...)):
"""Main chat endpoint"""
if not GROQ_AVAILABLE:
return {"response": "Error: Groq API not configured. Please check API key."}
try:
from core.language_detector import detect_input_language, get_system_prompt, generate_basic_code
from core.memory_db import tidb_memory
except ImportError as e:
return {"response": f"Error: Required modules not found - {str(e)}"}
lang_mode = detect_input_language(message)
system_prompt = get_system_prompt(lang_mode, AumCoreConfig.USERNAME)
# Check for code generation requests
msg_lower = message.lower()
CODE_KEYWORDS = ["python code", "write code", "generate code", "create script",
"program for", "function for", "mount google drive",
"colab notebook", "script for", "coding task"]
if any(k in msg_lower for k in CODE_KEYWORDS):
code_response = generate_basic_code(message)
try:
tidb_memory.save_chat(message, code_response, lang_mode)
except Exception as e:
print(f"โ ๏ธ TiDB save error: {e}")
return {"response": code_response}
# Get chat history
recent_chats = []
try:
recent_chats = tidb_memory.get_recent_chats(limit=10)
except Exception as e:
print(f"โ ๏ธ TiDB history fetch error: {e}")
# Prepare messages for Groq
api_messages = [{"role": "system", "content": system_prompt}]
for chat_row in recent_chats:
user_input, ai_response, _ = chat_row
api_messages.append({"role": "user", "content": user_input})
api_messages.append({"role": "assistant", "content": ai_response})
api_messages.append({"role": "user", "content": message})
# Call Groq API
try:
completion = client.chat.completions.create(
model="llama-3.3-70b-versatile",
messages=api_messages,
temperature=0.3,
max_tokens=1000
)
ai_response = completion.choices[0].message.content.strip()
# Save to database
try:
tidb_memory.save_chat(message, ai_response, lang_mode)
except Exception as e:
print(f"โ ๏ธ TiDB save error: {e}")
return {"response": ai_response}
except Exception as e:
error_msg = f"System Error: {str(e)}"
print(f"โ API Error: {error_msg}")
return {"response": error_msg}
# ============================================
# 6. SYSTEM MANAGEMENT ENDPOINTS
# ============================================
@app.get("/system/health")
async def system_health():
"""Overall system health check"""
health_data = {
"success": True,
"timestamp": asyncio.get_event_loop().time(),
"version": AumCoreConfig.VERSION,
"status": "OPERATIONAL",
"modules_loaded": len(module_manager.loaded_modules),
"groq_available": GROQ_AVAILABLE,
"health_score": 95 # Default high score
}
# Add module-specific health if available
diagnostics_module = module_manager.get_module("diagnostics")
if diagnostics_module and hasattr(diagnostics_module, 'get_health'):
try:
module_health = await diagnostics_module.get_health()
health_data.update(module_health)
except:
pass
return health_data
@app.get("/system/modules/status")
async def modules_status():
"""Get status of all loaded modules"""
return {
"success": True,
"total": len(module_manager.loaded_modules),
"modules": [
{
"name": name,
"status": info["status"],
"active": True
}
for name, info in module_manager.loaded_modules.items()
]
}
@app.get("/system/info")
async def system_info():
"""Get complete system information"""
return {
"success": True,
"system": {
"name": "AumCore AI",
"version": AumCoreConfig.VERSION,
"architecture": "Modular Microservices",
"developer": "Sanjay & AI Assistant"
},
"capabilities": {
"ai_chat": True,
"code_generation": True,
"hindi_english": True,
"memory_storage": True,
"system_monitoring": "diagnostics" in module_manager.loaded_modules,
"automated_testing": "testing" in module_manager.loaded_modules,
"task_orchestration": "orchestrator" in module_manager.loaded_modules
},
"endpoints": [
"/", "/chat", "/reset",
"/system/health", "/system/info", "/system/modules/status"
]
}
# ============================================
# 7. STARTUP AND SHUTDOWN EVENTS
# ============================================
@app.on_event("startup")
async def startup_event():
"""Initialize system on startup"""
print("=" * 60)
print("๐ AUMCORE AI - ULTIMATE FINAL VERSION")
print("=" * 60)
print(f"๐ Version: {AumCoreConfig.VERSION}")
print(f"๐ค Username: {AumCoreConfig.USERNAME}")
print(f"๐ Server: http://{AumCoreConfig.HOST}:{AumCoreConfig.PORT}")
print(f"๐ค AI Model: llama-3.3-70b-versatile")
print(f"๐พ Database: TiDB Cloud")
print(f"๐จ UI Features: Code formatting + Copy button")
# Load all modules
module_manager.load_all_modules()
# Initial health check
print("\n๐ Initial System Check:")
print(f" Groq API: {'โ
Available' if GROQ_AVAILABLE else 'โ Not Available'}")
print(f" Modules: {len(module_manager.loaded_modules)} loaded")
print(f" Directories: All created")
print("=" * 60)
print("โ
System ready! Waiting for requests...")
print("=" * 60)
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup on shutdown"""
print("\n๐ System shutting down...")
print("โ
Cleanup completed")
# ============================================
# 8. MAIN EXECUTION
# ============================================
if __name__ == "__main__":
uvicorn.run(
app,
host=AumCoreConfig.HOST,
port=AumCoreConfig.PORT,
log_level="info"
)