""" ========================================================================================= ARJUN 2.O - ENTERPRISE BACKEND ARCHITECTURE ========================================================================================= Author: Abhay Kumar Architecture: Microservices Pattern within Monolith Compatibility: Docker, Hugging Face Spaces, AWS, GCP Description: Highly robust, scalable, and fail-safe Flask backend serving LLM capabilities, advanced multimodal payload formatting, Edge-TTS synthesis, and system diagnostics. ========================================================================================= """ import os import sys import json import time import uuid import logging import asyncio import tempfile import threading import traceback import requests import dataclasses from datetime import datetime from functools import wraps from typing import Dict, Any, List, Optional, Generator, Union, Callable from flask import ( Flask, request, Response, stream_with_context, render_template_string, send_file, jsonify, make_response, g ) import edge_tts # ========================================================================================= # MODULE 1: ENTERPRISE LOGGING & TELEMETRY # ========================================================================================= class EnterpriseFormatter(logging.Formatter): """ Advanced logging formatter providing color-coded, heavily structured console output to track complex asynchronous events and HTTP requests effectively. """ CYAN = "\x1b[36;20m" GREY = "\x1b[38;20m" YELLOW = "\x1b[33;20m" RED = "\x1b[31;20m" BOLD_RED = "\x1b[31;1m" GREEN = "\x1b[32;20m" RESET = "\x1b[0m" FORMAT_TEMPLATE = "%(asctime)s | %(levelname)-8s | [ARJUN-CORE] | %(module)s:%(lineno)d | %(message)s" FORMATS = { logging.DEBUG: CYAN + FORMAT_TEMPLATE + RESET, logging.INFO: GREEN + FORMAT_TEMPLATE + RESET, logging.WARNING: YELLOW + FORMAT_TEMPLATE + RESET, logging.ERROR: RED + FORMAT_TEMPLATE + RESET, logging.CRITICAL: BOLD_RED + FORMAT_TEMPLATE + RESET } def format(self, record: logging.LogRecord) -> str: log_fmt = self.FORMATS.get(record.levelno, self.FORMAT_TEMPLATE) formatter = logging.Formatter(log_fmt, datefmt='%Y-%m-%d %H:%M:%S.%f') return formatter.format(record) def setup_enterprise_logger() -> logging.Logger: """Initializes the global telemetry and logging system.""" logger = logging.getLogger("ArjunEnterprise") logger.setLevel(logging.DEBUG) # Prevent duplicate handlers if module is reloaded if not logger.handlers: console_handler = logging.StreamHandler() console_handler.setFormatter(EnterpriseFormatter()) logger.addHandler(console_handler) return logger logger = setup_enterprise_logger() # ========================================================================================= # MODULE 2: EXCEPTION HIERARCHY # ========================================================================================= class ArjunSystemException(Exception): """Base exception for all Arjun Core related faults.""" def __init__(self, message: str, status_code: int = 500, payload: dict = None): super().__init__(message) self.message = message self.status_code = status_code self.payload = payload or {} class EnvironmentConfigurationFault(ArjunSystemException): """Raised when critical deployment secrets are missing.""" pass class UpstreamGatewayTimeout(ArjunSystemException): """Raised when the Hugging Face / LLM provider takes too long to respond.""" pass class PayloadFormattingError(ArjunSystemException): """Raised when the incoming JSON cannot be transformed into API-compliant schema.""" pass # ========================================================================================= # MODULE 3: DEPLOYMENT SECRETS & CONFIGURATION MANAGER # ========================================================================================= @dataclasses.dataclass class SystemState: boot_time: float = time.time() total_requests_served: int = 0 total_errors_caught: int = 0 state = SystemState() class ConfigurationManager: """ Singleton pattern configuration manager. Secures runtime variables and ensures zero hardcoded credentials exist within the executable code. """ _instance = None def __new__(cls): if cls._instance is None: cls._instance = super(ConfigurationManager, cls).__new__(cls) cls._instance._initialize() return cls._instance def _initialize(self): logger.info("Bootstrapping Configuration Manager...") self.api_key = os.environ.get("YOUR_VEDIKA_API_KEY", "") self.base_url = os.environ.get("BASE_URL", "") self.model_id = os.environ.get("MODEL_ID", "") self.invoke_url = self._build_secure_endpoint(self.base_url) self.verify_integrity() def _build_secure_endpoint(self, base: str) -> str: """Constructs the exact completion endpoint avoiding double slashes.""" if not base: return "" base = base.strip() if not base.endswith("/chat/completions"): return f"{base.rstrip('/')}/chat/completions" return base def verify_integrity(self) -> bool: """Audits the environment. Will not crash, but logs critical warnings.""" missing_vars = [] if not self.api_key: missing_vars.append("YOUR_VEDIKA_API_KEY") if not self.base_url: missing_vars.append("BASE_URL") if not self.model_id: missing_vars.append("MODEL_ID") if missing_vars: logger.critical(f"ENVIRONMENT AUDIT FAILED. Missing Secrets: {', '.join(missing_vars)}") return False logger.info(f"Environment Audit Passed. Target Model: {self.model_id}") return True config = ConfigurationManager() # ========================================================================================= # MODULE 4: CIRCUIT BREAKER & RETRY MECHANISMS # ========================================================================================= def with_retry(max_retries: int = 3, backoff_factor: float = 1.5): """ Enterprise decorator: Automatically retries failing network requests. Prevents the backend from failing immediately due to micro-outages. """ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): retries = 0 while retries < max_retries: try: return func(*args, **kwargs) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: retries += 1 logger.warning(f"Network fault detected ({str(e)}). Retry {retries}/{max_retries} executing...") time.sleep(backoff_factor * retries) logger.error("Maximum retries exhausted. Circuit Breaker triggered.") raise UpstreamGatewayTimeout("Upstream AI Matrix is currently unreachable.", 504) return wrapper return decorator # ========================================================================================= # MODULE 5: TEMPORARY ASSET & GARBAGE COLLECTION SYSTEM # ========================================================================================= class AssetGarbageCollector: """ Manages temporary files (like synthesized audio) to ensure Docker instances and Hugging Face spaces do not run out of ephemeral storage over time. """ def __init__(self): self.storage_directory = tempfile.gettempdir() self.registry = set() logger.info(f"Garbage Collector initialized at virtual mount: {self.storage_directory}") def allocate_file(self, prefix: str = "arjun_asset_", extension: str = ".tmp") -> str: """Allocates a cryptographically unique file path.""" file_id = uuid.uuid4().hex secure_path = os.path.join(self.storage_directory, f"{prefix}{file_id}{extension}") self.registry.add(secure_path) return secure_path def schedule_demolition(self, target_path: str, lifespan_seconds: int = 120): """Asynchronously wipes files from disk after they are served to the client.""" def _demolish(): time.sleep(lifespan_seconds) try: if os.path.exists(target_path): os.remove(target_path) logger.debug(f"[GC] Securely wiped temporary asset: {target_path}") if target_path in self.registry: self.registry.remove(target_path) except Exception as e: logger.error(f"[GC] Failed to wipe asset {target_path}: {str(e)}") demolition_thread = threading.Thread(target=_demolish, daemon=True) demolition_thread.start() gc_system = AssetGarbageCollector() # ========================================================================================= # MODULE 6: NEURAL ACOUSTIC ENGINE (EDGE TTS WRAPPER) # ========================================================================================= class NeuralAcousticEngine: """ Bridges Flask's synchronous architecture with Edge-TTS asynchronous nature. Utilizes isolated event loops to prevent thread blocking and server hangs. """ def __init__(self): # Default Voice: Christopher for English, but dynamically handled usually self.default_voice = "en-US-ChristopherNeural" self.default_pitch = "+0%" self.default_rate = "+0%" logger.info("Neural Acoustic Engine (Edge-TTS) armed and ready.") def synthesize_speech(self, text: str, output_path: str, lang_code: str = 'en') -> bool: """Executes the synthesis in an isolated asyncio loop.""" sanitized_text = self._sanitize_text(text) if not sanitized_text: logger.warning("TTS aborted: Text was empty post-sanitization.") return False # Dynamic voice selection based on detected or requested language target_voice = "hi-IN-MadhurNeural" if lang_code == 'hi' else self.default_voice logger.info(f"Initiating acoustic synthesis. Length: {len(sanitized_text)} chars. Voice: {target_voice}") async def _async_compile(): try: communicator = edge_tts.Communicate( text=sanitized_text, voice=target_voice, pitch=self.default_pitch, rate=self.default_rate ) await communicator.save(output_path) return True except Exception as e: logger.error(f"TTS Compilation Failure: {str(e)}") return False # Isolated Event Loop Execution isolated_loop = asyncio.new_event_loop() asyncio.set_event_loop(isolated_loop) try: success = isolated_loop.run_until_complete(_async_compile()) return success except Exception as err: logger.error(f"Event Loop Integrity Failure during TTS: {str(err)}") return False finally: isolated_loop.close() def _sanitize_text(self, raw_text: str) -> str: """Strips markdown, code blocks, and reasoning tags before passing to audio generator.""" import re text = re.sub(r'```.*? ```', ' Code block omitted for audio. ', raw_text, flags=re.DOTALL) text = re.sub(r'!\[.*?\]\(.*?\)', '', text) text = re.sub(r'\[.*?\]\(.*?\)', '', text) text = re.sub(r'.*?', '', text, flags=re.DOTALL) text = text.replace('*', '').replace('_', '').replace('`', '').replace('#', '') return text.strip() acoustic_engine = NeuralAcousticEngine() # ========================================================================================= # MODULE 7: LLM PAYLOAD ARCHITECT & STREAM MANAGER # ========================================================================================= class LLMGateway: """ The core logic hub for interacting with upstream Hugging Face Inference endpoints. Responsible for intelligent payload morphing based on attachment presence. """ def __init__(self, config_ref: ConfigurationManager): self.cfg = config_ref def format_history(self, history: List[Dict]) -> List[Dict]: """Ensures the history strictly matches OpenAI/HF specs (user/assistant).""" formatted = [] for msg in history: # Map frontend 'bot' to standard 'assistant' role = "assistant" if msg.get("role") == "bot" else "user" content = str(msg.get("content", "")) if content.strip(): formatted.append({"role": role, "content": content}) return formatted def build_payload(self, user_msg: str, attachments: List[Dict], sys_prompt: str, history: List[Dict]) -> List[Dict]: """ CRITICAL LOGIC: Morphs the payload based on context. If no attachments exist, 'content' MUST be a string to avoid 400 Bad Requests on standard text models. If attachments exist, 'content' becomes a multimodal List[Dict]. """ messages = [] # 1. System Directive if sys_prompt and sys_prompt.strip(): messages.append({"role": "system", "content": sys_prompt.strip()}) # 2. Historical Context messages.extend(self.format_history(history)) # 3. Current Turn Analysis if not attachments: # STANDARD TEXT MODE (Prevents API crashes) safe_text = user_msg.strip() if user_msg.strip() else "Hello." messages.append({"role": "user", "content": safe_text}) logger.debug("Payload formulated as STRICT TEXT (No attachments).") else: # MULTIMODAL MODE (Vision / Audio capable endpoints) content_array = [] if user_msg.strip(): content_array.append({"type": "text", "text": user_message.strip()}) for att in attachments: att_type = att.get("type") b64_data = att.get("data") if not b64_data: continue if att_type == "image": content_array.append({ "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64_data}"} }) elif att_type in ["audio", "file", "document", "video"]: # Depending on backend capability, we append it. # Note: Many endpoints ignore audio, but we construct it properly anyway. content_array.append({ "type": "input_audio", "input_audio": {"data": b64_data, "format": "wav"} }) if not content_array: content_array.append({"type": "text", "text": "Hello."}) messages.append({"role": "user", "content": content_array}) logger.debug(f"Payload formulated as MULTIMODAL. Items: {len(content_array)}") return messages def execute_stream(self, api_payload: Dict[str, Any]) -> Generator[str, None, None]: """ Opens a persistent connection to the Upstream LLM, reads the stream chunk by chunk, and yields it directly back to the Flask client. """ headers = { "Authorization": f"Bearer {self.cfg.api_key}", "Accept": "text/event-stream", "Content-Type": "application/json" } logger.info(f"Opening Streaming Vector to: {self.cfg.invoke_url}") try: with requests.post( self.cfg.invoke_url, headers=headers, json=api_payload, stream=True, timeout=180 # Extended timeout for massive reasoning models ) as response: if response.status_code != 200: err_txt = response.text logger.error(f"Upstream API Failure [{response.status_code}]: {err_txt}") error_json = json.dumps({"choices": [{"delta": {"content": f"\n\n**CRITICAL UPSTREAM ERROR {response.status_code}:**\nThe AI Provider rejected the payload. Ensure your model supports the requested features.\n\n`{err_txt[:200]}`"}}]}) yield f"data: {error_json}\n\n" yield "data: [DONE]\n\n" return # Successfully connected. Begin data relay. for raw_bytes in response.iter_lines(): if raw_bytes: decoded_string = raw_bytes.decode("utf-8") if decoded_string.startswith("data: "): yield decoded_string + "\n\n" except requests.exceptions.Timeout: logger.error("Upstream Gateway Timeout.") yield f"data: {json.dumps({'choices': [{'delta': {'content': '**System Alert:** Connection to the AI matrix timed out.'}}]})}\n\n" yield "data: [DONE]\n\n" except requests.exceptions.RequestException as e: logger.error(f"Upstream Network Disconnect: {str(e)}") yield f"data: {json.dumps({'choices': [{'delta': {'content': '**System Alert:** Network disconnect detected.'}}]})}\n\n" yield "data: [DONE]\n\n" except Exception as e: logger.error(f"Stream Interpreter Crash: {str(e)}") traceback.print_exc() yield f"data: {json.dumps({'choices': [{'delta': {'content': '**System Alert:** Internal processing failure during stream.'}}]})}\n\n" yield "data: [DONE]\n\n" llm_gateway = LLMGateway(config) # ========================================================================================= # MODULE 8: FLASK APPLICATION & ROUTING ARCHITECTURE # ========================================================================================= app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 50 * 1024 * 1024 # 50 MB Max upload size limitation @app.before_request def intercept_request(): """Middleware: Analytics and Security Checks""" g.start_time = time.time() logger.debug(f"Intercepted Request: {request.method} {request.path}") @app.after_request def inject_security_headers(response): """Middleware: Injects Cross-Origin Resource Sharing and Security Headers""" response.headers["Access-Control-Allow-Origin"] = "*" response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS" response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization" response.headers["X-Powered-By"] = "Arjun Intelligence Core" response.headers["X-Frame-Options"] = "SAMEORIGIN" if hasattr(g, 'start_time'): latency = (time.time() - g.start_time) * 1000 logger.debug(f"Request resolved in {latency:.2f}ms") return response # ----------------------------------------------------------------------------------------- # ENDPOINT: FRONTEND SERVING # ----------------------------------------------------------------------------------------- @app.route('/', methods=['GET']) def render_interface(): """Serves the primary monolithic HTML application.""" logger.info("Serving Application Interface.") try: base_dir = os.path.dirname(os.path.abspath(__file__)) html_path = os.path.join(base_dir, 'index.html') with open(html_path, 'r', encoding='utf-8') as file: return render_template_string(file.read()) except FileNotFoundError: logger.critical("Primary UI file 'index.html' is missing from the directory.") return make_response( "

