merge

.env_template

@@ -1,3 +1,5 @@
 TYPHOON_CHAT_API = *
 DEBUG_MODE = false
-OPENTHAIGPT_CHAT_API = *
+OPENTHAIGPT_CHAT_API = *
+BOTNOI_API_TOKEN = *
+VAJA9_API_KEY = *

.gitignore

@@ -1,9 +1,11 @@
+__pycache__/
+**/__pycache__/
+.ipynb_checkpoints
+.env
 .venv
-.vscode
 .env
-**/__pycache__/
-.pytest_cache
+__pycache__
 
-# temporary files
-runtime_log.csv
-speedtest.py    
+# Files
+speedtest.py
+*.csv

tts/tts.py

@@ -0,0 +1,186 @@
+from fastapi import FastAPI, HTTPException
+from pythainlp.tokenize import sent_tokenize
+from pydantic import BaseModel , Field
+import requests
+from fastapi.responses import FileResponse
+import uuid
+from dotenv import load_dotenv
+import os
+
+# Load environment variables
+load_dotenv()
+
+app = FastAPI()
+
+# Pydantic model for input validation
+class VoiceRequest(BaseModel):
+    text: str
+    audio_id: str = "EUOJF"
+    speaker: str = "52"
+    volume: int = 100
+    speed: float = 1
+    type_media: str = "mp3"
+    language: str = "th"
+    token: str = os.getenv("BOTNOI_API_TOKEN")
+
+# Function to split text for text delay
+def auto_generate_text_delay_with_pythainlp(text):
+    text_delay = sent_tokenize(text, engine="thaisum")
+    text_delay = " ".join(text_delay).strip()
+    return text_delay
+
+# Function to call Botnoi's API to generate voice
+def generate_voice(audio_id, text, text_delay, speaker, volume, speed, type_media, language, token):
+    url = "https://api-genvoice.botnoi.ai/voice/v1/generate_voice?provider=botnoivoice"
+    headers = {
+        "Accept": "application/json, text/plain, */*",
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "audio_id": audio_id,
+        "text": text,
+        "text_delay": text_delay,
+        "speaker": speaker,
+        "volume": str(volume),
+        "speed": str(speed),
+        "type_media": type_media,
+        "language": language,
+    }
+
+    response = requests.post(url, headers=headers, json=payload)
+    if response.status_code == 200:
+        data = response.json()
+        if "data" in data:
+            return data["data"]  # URL of the generated audio
+        else:
+            raise HTTPException(status_code=500, detail=data.get("message", "Unknown error"))
+    else:
+        raise HTTPException(status_code=response.status_code, detail="Voice generation failed")
+
+# Function to download MP3 from a URL
+def download_mp3(url, output_path):
+    headers = {
+        "Accept-Encoding": "identity;q=1, *;q=0",
+        "Range": "bytes=0-",
+        "Referer": "https://voice.botnoi.ai/",
+    }
+
+    response = requests.get(url, headers=headers, stream=True)
+    if response.status_code == 200:
+        with open(output_path, "wb") as file:
+            for chunk in response.iter_content(chunk_size=1024):
+                if chunk:  # Filter out keep-alive chunks
+                    file.write(chunk)
+    else:
+        raise HTTPException(status_code=response.status_code, detail="Failed to download MP3")
+
+# FastAPI endpoint to generate and download voice
+@app.post("/generate_voice_botnoi/")
+def generate_voice_endpoint(request: VoiceRequest):
+    text_delay = auto_generate_text_delay_with_pythainlp(request.text)
+    audio_url = generate_voice(
+        audio_id=request.audio_id,
+        text=request.text,
+        text_delay=text_delay,
+        speaker=request.speaker,
+        volume=request.volume,
+        speed=request.speed,
+        type_media=request.type_media,
+        language=request.language,
+        token=request.token,
+    )
+    
+    # Generate unique filename for the MP3
+    output_file = f"{uuid.uuid4()}.mp3"
+    download_mp3(audio_url, output_file)
+    
+    return FileResponse(output_file, media_type="audio/mpeg", filename="output.mp3")
+
+# -----------------------------------------------------------VAJA9-----------------------------------------------------------
+# VAJA9 Voice Generation
+class Vaja9Request(BaseModel):
+    text: str
+    speaker: int = 1
+    phrase_break: int = 0
+    audiovisual: int = 0
+
+def split_text_into_chunks(text: str, chunk_size: int = 20) -> list:
+    words = text.split()
+    return [' '.join(words[i:i + chunk_size]) for i in range(0, len(words), chunk_size)]
+
+def generate_vaja9_voice(text: str, speaker: int, phrase_break: int, audiovisual: int):
+    url = 'https://api.aiforthai.in.th/vaja9/synth_audiovisual'
+    headers = {
+        'Apikey': os.getenv("VAJA9_API_KEY"),
+        'Content-Type': 'application/json'
+    }
+    data = {
+        'input_text': text,
+        'speaker': speaker, 
+        'phrase_break': phrase_break,
+        'audiovisual': audiovisual
+    }
+    
+    try:
+        response = requests.post(url, json=data, headers=headers, timeout=60)  # Increased timeout to 60 seconds
+        if response.status_code == 200:
+            return response.json()['wav_url']
+        elif response.status_code == 502:
+            raise HTTPException(status_code=502, detail="Bad Gateway - The server received an invalid response from the upstream server")
+        else:
+            raise HTTPException(status_code=response.status_code, detail="Voice generation failed")
+    except requests.exceptions.ReadTimeout:
+        raise HTTPException(status_code=504, detail="Gateway Timeout - The server took too long to respond")
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(status_code=502, detail=f"Bad Gateway - Connection error: {str(e)}")
+
+def download_vaja9_wav(url: str, output_path: str):
+    headers = {'Apikey': os.getenv("VAJA9_API_KEY")}
+    try:
+        response = requests.get(url, headers=headers, timeout=60)  # Increased timeout to 60 seconds
+        if response.status_code == 200:
+            with open(output_path, 'wb') as file:
+                file.write(response.content)
+        elif response.status_code == 502:
+            raise HTTPException(status_code=502, detail="Bad Gateway - The server received an invalid response from the upstream server")
+        else:
+            raise HTTPException(status_code=response.status_code, detail="Failed to download WAV")
+    except requests.exceptions.ReadTimeout:
+        raise HTTPException(status_code=504, detail="Gateway Timeout - The server took too long to respond")
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(status_code=502, detail=f"Bad Gateway - Connection error: {str(e)}")
+
+@app.post("/generate_voice_vaja9/")
+def generate_voice_vaja9_endpoint(request: Vaja9Request):
+    try:
+        # Split text into chunks of 20 words if needed
+        text_chunks = split_text_into_chunks(request.text)
+        output_files = []
+        
+        # Process each chunk
+        for chunk in text_chunks:
+            audio_url = generate_vaja9_voice(
+                text=chunk,
+                speaker=request.speaker,
+                phrase_break=request.phrase_break,
+                audiovisual=request.audiovisual
+            )
+            
+            # Generate unique filename for each chunk
+            output_file = f"{uuid.uuid4()}.wav"
+            download_vaja9_wav(audio_url, output_file)
+            output_files.append(output_file)
+        
+        # If only one chunk, return it directly
+        if len(output_files) == 1:
+            return FileResponse(output_files[0], media_type="audio/wav", filename="output.wav")
+            
+        # TODO: If multiple chunks, they should be combined into a single audio file
+        # For now, return the first chunk
+        return FileResponse(output_files[0], media_type="audio/wav", filename="output.wav")
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=502, detail=f"Bad Gateway - Unexpected error: {str(e)}")