parkjihye's picture
Update app.py
5bd0edb verified
Raw History Blame Contribute Delete
21.6 kB
# app.py
from fastapi import FastAPI, Request, File, UploadFile, Form
from fastapi.responses import HTMLResponse, FileResponse
from pydantic import BaseModel
import tempfile
import os
import sys
import subprocess
import traceback
import torchaudio
from modelscope import snapshot_download
import threading
# μ „μ—­ 락 객체 생성
tts_lock = threading.Lock()
# ---------------- CosyVoice 경둜 μ„€μ • ----------------
sys.path.append('/app/model')
sys.path.append('/app/model/third_party/Matcha-TTS')
from cosyvoice.cli.cosyvoice import CosyVoice2
from cosyvoice.utils.file_utils import load_wav
# ---------------- μ „μ—­ λ³€μˆ˜ ----------------
cosyvoice_model = None
# ---------------- λͺ¨λΈ μ΄ˆκΈ°ν™” ν•¨μˆ˜ ----------------
def initialize_cosyvoice():
"""CosyVoice2 λͺ¨λΈμ„ μ΄ˆκΈ°ν™”ν•©λ‹ˆλ‹€."""
global cosyvoice_model
try:
print("=== CosyVoice2 λͺ¨λΈ μ΄ˆκΈ°ν™” μ‹œμž‘ ===")
# μž‘μ—… 디렉토리λ₯Ό cosyvoice λͺ¨λ“ˆ μœ„μΉ˜λ‘œ λ³€κ²½
original_cwd = os.getcwd()
cosyvoice_dir = '/app/model/cosyvoice'
print(f"μž‘μ—… 디렉토리 λ³€κ²½: {original_cwd} -> {cosyvoice_dir}")
os.chdir(cosyvoice_dir)
# λͺ¨λΈ 경둜 확인
model_path = '/app/pretrained_models/CosyVoice2-0.5B'
ttsfrd_path = '/app/pretrained_models/CosyVoice-ttsfrd'
resource_path = '/app/pretrained_models/CosyVoice-ttsfrd/resource'
print(f"λͺ¨λΈ 경둜 확인: {model_path}")
print(f"λͺ¨λΈ 경둜 쑴재: {os.path.exists(model_path)}")
print(f"ttsfrd 경둜 확인: {ttsfrd_path}")
print(f"ttsfrd 경둜 쑴재: {os.path.exists(ttsfrd_path)}")
print(f"λ¦¬μ†ŒμŠ€ 경둜 확인: {resource_path}")
print(f"λ¦¬μ†ŒμŠ€ 경둜 쑴재: {os.path.exists(resource_path)}")
if os.path.exists(ttsfrd_path):
print("ttsfrd 디렉토리 λ‚΄μš©:")
for item in os.listdir(ttsfrd_path):
item_path = os.path.join(ttsfrd_path, item)
print(f" {item} ({'dir' if os.path.isdir(item_path) else 'file'})")
if os.path.exists(resource_path):
print("resource 디렉토리 λ‚΄μš©:")
for item in os.listdir(resource_path):
print(f" {item}")
if not os.path.exists(model_path):
print(f"❌ λͺ¨λΈ κ²½λ‘œκ°€ μ‘΄μž¬ν•˜μ§€ μ•ŠμŠ΅λ‹ˆλ‹€: {model_path}")
return False
if not os.path.exists(resource_path):
print(f"❌ λ¦¬μ†ŒμŠ€ κ²½λ‘œκ°€ μ‘΄μž¬ν•˜μ§€ μ•ŠμŠ΅λ‹ˆλ‹€: {resource_path}")
return False
# ROOT_DIR κΈ°μ€€ μƒλŒ€ 경둜 확인
expected_resource_path = os.path.join(os.getcwd(), '../../pretrained_models/CosyVoice-ttsfrd/resource')
normalized_path = os.path.normpath(expected_resource_path)
print(f"CosyVoiceκ°€ μ°ΎλŠ” λ¦¬μ†ŒμŠ€ 경둜: {normalized_path}")
print(f"ν•΄λ‹Ή 경둜 쑴재 μ—¬λΆ€: {os.path.exists(normalized_path)}")
# λͺ¨λΈ λ‘œλ“œ
print("CosyVoice2 λͺ¨λΈ λ‘œλ“œ 쀑...")
cosyvoice_model = CosyVoice2(
model_path,
load_jit=False,
load_trt=False,
fp16=False,
)
# μž‘μ—… 디렉토리 볡원
os.chdir(original_cwd)
print("βœ… CosyVoice2 λͺ¨λΈ μ΄ˆκΈ°ν™” μ™„λ£Œ!")
return True
except Exception as e:
# μž‘μ—… 디렉토리 볡원
try:
os.chdir(original_cwd)
except:
pass
print(f"❌ λͺ¨λΈ μ΄ˆκΈ°ν™” μ‹€νŒ¨: {str(e)}")
traceback.print_exc()
return False
# ---------------- μ„œλ²„ μ‹œμž‘ μ‹œ λͺ¨λΈ μ΄ˆκΈ°ν™” ----------------
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
"""μ„œλ²„ μ‹œμž‘ μ‹œ λͺ¨λΈμ„ μ΄ˆκΈ°ν™”ν•©λ‹ˆλ‹€."""
print("πŸš€ μ„œλ²„ μ‹œμž‘ - λͺ¨λΈ μ΄ˆκΈ°ν™” 쀑...")
initialize_cosyvoice()
yield
# FastAPI 앱에 lifespan 적용
app = FastAPI(
title="CosyVoice2 Korean TTS API",
description="FastAPI + CosyVoice2 기반 ν•œκ΅­μ–΄ μŒμ„± ν•©μ„± μ„œλ²„",
version="1.0.0",
lifespan=lifespan
)
# ---------------- μž…λ ₯/좜λ ₯ λͺ¨λΈ ----------------
class TTSRequest(BaseModel):
text: str
prompt_text: str
class TTSResponse(BaseModel):
status: str
message: str
audio_path: str = None
# ---------------- API: JSON POST ----------------
@app.post("/synthesize", response_model=TTSResponse)
async def synthesize_speech(request: TTSRequest, prompt_audio: UploadFile = File(...)):
"""
μŒμ„± ν•©μ„± API
- text: ν•©μ„±ν•  ν…μŠ€νŠΈ
- prompt_text: ν”„λ‘¬ν”„νŠΈ μŒμ„±μ˜ ν…μŠ€νŠΈ
- prompt_audio: ν”„λ‘¬ν”„νŠΈ μŒμ„± 파일 (wav, mp3, flac λ“±)
"""
if cosyvoice_model is None:
return TTSResponse(
status="error",
message="λͺ¨λΈμ΄ μ΄ˆκΈ°ν™”λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€. μ„œλ²„ 둜그λ₯Ό ν™•μΈν•΄μ£Όμ„Έμš”."
)
try:
# μž„μ‹œ 파일둜 ν”„λ‘¬ν”„νŠΈ μŒμ„± μ €μž₯ (ν™•μž₯자 μœ μ§€)
temp_file_extension = os.path.splitext(prompt_audio.filename)[1].lower()
if not temp_file_extension:
temp_file_extension = '.wav' # κΈ°λ³Έκ°’
with tempfile.NamedTemporaryFile(delete=False, suffix=temp_file_extension) as temp_file:
temp_file.write(await prompt_audio.read())
temp_path = temp_file.name
# ν”„λ‘¬ν”„νŠΈ μŒμ„± λ‘œλ“œ (16kHz)
try:
prompt_speech_16k = load_wav(temp_path, 16000)
except Exception as e:
print(f"load_wav μ‹€νŒ¨: {e}")
# fallback: librosa 직접 μ‚¬μš©
import librosa
import torch
audio_data, sr = librosa.load(temp_path, sr=16000)
prompt_speech_16k = torch.from_numpy(audio_data).unsqueeze(0)
# μŒμ„± ν•©μ„± μ‹€ν–‰
results_generator = cosyvoice_model.inference_zero_shot(
request.text,
prompt_text=request.prompt_text,
prompt_speech_16k=prompt_speech_16k,
text_frontend=True
)
# generatorλ₯Ό 리슀트둜 λ³€ν™˜
results = list(results_generator)
if not results:
return TTSResponse(
status="error",
message="μŒμ„± ν•©μ„± κ²°κ³Όκ°€ λΉ„μ–΄μžˆμŠ΅λ‹ˆλ‹€."
)
# κ²°κ³Ό μ €μž₯ (좜λ ₯ 디렉토리 μ§€μ •)
output_dir = '/app/outputs'
os.makedirs(output_dir, exist_ok=True)
output_filename = f'output_{hash(request.text)}.wav'
output_path = os.path.join(output_dir, output_filename)
torchaudio.save(output_path, results[0]['tts_speech'], cosyvoice_model.sample_rate)
# μž„μ‹œ 파일 정리
os.unlink(temp_path)
return TTSResponse(
status="success",
message="μŒμ„± 합성이 μ™„λ£Œλ˜μ—ˆμŠ΅λ‹ˆλ‹€.",
audio_path=f'outputs/{output_filename}'
)
except Exception as e:
return TTSResponse(
status="error",
message=f"μŒμ„± ν•©μ„± 쀑 였λ₯˜κ°€ λ°œμƒν–ˆμŠ΅λ‹ˆλ‹€: {str(e)}"
)
# ---------------- μ˜€λ””μ˜€ 파일 λ‹€μš΄λ‘œλ“œ ----------------
@app.get("/download/{filepath:path}")
async def download_audio(filepath: str):
"""ν•©μ„±λœ μ˜€λ””μ˜€ νŒŒμΌμ„ λ‹€μš΄λ‘œλ“œν•©λ‹ˆλ‹€."""
full_path = os.path.join('/app', filepath)
if os.path.exists(full_path):
filename = os.path.basename(filepath)
return FileResponse(full_path, media_type="audio/wav", filename=filename)
else:
return {"error": "νŒŒμΌμ„ 찾을 수 μ—†μŠ΅λ‹ˆλ‹€."}
# ---------------- HTML UI ----------------
@app.get("/", response_class=HTMLResponse)
async def main_ui():
return """
<html>
<head>
<title>CosyVoice2 Korean TTS</title>
<meta charset="UTF-8">
<style>
body {
font-family: 'Segoe UI', Arial, sans-serif;
max-width: 900px;
margin: auto;
padding: 2rem;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 100vh;
}
.container {
background-color: white;
padding: 2.5rem;
border-radius: 15px;
box-shadow: 0 10px 30px rgba(0,0,0,0.2);
}
.header {
text-align: center;
margin-bottom: 2rem;
}
.header h1 {
color: #333;
margin-bottom: 0.5rem;
}
.header p {
color: #666;
font-size: 1.1rem;
}
.form-group {
margin-bottom: 1.5rem;
}
label {
display: block;
margin-bottom: 0.5rem;
font-weight: 600;
color: #333;
}
input[type="text"], textarea {
width: 100%;
padding: 0.75rem;
border: 2px solid #e0e0e0;
border-radius: 8px;
font-size: 1rem;
box-sizing: border-box;
transition: border-color 0.3s;
}
input[type="text"]:focus, textarea:focus {
outline: none;
border-color: #667eea;
}
input[type="file"] {
width: 100%;
padding: 0.75rem;
border: 2px dashed #ccc;
border-radius: 8px;
background-color: #f9f9f9;
box-sizing: border-box;
transition: all 0.3s;
}
input[type="file"]:hover {
border-color: #667eea;
background-color: #f0f4ff;
}
input[type="submit"] {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 1rem 2rem;
border: none;
border-radius: 8px;
cursor: pointer;
font-size: 1.1rem;
font-weight: 600;
transition: transform 0.2s;
width: 100%;
}
input[type="submit"]:hover {
transform: translateY(-2px);
}
.info {
background: linear-gradient(135deg, #e3f2fd 0%, #f3e5f5 100%);
padding: 1.5rem;
border-radius: 10px;
margin-bottom: 2rem;
border-left: 4px solid #667eea;
}
.example {
background-color: #f8f9fa;
padding: 1rem;
border-radius: 8px;
margin-top: 0.5rem;
border-left: 3px solid #28a745;
}
.example strong {
color: #28a745;
}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>🎀 CosyVoice2 μŒμ„± ν•©μ„±κΈ°</h1>
<p>ν•œκ΅­μ–΄ ν…μŠ€νŠΈλ₯Ό μžμ—°μŠ€λŸ¬μš΄ μŒμ„±μœΌλ‘œ λ³€ν™˜ν•΄λ³΄μ„Έμš”!</p>
</div>
<div class="info">
<strong>πŸ“‹ μ‚¬μš© 방법:</strong><br>
1. <strong>ν”„λ‘¬ν”„νŠΈ μŒμ„±:</strong> λͺ©μ†Œλ¦¬ μŠ€νƒ€μΌμ˜ 기쀀이 될 μŒμ„± νŒŒμΌμ„ μ—…λ‘œλ“œν•˜μ„Έμš”<br>
2. <strong>ν”„λ‘¬ν”„νŠΈ ν…μŠ€νŠΈ:</strong> μ—…λ‘œλ“œν•œ μŒμ„±μ˜ μ‹€μ œ λ‚΄μš©μ„ μž…λ ₯ν•˜μ„Έμš”<br>
3. <strong>ν•©μ„±ν•  ν…μŠ€νŠΈ:</strong> μƒˆλ‘œ μƒμ„±ν•˜κ³  싢은 μŒμ„±μ˜ ν…μŠ€νŠΈλ₯Ό μž…λ ₯ν•˜μ„Έμš”<br><br>
<strong>지원 ν˜•μ‹:</strong> WAV
</div>
<form action="/submit" method="post" enctype="multipart/form-data">
<div class="form-group">
<label for="prompt_audio">🎡 ν”„λ‘¬ν”„νŠΈ μŒμ„± 파일:</label>
<input type="file" id="prompt_audio" name="prompt_audio" accept=".wav" required>
<div class="example">
<strong>μ˜ˆμ‹œ:</strong> "μ•ˆλ…•ν•˜μ„Έμš”"라고 λ§ν•˜λŠ” μŒμ„± 파일
</div>
</div>
<div class="form-group">
<label for="prompt_text">πŸ“ ν”„λ‘¬ν”„νŠΈ ν…μŠ€νŠΈ:</label>
<input type="text" id="prompt_text" name="prompt_text"
placeholder="μ—…λ‘œλ“œν•œ μŒμ„±μ˜ μ‹€μ œ λ‚΄μš©"
value="μ•ˆλ…•ν•˜μ„Έμš”" required>
<div class="example">
<strong>μ˜ˆμ‹œ:</strong> μ•ˆλ…•ν•˜μ„Έμš” (μ—…λ‘œλ“œν•œ μŒμ„± 파일의 μ‹€μ œ λ‚΄μš©)
</div>
</div>
<div class="form-group">
<label for="text">🎯 ν•©μ„±ν•  ν…μŠ€νŠΈ:</label>
<textarea id="text" name="text" rows="3"
placeholder="μƒˆλ‘œ μƒμ„±ν•˜κ³  싢은 μŒμ„±μ˜ ν…μŠ€νŠΈλ₯Ό μž…λ ₯ν•˜μ„Έμš”"
required>곡룑이 밀양갱을 λͺ°λž˜ λ¨Ήκ³  λ„λ§μ³€μ–΄μš”.</textarea>
<div class="example">
<strong>μ˜ˆμ‹œ:</strong> 곡룑이 밀양갱을 λͺ°λž˜ λ¨Ήκ³  λ„λ§μ³€μ–΄μš”.
</div>
</div>
<input type="submit" value="πŸš€ μŒμ„± ν•©μ„± μ‹œμž‘">
</form>
</div>
</body>
</html>
"""
# ---------------- κ²°κ³Ό λ Œλ”λ§ ----------------
@app.post("/submit", response_class=HTMLResponse)
async def handle_form(
request: Request,
text: str = Form(...),
prompt_text: str = Form(...),
prompt_audio: UploadFile = File(...)
):
try:
if cosyvoice_model is None:
return """
<html>
<head><title>μ—λŸ¬</title><meta charset="UTF-8"></head>
<body style="font-family: Arial, sans-serif; max-width: 600px; margin: auto; padding: 2rem;">
<h1>❌ λͺ¨λΈ μ΄ˆκΈ°ν™” 였λ₯˜</h1>
<p>CosyVoice2 λͺ¨λΈμ΄ 아직 μ΄ˆκΈ°ν™”λ˜μ§€ μ•Šμ•˜μŠ΅λ‹ˆλ‹€.</p>
<p>μ„œλ²„ 둜그λ₯Ό ν™•μΈν•˜κ³  μž μ‹œ ν›„ λ‹€μ‹œ μ‹œλ„ν•΄μ£Όμ„Έμš”.</p>
<br>
<a href="/" style="color: #667eea; text-decoration: none;">← λŒμ•„κ°€κΈ°</a>
</body>
</html>
"""
# 파일 ν˜•μ‹ 검증
if not prompt_audio.filename.lower().endswith('.wav'):
return """
<html>
<head><title>μ—λŸ¬</title><meta charset="UTF-8"></head>
<body style="font-family: Arial, sans-serif; max-width: 600px; margin: auto; padding: 2rem;">
<h1>❌ 파일 ν˜•μ‹ 였λ₯˜</h1>
<p>WAV 파일만 μ§€μ›λ©λ‹ˆλ‹€.</p>
<p><strong>지원 ν˜•μ‹:</strong> WAV</p>
<br>
<a href="/" style="color: #667eea; text-decoration: none;">← λŒμ•„κ°€κΈ°</a>
</body>
</html>
"""
# μž„μ‹œ 파일둜 ν”„λ‘¬ν”„νŠΈ μŒμ„± μ €μž₯
temp_file_extension = os.path.splitext(prompt_audio.filename)[1].lower()
if not temp_file_extension:
temp_file_extension = '.wav' # κΈ°λ³Έκ°’
with tempfile.NamedTemporaryFile(delete=False, suffix=temp_file_extension) as temp_file:
temp_file.write(await prompt_audio.read())
temp_path = temp_file.name
print(f"μ—…λ‘œλ“œλœ 파일: {prompt_audio.filename}")
print(f"μž„μ‹œ 파일 경둜: {temp_path}")
print(f"파일 크기: {os.path.getsize(temp_path)} bytes")
# ν”„λ‘¬ν”„νŠΈ μŒμ„± λ‘œλ“œ (16kHz) - 더 μ•ˆμ „ν•œ λ°©λ²•μœΌλ‘œ
try:
prompt_speech_16k = load_wav(temp_path, 16000)
print(f"μ˜€λ””μ˜€ λ‘œλ“œ 성곡: shape={prompt_speech_16k.shape}")
except Exception as e:
print(f"load_wav μ‹€νŒ¨: {e}")
# fallback: librosa 직접 μ‚¬μš©
import librosa
import torch
audio_data, sr = librosa.load(temp_path, sr=16000)
prompt_speech_16k = torch.from_numpy(audio_data).unsqueeze(0)
print(f"librosa fallback 성곡: shape={prompt_speech_16k.shape}")
# μŒμ„± ν•©μ„± μ‹€ν–‰
print(f"μŒμ„± ν•©μ„± μ‹œμž‘: text='{text}', prompt_text='{prompt_text}'")
results_generator = cosyvoice_model.inference_zero_shot(
text,
prompt_text=prompt_text,
prompt_speech_16k=prompt_speech_16k,
text_frontend=True
)
# generatorλ₯Ό 리슀트둜 λ³€ν™˜
results = list(results_generator)
print(f"μŒμ„± ν•©μ„± μ™„λ£Œ! κ²°κ³Ό 개수: {len(results)}")
if not results:
raise Exception("μŒμ„± ν•©μ„± κ²°κ³Όκ°€ λΉ„μ–΄μžˆμŠ΅λ‹ˆλ‹€.")
# κ²°κ³Ό μ €μž₯ (좜λ ₯ 디렉토리 μ§€μ •)
output_dir = '/app/outputs'
os.makedirs(output_dir, exist_ok=True)
output_filename = f'korean_tts_output_{hash(text)}.wav'
output_path = os.path.join(output_dir, output_filename)
torchaudio.save(output_path, results[0]['tts_speech'], cosyvoice_model.sample_rate)
print(f"μ˜€λ””μ˜€ 파일 μ €μž₯ μ™„λ£Œ: {output_path}")
# λ‹€μš΄λ‘œλ“œμš© μƒλŒ€ 경둜
download_filename = f'outputs/{output_filename}'
# μž„μ‹œ 파일 정리
os.unlink(temp_path)
except Exception as e:
error_details = traceback.format_exc()
return f"""
<html>
<head><title>μ—λŸ¬</title><meta charset="UTF-8"></head>
<body style="font-family: Arial, sans-serif; max-width: 700px; margin: auto; padding: 2rem;">
<h1>❌ μ„œλ²„ 였λ₯˜ λ°œμƒ</h1>
<p><strong>였λ₯˜ λ©”μ‹œμ§€:</strong></p>
<pre style="background-color: #f8f9fa; padding: 1rem; border-radius: 5px; overflow-x: auto;">{str(e)}</pre>
<hr>
<details>
<summary><strong>μ—λŸ¬ 상세 (ν΄λ¦­ν•˜μ—¬ 펼치기)</strong></summary>
<pre style="background-color: #f8f9fa; padding: 1rem; border-radius: 5px; overflow-x: auto;">{error_details}</pre>
</details>
<br>
<a href="/" style="color: #667eea; text-decoration: none;">← λŒμ•„κ°€κΈ°</a>
</body>
</html>
"""
return f"""
<html>
<head><title>ν•©μ„± κ²°κ³Ό</title><meta charset="UTF-8"></head>
<body style="font-family: Arial, sans-serif; max-width: 700px; margin: auto; padding: 2rem;">
<h1>βœ… μŒμ„± ν•©μ„± μ™„λ£Œ!</h1>
<div style="background: linear-gradient(135deg, #e3f2fd 0%, #f3e5f5 100%); padding: 1.5rem; border-radius: 10px; margin: 1.5rem 0;">
<h3>πŸ“‹ μž…λ ₯ 정보</h3>
<p><strong>ν”„λ‘¬ν”„νŠΈ μŒμ„±:</strong> {prompt_audio.filename}</p>
<p><strong>ν”„λ‘¬ν”„νŠΈ ν…μŠ€νŠΈ:</strong> {prompt_text}</p>
<p><strong>ν•©μ„±ν•  ν…μŠ€νŠΈ:</strong> {text}</p>
</div>
<div style="background-color: #f8f9fa; padding: 1.5rem; border-radius: 10px; border-left: 4px solid #28a745;">
<h3>🎡 ν•©μ„±λœ μŒμ„±</h3>
<audio controls style="width: 100%; margin: 1rem 0;">
<source src="/download/{download_filename}" type="audio/wav">
λΈŒλΌμš°μ €κ°€ μ˜€λ””μ˜€λ₯Ό μ§€μ›ν•˜μ§€ μ•ŠμŠ΅λ‹ˆλ‹€.
</audio>
<br>
<a href="/download/{download_filename}"
style="background: linear-gradient(135deg, #28a745 0%, #20c997 100%);
color: white; padding: 0.75rem 1.5rem; text-decoration: none;
border-radius: 8px; display: inline-block; margin-top: 1rem;">
πŸ“₯ 파일 λ‹€μš΄λ‘œλ“œ
</a>
</div>
<br>
<a href="/" style="color: #667eea; text-decoration: none; font-size: 1.1rem;">← λ‹€μ‹œ μ‹œλ„ν•˜κΈ°</a>
</body>
</html>
"""
# ---------------- ν—¬μŠ€ 체크 ----------------
@app.get("/health")
async def health_check():
return {
"status": "ok" if cosyvoice_model is not None else "initializing",
"model_loaded": cosyvoice_model is not None,
"description": "CosyVoice2 Korean TTS Server"
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)