Fatal Error 404

The core UI file 'index.html' could not be located on the server disk.

", 404 ) except Exception as e: logger.error(f"Template Rendering Failure: {str(e)}") return make_response(f"

Server Fault 500

{str(e)}

", 500) # ----------------------------------------------------------------------------------------- # ENDPOINT: LLM CHAT PROCESSING # ----------------------------------------------------------------------------------------- @app.route('/api/chat', methods=['POST', 'OPTIONS']) def process_chat_transaction(): """ Main Gateway Endpoint. Receives JSON from frontend, builds appropriate payload, and proxies the streaming response back to the client. """ if request.method == 'OPTIONS': return Response(status=200) state.total_requests_served += 1 # Configuration Guard if not config.api_key or not config.invoke_url or not config.model_id: logger.critical("Transaction blocked. System configuration invalid.") return jsonify({"error": "Backend misconfigured. API Keys missing."}), 500 # JSON Parsing Guard try: payload_data = request.get_json() or {} except Exception as e: logger.error(f"Malformed JSON intercepted: {str(e)}") return jsonify({"error": "Payload must be strictly valid JSON."}), 400 user_text = payload_data.get("message", "") attachments = payload_data.get("attachments", []) system_prompt = payload_data.get("system_prompt", "") history = payload_data.get("history", []) max_tokens = payload_data.get("max_tokens", 4096) temperature = payload_data.get("temperature", 0.75) logger.info(f"Processing transaction. History Length: {len(history)} | Attachments: {len(attachments)}") # Construct the highly-optimized payload try: mapped_messages = llm_gateway.build_payload(user_text, attachments, system_prompt, history) except Exception as e: logger.error(f"Payload Compilation Error: {str(e)}") traceback.print_exc() return jsonify({"error": "Internal Error formatting LLM payload."}), 500 api_dispatch_json = { "model": config.model_id, "messages": mapped_messages, "max_tokens": int(max_tokens), "temperature": float(temperature), "top_p": 0.85, # Slightly broadened for better code generation "stream": True } # Execute stream relay return Response( stream_with_context(llm_gateway.execute_stream(api_dispatch_json)), mimetype='text/event-stream' ) # ----------------------------------------------------------------------------------------- # ENDPOINT: NATIVE TTS FALLBACK (EDGE-TTS) # ----------------------------------------------------------------------------------------- @app.route('/api/tts', methods=['POST', 'OPTIONS']) def synthesize_audio(): """ Fallback TTS Endpoint. Though frontend uses Gradio primarily, this remains as an internal fallback and microservice utility. Converts text to speech locally. """ if request.method == 'OPTIONS': return Response(status=200) try: data = request.get_json() or {} text_content = data.get("text", "").strip() lang_mode = data.get("lang", "en") # 'en' or 'hi' except Exception as e: logger.error(f"Malformed JSON in TTS request: {str(e)}") return jsonify({"error": "Invalid payload format"}), 400 if not text_content: return jsonify({"error": "Empty text payload provided."}), 400 logger.info(f"TTS requested for {len(text_content)} characters. Lang: {lang_mode}") output_path = gc_system.allocate_file(prefix="arjun_tts_", extension=".mp3") # Synchronously await asynchronous compilation success = acoustic_engine.synthesize_speech(text_content, output_path, lang_code=lang_mode) if not success or not os.path.exists(output_path): logger.error("Acoustic engine failed to finalize the asset.") return jsonify({"error": "Acoustic generation failed."}), 500 logger.info(f"Acoustic asset finalized: {output_path}") # Fire-and-forget destruction timer gc_system.schedule_demolition(output_path, lifespan_seconds=90) try: return send_file( output_path, mimetype="audio/mpeg", as_attachment=False, download_name="arjun_transmission.mp3" ) except Exception as e: logger.error(f"Transmission failure of acoustic asset: {str(e)}") return jsonify({"error": "Failed to transmit binary audio data."}), 500 # ----------------------------------------------------------------------------------------- # ENDPOINT: SYSTEM DIAGNOSTICS & HEALTH # ----------------------------------------------------------------------------------------- @app.route('/health', methods=['GET']) def system_health_check(): """Detailed health check endpoint for Load Balancers and Docker health checks.""" uptime_seconds = time.time() - state.boot_time health_report = { "status": "OPERATIONAL", "system_name": "Arjun Core Version 2.0", "author": "Abhay Kumar", "diagnostics": { "uptime_seconds": round(uptime_seconds, 2), "requests_processed": state.total_requests_served, "temporary_assets_tracked": len(gc_system.registry), "configuration_status": "Valid" if config.verify_integrity() else "Compromised" }, "timestamp_utc": datetime.utcnow().isoformat() } return jsonify(health_report), 200 # ========================================================================================= # MODULE 9: GLOBAL ERROR HANDLERS # ========================================================================================= @app.errorhandler(404) def handle_404(error): state.total_errors_caught += 1 logger.warning(f"Unmatched Route Requested: {request.path}") return jsonify({"error": "Endpoint not found within the Arjun routing matrix."}), 404 @app.errorhandler(405) def handle_405(error): return jsonify({"error": "Method Not Allowed."}), 405 @app.errorhandler(500) def handle_500(error): state.total_errors_caught += 1 logger.error(f"Unhandled Server Exception Triggered on {request.path}") return jsonify({"error": "Internal Server Failure. Check logs."}), 500 # ========================================================================================= # MODULE 10: EXECUTION BOOTSTRAP # ========================================================================================= def display_terminal_splash(): """Prints an impressive ASCII art splash screen to the standard output upon boot.""" splash_text = """ ======================================================================= █████╗ ██████╗ ██╗██╗ ██╗███╗ ██╗ ██████╗ ██████╗ ██╔══██╗██╔══██╗ ██║██║ ██║████╗ ██║ ╚════██╗ ██╔═══██╗ ███████║██████╔╝ ██║██║ ██║██╔██╗ ██║ █████╔╝ ██║ ██║ ██╔══██║██╔══██╗██ ██║██║ ██║██║╚██╗██║ ██╔═══╝ ██║ ██║ ██║ ██║██║ ██║╚█████╔╝╚██████╔╝██║ ╚████║ ███████╗ ╚██████╔╝ ╚═╝ ╚═╝╚═╝ ╚═╝ ╚════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚══════╝ ╚═════╝ ======================================================================= Architecture : Enterprise AI Matrix Author : Abhay Kumar Version : 2.0 (Production Stable) Protocols : Edge-TTS, Dynamic Context Mapping, Secure Streams ======================================================================= """ # Print securely print("\x1b[36;1m" + splash_text + "\x1b[0m", flush=True) if __name__ == '__main__': # Initialize terminal visuals display_terminal_splash() # Retrieve port bindings (Default 7860 for Hugging Face Spaces compatibility) target_port = int(os.environ.get("PORT", 7860)) logger.info(f"Ignition sequence initiated. Binding server to 0.0.0.0:{target_port}") # Using Flask's threaded server. # For hardcore production on AWS/GCP, wrap this with Gunicorn: gunicorn -w 4 -k gevent app:app app.run(host='0.0.0.0', port=target_port, threaded=True, debug=False)