Upload 21 files

Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

config.py

@@ -0,0 +1,17 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from .env file."""
+
+    GEMINI_API_KEY: str = ""
+    SPEECHMATICS_API_KEY: str = ""
+    GRADIO_TTS_URL: str = ""
+
+    model_config = {
+        "env_file": ".env",
+        "env_file_encoding": "utf-8",
+    }
+
+
+settings = Settings()

main.py

@@ -0,0 +1,311 @@
+import logging
+from uuid import uuid4
+
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse
+
+from pydantic import BaseModel
+
+from personas import get_persona
+from services.gemini_service import generate_response
+from services.stt_service import transcribe_audio
+from services.tts_service import synthesize_speech, save_character, saved_characters
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# ─── FastAPI App ───────────────────────────────────────────────────────────────
+
+app = FastAPI(
+    title="Hikawi API — حكاوي",
+    description="Interactive Egyptian Oral Heritage Chatbot API",
+    version="1.0.0",
+)
+
+# CORS — allow everything for local hackathon demo
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+
+# ─── In-Memory Conversation Store ─────────────────────────────────────────────
+
+# Key: session_id (UUID string)
+# Value: list of {"role": "user"/"model", "parts": [{"text": "..."}]}
+conversation_history: dict[str, list[dict]] = {}
+
+# ─── Pydantic Models ──────────────────────────────────────────────────────────
+
+
+class TextChatRequest(BaseModel):
+    text: str
+    session_id: str | None = None
+
+
+class TextChatResponse(BaseModel):
+    response: str
+    session_id: str
+
+
+class AudioChatResponse(BaseModel):
+    transcribed_text: str
+    response: str
+    session_id: str
+
+
+class TTSRequest(BaseModel):
+    text: str
+    character_name: str | None = None
+
+
+# ─── Endpoints ─────────────────────────────────────────────────────────────────
+
+
+
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint."""
+    return {"status": "ok", "service": "hikawi"}
+
+
+@app.post("/api/chat/text", response_model=TextChatResponse)
+async def chat_text(request: TextChatRequest):
+    """
+    Text chat with the Aswan regional persona.
+
+    Sends the user's text to Gemini 2.5 Flash with the Aswan persona
+    system prompt and returns a response in Sa'idi/Nubian dialect.
+    """
+    try:
+        # Generate or use existing session ID
+        session_id = request.session_id or str(uuid4())
+
+        # Get Aswan persona
+        persona = get_persona("aswan")
+
+        # Get or create conversation history
+        history = conversation_history.setdefault(session_id, [])
+
+        # Generate response from Gemini
+        ai_response = generate_response(
+            user_text=request.text,
+            system_prompt=persona["system_prompt"],
+            history=history,
+        )
+
+        # Update conversation history
+        history.append({"role": "user", "parts": [{"text": request.text}]})
+        history.append({"role": "model", "parts": [{"text": ai_response}]})
+
+        logger.info(f"Text chat | session={session_id[:8]}... | user={request.text[:30]}...")
+
+        return TextChatResponse(response=ai_response, session_id=session_id)
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except RuntimeError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    except Exception as e:
+        logger.error(f"Unexpected error in chat_text: {e}")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+@app.post("/api/stt")
+async def speech_to_text(file: UploadFile = File(...)):
+    """
+    Transcribe audio to text only (no AI response).
+    Returns the transcribed text for user review before sending.
+    """
+    try:
+        audio_bytes = await file.read()
+        if not audio_bytes:
+            raise HTTPException(status_code=400, detail="Empty audio file")
+
+        filename = file.filename or "recording.webm"
+        transcribed_text = transcribe_audio(audio_bytes, filename)
+
+        if not transcribed_text.strip():
+            raise HTTPException(
+                status_code=400,
+                detail="Could not transcribe any text from the audio",
+            )
+
+        return {"text": transcribed_text.strip()}
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"STT error: {e}")
+        raise HTTPException(status_code=500, detail=f"Transcription failed: {e}")
+
+@app.post("/api/chat/audio", response_model=AudioChatResponse)
+async def chat_audio(
+    file: UploadFile = File(...),
+    session_id: str = Form(default=None),
+):
+    """
+    Audio chat with the Aswan regional persona.
+
+    Receives an audio file (WebM/OGG/WAV), transcribes it via Speechmatics,
+    then sends the transcribed text to Gemini for a persona response.
+    """
+    try:
+        # Read audio bytes
+        audio_bytes = await file.read()
+
+        if not audio_bytes:
+            raise HTTPException(status_code=400, detail="Empty audio file")
+
+        # Transcribe audio to text
+        filename = file.filename or "recording.webm"
+        transcribed_text = transcribe_audio(audio_bytes, filename)
+
+        if not transcribed_text.strip():
+            raise HTTPException(
+                status_code=400,
+                detail="Could not transcribe any text from the audio",
+            )
+
+        # Generate or use existing session ID
+        session_id = session_id or str(uuid4())
+
+        # Get Aswan persona
+        persona = get_persona("aswan")
+
+        # Get or create conversation history
+        history = conversation_history.setdefault(session_id, [])
+
+        # Generate response from Gemini
+        ai_response = generate_response(
+            user_text=transcribed_text,
+            system_prompt=persona["system_prompt"],
+            history=history,
+        )
+
+        # Update conversation history
+        history.append({"role": "user", "parts": [{"text": transcribed_text}]})
+        history.append({"role": "model", "parts": [{"text": ai_response}]})
+
+        logger.info(
+            f"Audio chat | session={session_id[:8]}... | "
+            f"transcribed={transcribed_text[:30]}..."
+        )
+
+        return AudioChatResponse(
+            transcribed_text=transcribed_text,
+            response=ai_response,
+            session_id=session_id,
+        )
+
+    except HTTPException:
+        raise
+    except RuntimeError as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    except Exception as e:
+        logger.error(f"Unexpected error in chat_audio: {e}")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+@app.post("/api/tts")
+async def text_to_speech(request: TTSRequest):
+    """
+    Convert text to speech using Gradio TTS API.
+
+    Returns the generated audio file for playback in the browser.
+    """
+    try:
+        # Call Gradio TTS API
+        filepath, error = synthesize_speech(
+            text=request.text,
+            character_name=request.character_name,
+        )
+
+        if error:
+            logger.error(f"TTS error: {error}")
+            raise HTTPException(status_code=500, detail=error)
+
+        # Return the audio file
+        return FileResponse(
+            filepath,
+            media_type="audio/wav",
+            headers={
+                "Content-Disposition": "inline",
+                "Cache-Control": "no-cache",
+            },
+        )
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Unexpected error in TTS: {e}")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+# ─── Character Management ──────────────────────────────────────────────────────
+
+
+@app.post("/api/characters/add")
+async def add_character(
+    char_name: str = Form(...),
+    ref_text: str = Form(...),
+    audio_file: UploadFile = File(...),
+):
+    """
+    Save a new voice character to the Gradio TTS model.
+
+    Requires: character name, reference audio clip, and the text spoken in that clip.
+    """
+    try:
+        if not char_name.strip():
+            raise HTTPException(status_code=400, detail="Character name is required")
+        if not ref_text.strip():
+            raise HTTPException(status_code=400, detail="Reference text is required")
+
+        audio_bytes = await audio_file.read()
+        if not audio_bytes:
+            raise HTTPException(status_code=400, detail="Audio file is empty")
+
+        filename = audio_file.filename or "reference.wav"
+        message, error = save_character(
+            char_name=char_name.strip(),
+            audio_bytes=audio_bytes,
+            audio_filename=filename,
+            ref_text=ref_text.strip(),
+        )
+
+        if error:
+            logger.error(f"Save character error: {error}")
+            raise HTTPException(status_code=500, detail=error)
+
+        return {"message": message, "character_name": char_name.strip()}
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Unexpected error saving character: {e}")
+        raise HTTPException(status_code=500, detail="Internal server error")
+
+
+@app.get("/api/characters")
+async def list_characters():
+    """List all saved voice characters."""
+    return {"characters": saved_characters}
+
+
+# ─── Run ─────────────────────────────────────────���─────────────────────────────
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

