# 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 """ CosyVoice2 Korean TTS

🎤 CosyVoice2 음성 합성기

한국어 텍스트를 자연스러운 음성으로 변환해보세요!

📋 사용 방법:
1. 프롬프트 음성: 목소리 스타일의 기준이 될 음성 파일을 업로드하세요
2. 프롬프트 텍스트: 업로드한 음성의 실제 내용을 입력하세요
3. 합성할 텍스트: 새로 생성하고 싶은 음성의 텍스트를 입력하세요

지원 형식: WAV
예시: "안녕하세요"라고 말하는 음성 파일
예시: 안녕하세요 (업로드한 음성 파일의 실제 내용)
예시: 공룡이 밤양갱을 몰래 먹고 도망쳤어요.
""" # ---------------- 결과 렌더링 ---------------- @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 """ 에러

❌ 모델 초기화 오류

CosyVoice2 모델이 아직 초기화되지 않았습니다.

서버 로그를 확인하고 잠시 후 다시 시도해주세요.


← 돌아가기 """ # 파일 형식 검증 if not prompt_audio.filename.lower().endswith('.wav'): return """ 에러

❌ 파일 형식 오류

WAV 파일만 지원됩니다.

지원 형식: WAV


← 돌아가기 """ # 임시 파일로 프롬프트 음성 저장 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""" 에러

❌ 서버 오류 발생

오류 메시지:

{str(e)}
에러 상세 (클릭하여 펼치기) 
{error_details}

← 돌아가기 """ return f""" 합성 결과

✅ 음성 합성 완료!

📋 입력 정보

프롬프트 음성: {prompt_audio.filename}

프롬프트 텍스트: {prompt_text}

합성할 텍스트: {text}

🎵 합성된 음성


📥 파일 다운로드

← 다시 시도하기 """ # ---------------- 헬스 체크 ---------------- @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)