Spaces:
Running
Running
| import os | |
| import time | |
| import hmac | |
| import hashlib | |
| import edge_tts | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import StreamingResponse, JSONResponse | |
| from fastapi.middleware.cors import CORSMiddleware | |
| app = FastAPI(title="VCLOUD Edge TTS API") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| SECRET_KEY = os.environ.get("API_SECRET_KEY") | |
| DEFAULT_VOICE = "vi-VN-HoaiMyNeural" | |
| async def read_tts_payload(request: Request): | |
| text = "" | |
| voice = DEFAULT_VOICE | |
| bot_uid = "" | |
| req_time = "0" | |
| token = "" | |
| if request.method == "POST": | |
| try: | |
| data = await request.json() | |
| except Exception: | |
| data = {} | |
| text = str(data.get("text", "")).strip() | |
| voice = str(data.get("voice", voice)).strip() or DEFAULT_VOICE | |
| bot_uid = str(data.get("bot_uid", "")).strip() | |
| req_time = str(data.get("time", "0")).strip() | |
| token = str(data.get("token", "")).strip() | |
| else: | |
| text = request.query_params.get("text", "").strip() | |
| voice = request.query_params.get("voice", voice).strip() or DEFAULT_VOICE | |
| bot_uid = request.query_params.get("bot_uid", "").strip() | |
| req_time = request.query_params.get("time", "0").strip() | |
| token = request.query_params.get("token", "").strip() | |
| return text, voice, bot_uid, req_time, token | |
| async def root(): | |
| return JSONResponse({"status": "VCLOUD Voice API is running optimally on Hugging Face!"}) | |
| async def healthcheck(): | |
| return JSONResponse({"status": "ok", "message": "Trạm Voice đang hoạt động xuất sắc!"}) | |
| async def generate_tts(request: Request): | |
| text, voice, bot_uid, req_time, token = await read_tts_payload(request) | |
| if not text: | |
| return JSONResponse({"status": "success", "message": "Endpoint TTS đã sẵn sàng nhận văn bản."}, status_code=200) | |
| try: | |
| if not SECRET_KEY: | |
| return JSONResponse({"error": "Lỗi máy chủ: Chưa cấu hình Secret Key (API_SECRET_KEY) trong Settings!"}, status_code=500) | |
| current_time = int(time.time()) | |
| if current_time - int(req_time) > 43200: | |
| return JSONResponse({"error": "Truy cập bị từ chối! Token đã hết hạn."}, status_code=401) | |
| message = f"{bot_uid}{req_time}".encode('utf-8') | |
| expected_token = hmac.new(SECRET_KEY.encode('utf-8'), message, hashlib.sha256).hexdigest() | |
| if not hmac.compare_digest(expected_token, token): | |
| return JSONResponse({"error": "Truy cập bị từ chối! Sai Token bảo mật (HMAC không khớp)."}, status_code=403) | |
| except Exception as e: | |
| return JSONResponse({"error": f"Lỗi xác thực bảo mật: {str(e)}"}, status_code=401) | |
| try: | |
| communicate = edge_tts.Communicate(text, voice) | |
| async def audio_stream(): | |
| async for chunk in communicate.stream(): | |
| if chunk["type"] == "audio": | |
| yield chunk["data"] | |
| return StreamingResponse(audio_stream(), media_type="audio/mpeg") | |
| except Exception as e: | |
| return JSONResponse({"error": f"Lỗi quá trình tạo âm thanh: {str(e)}"}, status_code=500) |