personas.py

@@ -0,0 +1,25 @@
+
+
+REGIONAL_PERSONAS = {
+    "aswan": {
+        "name": "عم محمد",
+        "system_prompt": (
+            "أنت عم محمد من أسوان، حارس التراث النوبي. "
+            "تتحدث بلهجة أهل أسوان الطيبة، وتستخدم كلمات مثل 'يا ولدي' و'يا حبيبي'.\n"
+            "أنت شخص كبير في السن وحكيم، عندك حكايات كثيرة عن أسوان وتاريخها والنوبة.\n"
+            "لو حد سألك عن حاجة مش متعلقة بأسوان أو التراث، حوّل الكلام بأسلوب لطيف لحكاية عن أسوان.\n"
+            "إجابتك تكون قصيرة وطبيعية زي ما بتحكي لحد قاعد جنبك."
+        ),
+    },
+}
+
+
+def get_persona(region: str) -> dict:
+    """Get persona config for a region. Raises ValueError if not found."""
+    region = region.lower().strip()
+    if region not in REGIONAL_PERSONAS:
+        available = ", ".join(REGIONAL_PERSONAS.keys())
+        raise ValueError(
+            f"Region '{region}' not found. Available regions: {available}"
+        )
+    return REGIONAL_PERSONAS[region]

