import logging import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from gradio import Server from gradio.data_classes import FileData from fastapi.responses import HTMLResponse, FileResponse from core import process_turn, end_session_click, reset_session, make_initial_state from markdown import markdown log = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, force=True) app = Server() @app.api(name="process_turn") def api_process_turn(audio: FileData, state: dict) -> dict: """Process one voice turn: STT → LLM → TTS. Returns chat, audio, state, feedback.""" try: audio_path = audio.get("path") if isinstance(audio, dict) else audio.path if hasattr(audio, "path") else str(audio) log.info("process_turn audio_path=%s state_turns=%s", audio_path, state.get("turn_count")) result = process_turn(audio_path, state) log.info("process_turn done status=%s", result.get("status")) except Exception as e: log.error("process_turn error: %s", e) return { "chat": state.get("messages", []), "audio_url": None, "state": state, "feedback_intro": "", "feedback_points_forts": [], "feedback_table": [], "feedback_vocabulaire": [], "feedback_priorite": [], "feedback_bilan": {}, "feedback_open": False, "status": f"Erreur: {e}", } audio_url = None if result["audio_file"]: audio_url = f"/audio/{result['audio_file']}" return { "chat": result["chat"], "audio_url": audio_url, "state": result["state"], "feedback_intro": result["feedback_intro"], "feedback_points_forts": result["feedback_points_forts"], "feedback_table": result["feedback_table"], "feedback_vocabulaire": result["feedback_vocabulaire"], "feedback_priorite": result["feedback_priorite"], "feedback_bilan": result["feedback_bilan"], "feedback_open": result["feedback_open"], "status": result["status"], } @app.api(name="end_session") def api_end_session(state: dict) -> dict: """End the session and generate a structured recap.""" result = end_session_click(state) audio_url = None if result["audio_file"]: audio_url = f"/audio/{result['audio_file']}" return { "chat": result["chat"], "audio_url": audio_url, "state": result["state"], "feedback_intro": result["feedback_intro"], "feedback_points_forts": result["feedback_points_forts"], "feedback_table": result["feedback_table"], "feedback_vocabulaire": result["feedback_vocabulaire"], "feedback_priorite": result["feedback_priorite"], "feedback_bilan": result["feedback_bilan"], "feedback_open": result["feedback_open"], "status": result["status"], } @app.api(name="reset_session") def api_reset_session() -> dict: """Reset to a fresh session.""" state = reset_session() return {"state": state} @app.get("/audio/{filename}") async def serve_audio(filename: str): """Serve generated TTS audio files.""" path = os.path.join(tempfile.gettempdir(), filename) if not os.path.exists(path): return {"error": "Audio not found"} return FileResponse(path, media_type="audio/wav") @app.get("/", response_class=HTMLResponse) async def homepage(): html_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend", "custom_index.html") with open(html_path, "r", encoding="utf-8") as f: return f.read() _FIELD_NOTES_HTML_TEMPLATE = """\