File size: 3,907 Bytes
224a44e
d0b1d85
996d0ad
224a44e
 
d0b1d85
996d0ad
 
 
d0b1d85
 
 
224a44e
d0b1d85
d638b54
224a44e
d0b1d85
 
224a44e
d0b1d85
224a44e
d0b1d85
 
 
224a44e
 
d0b1d85
 
 
 
 
 
 
 
 
 
996d0ad
 
d0b1d85
 
996d0ad
d0b1d85
224a44e
d0b1d85
 
 
 
 
 
224a44e
d0b1d85
 
 
 
 
 
 
 
 
 
 
 
 
 
224a44e
d0b1d85
224a44e
d0b1d85
224a44e
 
d0b1d85
224a44e
996d0ad
d0b1d85
 
 
 
 
224a44e
996d0ad
d0b1d85
 
 
 
 
 
 
 
 
 
 
996d0ad
224a44e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
996d0ad
d0b1d85
 
996d0ad
224a44e
d0b1d85
 
224a44e
d0b1d85
 
 
996d0ad
d0b1d85
996d0ad
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from fastapi import FastAPI, Query
import uvicorn
import requests
import os
from pydub import AudioSegment
from uuid import uuid4

app = FastAPI()

# βœ… Create directory to store audio files
AUDIO_DIR = "audio_files"
os.makedirs(AUDIO_DIR, exist_ok=True)

# βœ… Replace with your actual TTS API URL
TTS_API_URL = "https://hivecorp-s8test.hf.space/tts"  # Change this


# πŸ”Ή Function: Splitting long text into smaller chunks
def split_text(text, max_length=500):
    """Splits text into chunks of max_length characters"""
    words = text.split()
    chunks = []
    current_chunk = []
    current_length = 0
    
    for word in words:
        if current_length + len(word) + 1 > max_length:
            chunks.append(" ".join(current_chunk))
            current_chunk = []
            current_length = 0
        current_chunk.append(word)
        current_length += len(word) + 1

    if current_chunk:
        chunks.append(" ".join(current_chunk))

    return chunks


# πŸ”Ή Function: Call external TTS API
def call_tts_api(text, voice):
    """Calls external TTS API and returns the saved MP3 filename."""
    try:
        response = requests.post(
            TTS_API_URL,
            json={"text": text, "voice": voice},
            headers={"Content-Type": "application/json"}
        )
        data = response.json()
        
        # βœ… Log response for debugging
        print("TTS API Response:", data)
        
        if "audio_url" in data:
            audio_url = data["audio_url"]
            filename = f"audio_{uuid4().hex}.mp3"
            audio_path = os.path.join(AUDIO_DIR, filename)

            # βœ… Download the MP3 file
            with requests.get(audio_url, stream=True) as r:
                r.raise_for_status()
                with open(audio_path, "wb") as f:
                    for chunk in r.iter_content(chunk_size=8192):
                        f.write(chunk)
            
            return audio_path  # βœ… Return saved file path
        else:
            print("Error: Invalid TTS API response", data)
            return None
    except Exception as e:
        print("TTS API Error:", e)
        return None


# πŸ”Ή Function: Merge multiple MP3s into one
def merge_audio(audio_files):
    """Merges multiple MP3 files into one final MP3."""
    if not audio_files:
        return None

    final_audio = AudioSegment.empty()
    for file in audio_files:
        final_audio += AudioSegment.from_mp3(file)

    final_filename = f"{AUDIO_DIR}/final_{uuid4().hex}.mp3"
    final_audio.export(final_filename, format="mp3")
    
    return final_filename


# βœ… API Endpoint: Generate Audio from Text
@app.post("/generate-audio")
def generate_audio(text: str = Query(...), voice: str = Query(...)):
    """Splits text, calls TTS API, merges audio, and returns the final MP3."""
    chunks = split_text(text)
    print("Total Chunks:", len(chunks))
    
    audio_files = []
    
    for chunk in chunks:
        filename = call_tts_api(chunk, voice)
        if filename:
            audio_files.append(filename)
    
    if not audio_files:
        return {"error": "TTS API failed for all parts"}
    
    final_audio = merge_audio(audio_files)
    
    if final_audio:
        return {"success": True, "audio_url": f"/download/{os.path.basename(final_audio)}"}
    else:
        return {"error": "Failed to merge audio files"}


# βœ… API Endpoint: Download the generated MP3
@app.get("/download/{filename}")
def download_file(filename: str):
    """Allows users to download the generated MP3 file."""
    file_path = os.path.join(AUDIO_DIR, filename)
    if os.path.exists(file_path):
        return {"download_url": f"https://hivecorp-manager1.hf.space/audio_files/{filename}"}
    return {"error": "File not found"}


# βœ… Ensure the app starts when running in Hugging Face Spaces
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)