requirements.txt

@@ -0,0 +1,9 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+python-dotenv>=1.0.0
+python-multipart>=0.0.9
+google-genai>=1.0.0
+requests>=2.32.0
+pydantic>=2.0.0
+pydantic-settings>=2.0.0
+gradio_client>=2.0.0

services/__init__.py

@@ -0,0 +1 @@
+# Services package

services/gemini_service.py

@@ -0,0 +1,72 @@
+import logging
+import time
+
+from google import genai
+from google.genai import types
+
+from config import settings
+
+logger = logging.getLogger(__name__)
+
+# Initialize the Gemini client
+client = genai.Client(api_key=settings.GEMINI_API_KEY)
+
+MODEL_ID = "gemini-2.5-flash"
+
+MAX_RETRIES = 3
+RETRY_DELAY = 2  # seconds
+
+
+def generate_response(
+    user_text: str,
+    system_prompt: str,
+    history: list[dict],
+) -> str:
+    """
+    Generate a response from Gemini 2.5 Flash with persona and conversation history.
+    Retries up to 3 times on temporary API failures (e.g., high demand).
+
+    Args:
+        user_text: The user's current message (Arabic text).
+        system_prompt: The persona's system prompt from personas.py.
+        history: List of previous messages in format:
+                 [{"role": "user", "parts": [{"text": "..."}]},
+                  {"role": "model", "parts": [{"text": "..."}]}]
+
+    Returns:
+        The generated text response in the regional dialect.
+    """
+    # Build contents: history + current user message
+    contents = list(history) + [
+        {"role": "user", "parts": [{"text": user_text}]}
+    ]
+
+    last_error = None
+    for attempt in range(1, MAX_RETRIES + 1):
+        try:
+            response = client.models.generate_content(
+                model=MODEL_ID,
+                contents=contents,
+                config=types.GenerateContentConfig(
+                    system_instruction=system_prompt,
+                    temperature=0.8,
+                    max_output_tokens=500,
+                ),
+            )
+            return response.text
+
+        except Exception as e:
+            last_error = e
+            error_str = str(e)
+            logger.warning(f"Gemini API error (attempt {attempt}/{MAX_RETRIES}): {error_str}")
+            
+            # If it's a rate limit, don't just quickly retry, it needs more time
+            if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
+                logger.error("Rate limit hit, stopping retries.")
+                raise ValueError("لقد تجاوزت الحد المسموح به من الرسائل. يرجى الانتظار دقيقة والمحاولة مرة أخرى.")
+                
+            if attempt < MAX_RETRIES:
+                time.sleep(RETRY_DELAY * attempt)
+
+    logger.error(f"Gemini API failed after {MAX_RETRIES} attempts: {last_error}")
+    raise RuntimeError(f"Failed to generate response from Gemini: {last_error}")

