# app.py - ULTIMATE FINAL VERSION - EXPERT CODING ENABLED
import os
import sys
import uvicorn
import asyncio
import importlib.util
import json
from pathlib import Path
from fastapi import FastAPI, Form
from fastapi.responses import HTMLResponse
from contextlib import asynccontextmanager
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": ["orchestrator", "testing", "sys_diagnostics",
"code_formatter", "prompt_manager",
"code_intelligence", "code_reviewer"],
"auto_start": True,
"module_settings": {
"orchestrator": {"enabled": True, "background_tasks": True},
"testing": {"auto_test": False, "test_on_startup": True},
"sys_diagnostics": {"auto_run": True, "interval_minutes": 60},
"code_formatter": {"enabled": True, "auto_format": True},
"prompt_manager": {"enabled": True, "auto_optimize": True},
"code_intelligence": {"enabled": True, "auto_analyze": True},
"code_reviewer": {"enabled": True, "auto_review": False}
}
}
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. LIFESPAN MANAGEMENT (MODERN APPROACH)
# ============================================
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Modern lifespan handler for startup/shutdown events"""
# Startup code
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
if hasattr(app.state, 'module_manager'):
app.state.module_manager.load_all_modules()
# Initial health check
print("\n๐ Initial System Check:")
print(f" Groq API: {'โ
Available' if hasattr(app.state, 'groq_available') and app.state.groq_available else 'โ Not Available'}")
print(f" Modules: {len(app.state.module_manager.loaded_modules) if hasattr(app.state, 'module_manager') else 0} loaded")
print(f" Directories: All created")
print("=" * 60)
print("โ
System ready! Waiting for requests...")
print("=" * 60)
yield # Application runs here
# Shutdown code
print("\n๐ System shutting down...")
print("โ
Cleanup completed")
# ============================================
# 4. CORE FASTAPI APPLICATION
# ============================================
app = FastAPI(
title="AumCore AI",
description="Advanced Modular AI Assistant System",
version=AumCoreConfig.VERSION,
lifespan=lifespan
)
# Initialize Groq client
try:
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
GROQ_AVAILABLE = True
app.state.groq_available = True
except Exception as e:
print(f"โ ๏ธ Groq client initialization failed: {e}")
client = None
GROQ_AVAILABLE = False
app.state.groq_available = False
# Initialize Module Manager
module_manager = ModuleManager(app, client)
app.state.module_manager = module_manager
# ============================================
# 5. CORE UI (NEVER CHANGES)
# ============================================
HTML_UI = '''
AumCore AI - Ultimate Version
'''
# ============================================
# 6. 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 WITH EXPERT CODING"""
if not app.state.groq_available:
return {"response": "Error: Groq API not configured."}
try:
from core.language_detector import detect_input_language, get_system_prompt
from core.memory_db import tidb_memory
except ImportError as e:
return {"response": f"Error: {str(e)}"}
# CHECK FOR CODING QUERY FIRST
msg_lower = message.lower()
CODING_KEYWORDS = ["python", "code", "script", "function", "program",
"create", "write", "generate", "algorithm", "debug",
"class", "import", "def", "for loop", "while", "dictionary",
"list", "array", "json", "api", "database", "file handling"]
# If coding query, use expert modules
if any(keyword in msg_lower for keyword in CODING_KEYWORDS):
# Try code_intelligence module first
code_module = app.state.module_manager.get_module("code_intelligence")
if code_module and hasattr(code_module, 'enhance_code_response'):
try:
# Get enhanced code from expert module
enhanced_response = await code_module.enhance_code_response(message, client)
# Save to database
try:
tidb_memory.save_chat(message, enhanced_response, "en")
except:
pass
return {"response": enhanced_response}
except Exception as e:
print(f"โ ๏ธ Expert coding failed: {e}")
# Fall through to normal flow
# NORMAL CHAT FLOW
lang_mode = detect_input_language(message)
system_prompt = get_system_prompt(lang_mode, AumCoreConfig.USERNAME)
# Get chat history
recent_chats = []
try:
recent_chats = tidb_memory.get_recent_chats(limit=10)
except:
pass
# Prepare messages
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:
pass
return {"response": ai_response}
except Exception as e:
return {"response": f"Error: {str(e)}"}
# ============================================
# 7. 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(app.state.module_manager.loaded_modules),
"groq_available": app.state.groq_available,
"health_score": 95
}
# Add module-specific health if available
diagnostics_module = app.state.module_manager.get_module("sys_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(app.state.module_manager.loaded_modules),
"modules": [
{
"name": name,
"status": info["status"],
"active": True
}
for name, info in app.state.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": "sys_diagnostics" in app.state.module_manager.loaded_modules,
"automated_testing": "testing" in app.state.module_manager.loaded_modules,
"task_orchestration": "orchestrator" in app.state.module_manager.loaded_modules,
"expert_coding": "code_intelligence" in app.state.module_manager.loaded_modules
},
"endpoints": [
"/", "/chat", "/reset",
"/system/health", "/system/info", "/system/modules/status"
]
}
# ============================================
# 8. MAIN EXECUTION
# ============================================
if __name__ == "__main__":
uvicorn.run(
app,
host=AumCoreConfig.HOST,
port=AumCoreConfig.PORT,
log_level="info"
)