#!/usr/bin/env python3 """ Application Barouia-Cortex avec Chat Réel - Version Adaptée Interface web + moteur de conversation intelligent """ from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import uvicorn import logging import json import uuid import time import asyncio from typing import Dict, List, Any, Optional # Configuration du logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("barouia_cortex") # Moteur de chat intégré (fallback) class RealChatEngine: def __init__(self): self.conversations: Dict[str, List[Dict]] = {} self.responses = { "salutations": [ "Bonjour ! Je suis Barouia-Cortex, votre assistant IA. Comment puis-je vous aider ?", "Salut ! Ravie de vous rencontrer. Que souhaitez-vous discuter aujourd'hui ?", "Bonjour ! Je suis là pour vous accompagner. Quelle est votre question ?" ], "questions_simples": { "ça va": "Je fonctionne parfaitement bien, merci ! Et de votre côté, comment allez-vous ?", "heure": f"Selon mon horloge, il est actuellement {time.strftime('%H:%M')}.", "nom": "Je m'appelle Barouia-Cortex, un assistant conversationnel intelligent.", "aide": "Je peux vous aider à répondre à des questions, discuter de divers sujets, ou vous assister dans vos recherches." }, "reponses_generiques": [ "C'est une question intéressante. Que pensez-vous de ce sujet ?", "Je comprends votre point de vue. Pourriez-vous développer un peu plus ?", "Très bonne question ! Mes algorithmes traitent actuellement votre requête.", "Je note votre interrogation. Y a-t-il autre chose que vous aimeriez savoir ?" ] } logger.info("🔧 Moteur Barouia-Cortex initialisé") def process_message(self, conversation_id: str, message: str) -> str: try: logger.info(f"📨 Message reçu: '{message}' (conv: {conversation_id[:8]}...)") # Initialiser la conversation si nécessaire if conversation_id not in self.conversations: self.conversations[conversation_id] = [] # Ajouter à l'historique self.conversations[conversation_id].append({ "role": "user", "content": message, "timestamp": time.time() }) # Nettoyer le message msg_lower = message.lower().strip() # Réponses aux salutations if any(word in msg_lower for word in ['bonjour', 'salut', 'hello', 'coucou', 'yo']): response = self.responses["salutations"][len(conversation_id) % len(self.responses["salutations"])] # Réponses aux questions simples elif "ça va" in msg_lower or "comment vas" in msg_lower: response = self.responses["questions_simples"]["ça va"] elif "heure" in msg_lower: response = self.responses["questions_simples"]["heure"] elif "appelles" in msg_lower or "nom" in msg_lower: response = self.responses["questions_simples"]["nom"] elif "aide" in msg_lower or "que peux" in msg_lower or "fonction" in msg_lower: response = self.responses["questions_simples"]["aide"] # Réponses aux remerciements elif any(word in msg_lower for word in ['merci', 'thank', 'nice']): response = "Je vous en prie ! N'hésitez pas si vous avez d'autres questions." # Au revoir elif any(word in msg_lower for word in ['au revoir', 'bye', 'adieu', 'quit']): response = "Au revoir ! Ce fut un plaisir de discuter avec vous. Revenez quand vous voulez !" # Réponse générique else: response = self.responses["reponses_generiques"][len(msg_lower) % len(self.responses["reponses_generiques"])] response += f"\n\nJ'ai analysé votre message : « {message} »" # Sauvegarder la réponse self.conversations[conversation_id].append({ "role": "assistant", "content": response, "timestamp": time.time() }) logger.info(f"📤 Réponse envoyée: '{response[:50]}...'") return response except Exception as e: logger.error(f"❌ Erreur dans process_message: {e}") return "Désolé, une erreur s'est produite lors du traitement de votre message. Veuillez réessayer." # Initialisation de FastAPI app = FastAPI( title="Barouia-Cortex Real Chat", description="Système de conversation IA réel avec Barouia-Cortex", version="4.0.0", docs_url="/api/docs", redoc_url="/api/redoc" ) # CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["*"], ) # Initialisation du moteur chat_engine = RealChatEngine() # Modèles de données class ChatRequest(BaseModel): conversation_id: str message: str class StartChatRequest(BaseModel): user_id: Optional[str] = "default_user" class HealthResponse(BaseModel): status: str service: str timestamp: float active_conversations: int # Gestionnaire WebSocket class ConnectionManager: def __init__(self): self.active_connections: Dict[str, WebSocket] = {} async def connect(self, websocket: WebSocket, conversation_id: str): await websocket.accept() self.active_connections[conversation_id] = websocket logger.info(f"🔗 WebSocket connecté: {conversation_id[:8]}...") def disconnect(self, conversation_id: str): if conversation_id in self.active_connections: del self.active_connections[conversation_id] logger.info(f"🔌 WebSocket déconnecté: {conversation_id[:8]}...") async def send_message(self, conversation_id: str, message: str): if conversation_id in self.active_connections: await self.active_connections[conversation_id].send_text(message) connection_manager = ConnectionManager() # Routes API @app.get("/") async def root(): return {"message": "Barouia-Cortex API", "version": "4.0.0", "status": "active"} @app.get("/health") async def health_check(): return HealthResponse( status="ok", service="Barouia-Cortex Chat", timestamp=time.time(), active_conversations=len(chat_engine.conversations) ) @app.post("/api/chat/start") async def start_chat(request: StartChatRequest = StartChatRequest()): conversation_id = str(uuid.uuid4()) logger.info(f"🚀 Nouvelle conversation {conversation_id[:8]}... démarrée pour {request.user_id}") return { "success": True, "conversation_id": conversation_id, "message": "Conversation Barouia-Cortex démarrée" } @app.post("/api/chat/send") async def send_message(request: ChatRequest): try: logger.info(f"📝 Traitement message pour {request.conversation_id[:8]}...") # Simulation délai de traitement await asyncio.sleep(0.5) response = chat_engine.process_message(request.conversation_id, request.message) return { "success": True, "response": response, "conversation_id": request.conversation_id, "timestamp": time.time() } except Exception as e: logger.error(f"❌ Erreur endpoint /api/chat/send: {e}") return { "success": False, "response": "Erreur interne du serveur Barouia-Cortex.", "error": str(e) } # Route WebSocket @app.websocket("/ws/{conversation_id}") async def websocket_endpoint(websocket: WebSocket, conversation_id: str): await connection_manager.connect(websocket, conversation_id) try: while True: data = await websocket.receive_text() logger.info(f"🔁 WebSocket message: {data[:50]}...") # Traitement du message response = chat_engine.process_message(conversation_id, data) await connection_manager.send_message(conversation_id, response) except WebSocketDisconnect: connection_manager.disconnect(conversation_id) except Exception as e: logger.error(f"❌ Erreur WebSocket: {e}") connection_manager.disconnect(conversation_id) # Interface Web principale @app.get("/chat", response_class=HTMLResponse) async def chat_interface(): return get_chat_html() def get_chat_html(): return """ Barouia-Cortex - Chat Intelligent

🧠 Barouia-Cortex

Assistant IA conversationnel intelligent

Bonjour ! Je suis Barouia-Cortex, votre assistant IA personnel. 🤖

Je suis là pour discuter, répondre à vos questions et vous accompagner dans vos réflexions. Comment puis-je vous aider aujourd'hui ?
Barouia-Cortex réfléchit...
Connecté à Barouia-Cortex
Messages: 1
""" # Point d'entrée principal if __name__ == "__main__": logger.info("🚀 Démarrage de Barouia-Cortex v4.0...") uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info", access_log=True )