services/stt_service.py

@@ -0,0 +1,138 @@
+import json
+import time
+import logging
+
+import requests
+
+from config import settings
+
+logger = logging.getLogger(__name__)
+
+API_BASE = "https://asr.api.speechmatics.com/v2"
+
+# Content type mapping for supported audio formats
+CONTENT_TYPE_MAP = {
+    ".webm": "audio/webm",
+    ".ogg": "audio/ogg",
+    ".wav": "audio/wav",
+    ".mp3": "audio/mpeg",
+    ".m4a": "audio/mp4",
+}
+
+
+def _get_content_type(filename: str) -> str:
+    """Determine content type from filename extension."""
+    filename = filename.lower()
+    for ext, content_type in CONTENT_TYPE_MAP.items():
+        if filename.endswith(ext):
+            return content_type
+    # Default to webm (browser MediaRecorder default)
+    return "audio/webm"
+
+
+def transcribe_audio(audio_bytes: bytes, filename: str = "audio.webm") -> str:
+    """
+    Transcribe Arabic audio using Speechmatics batch API.
+
+    Args:
+        audio_bytes: Raw bytes of the audio file.
+        filename: Original filename (used to detect content type).
+                  Supported: .webm, .ogg, .wav, .mp3, .m4a
+
+    Returns:
+        Transcribed Arabic text string.
+
+    Raises:
+        RuntimeError: If transcription fails or times out.
+    """
+    headers = {
+        "Authorization": f"Bearer {settings.SPEECHMATICS_API_KEY}",
+    }
+
+    content_type = _get_content_type(filename)
+
+    # STEP 1: Submit transcription job
+    config = json.dumps({
+        "type": "transcription",
+        "transcription_config": {"language": "ar"},
+    })
+
+    files = {
+        "data_file": (filename, audio_bytes, content_type),
+        "config": (None, config, "application/json"),
+    }
+
+    try:
+        response = requests.post(
+            f"{API_BASE}/jobs/",
+            headers=headers,
+            files=files,
+        )
+        response.raise_for_status()
+        job_id = response.json()["id"]
+        logger.info(f"Speechmatics job submitted: {job_id}")
+    except Exception as e:
+        logger.error(f"Failed to submit STT job: {e}")
+        raise RuntimeError(f"Failed to submit transcription job: {e}")
+
+    # STEP 2: Poll for completion
+    max_attempts = 60
+    for attempt in range(max_attempts):
+        time.sleep(1)
+
+        try:
+            response = requests.get(
+                f"{API_BASE}/jobs/{job_id}",
+                headers=headers,
+            )
+            # CRITICAL: Set UTF-8 encoding before reading Arabic text
+            response.encoding = "utf-8"
+            response.raise_for_status()
+
+            job_status = response.json()["job"]["status"]
+            logger.debug(f"Job {job_id} status: {job_status} (attempt {attempt + 1})")
+
+            if job_status == "done":
+                break
+            elif job_status == "rejected":
+                raise RuntimeError(
+                    f"Transcription job {job_id} was rejected by Speechmatics"
+                )
+        except RuntimeError:
+            raise
+        except Exception as e:
+            logger.warning(f"Error polling job status: {e}")
+    else:
+        raise RuntimeError(
+            f"Transcription job {job_id} timed out after {max_attempts} seconds"
+        )
+
+    # STEP 3: Fetch transcript
+    try:
+        response = requests.get(
+            f"{API_BASE}/jobs/{job_id}/transcript",
+            headers={
+                **headers,
+                "Accept": "application/json",
+            },
+        )
+        # CRITICAL: Set UTF-8 encoding before reading Arabic text
+        response.encoding = "utf-8"
+        response.raise_for_status()
+
+        transcript_data = response.json()
+        # Extract text from all results
+        texts = []
+        for result in transcript_data.get("results", []):
+            for alt in result.get("alternatives", []):
+                content = alt.get("content", "")
+                if content:
+                    texts.append(content)
+
+        transcribed_text = " ".join(texts)
+        logger.info(f"Transcription complete: {transcribed_text[:50]}...")
+        return transcribed_text
+
+    except Exception as e:
+        logger.error(f"Failed to fetch transcript: {e}")
+        raise RuntimeError(f"Failed to fetch transcript for job {job_id}: {e}")

services/tts_service.py

@@ -0,0 +1,117 @@
+import logging
+import os
+import tempfile
+
+from gradio_client import Client, handle_file
+
+from config import settings
+
+logger = logging.getLogger(__name__)
+
+# Initialize Gradio client (lazy — connects on first call)
+_gradio_client = None
+
+# Track saved characters
+saved_characters: list[str] = []
+
+
+def _get_client() -> Client:
+    """Lazy-initialize the Gradio TTS client."""
+    global _gradio_client
+    if _gradio_client is None:
+        logger.info(f"Connecting to Gradio TTS at: {settings.GRADIO_TTS_URL}")
+        _gradio_client = Client(settings.GRADIO_TTS_URL)
+    return _gradio_client
+
+
+def save_character(
+    char_name: str,
+    audio_bytes: bytes,
+    audio_filename: str,
+    ref_text: str,
+) -> tuple[str, None] | tuple[None, str]:
+    """
+    Save a new character voice to the Gradio TTS model.
+
+    Args:
+        char_name: Name for the character.
+        audio_bytes: Raw audio bytes of the reference voice clip.
+        audio_filename: Original filename of the audio.
+        ref_text: The reference text spoken in the audio clip.
+
+    Returns:
+        (success_message, None) on success.
+        (None, error_message) on failure.
+    """
+    try:
+        client = _get_client()
+
+        # Write audio bytes to a temp file (Gradio needs a file path)
+        suffix = os.path.splitext(audio_filename)[1] or ".wav"
+        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+            tmp.write(audio_bytes)
+            tmp_path = tmp.name
+
+        try:
+            result = client.predict(
+                char_name=char_name,
+                audio_path=handle_file(tmp_path),
+                ref_text=ref_text,
+                api_name="/save_new_character",
+            )
+
+            # result is (markdown_message, dropdown_options)
+            message = result[0] if isinstance(result, (list, tuple)) else str(result)
+            logger.info(f"Character saved: {char_name} — {message}")
+
+            # Track the saved character
+            if char_name not in saved_characters:
+                saved_characters.append(char_name)
+
+            return message, None
+
+        finally:
+            # Clean up temp file
+            if os.path.exists(tmp_path):
+                os.unlink(tmp_path)
+
+    except Exception as e:
+        logger.error(f"Failed to save character: {e}")
+        global _gradio_client
+        _gradio_client = None
+        return None, f"Failed to save character: {e}"
+
+
+def synthesize_speech(text: str, character_name: str) -> tuple[str, None] | tuple[None, str]:
+    """
+    Synthesize speech using the Gradio TTS API.
+
+    Args:
+        text: The Arabic text to synthesize.
+        character_name: Character name from the saved characters.
+
+    Returns:
+        (filepath, None) on success — path to the generated audio file.
+        (None, error_message) on failure.
+    """
+    try:
+        client = _get_client()
+
+        result = client.predict(
+            text=text,
+            custom_name=character_name,
+            api_name="/generate",
+        )
+
+        # result is a filepath to the generated audio
+        if result and os.path.exists(result):
+            logger.info(f"TTS audio generated: {result}")
+            return result, None
+        else:
+            return None, f"Gradio TTS returned no audio file: {result}"
+
+    except Exception as e:
+        logger.error(f"Gradio TTS error: {e}")
+        global _gradio_client
+        _gradio_client = None
+        return None, f"Gradio TTS failed: {e}"