Update app.py

app.py

@@ -1,317 +1,317 @@
-"""
-Hugging Face Spaces 入口點
-基於 main.py 但針對 Hugging Face Spaces 進行優化
-"""
-
-from fastapi import FastAPI, HTTPException, Query, Request
-from fastapi.responses import FileResponse, JSONResponse
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.staticfiles import StaticFiles
-from pydantic import BaseModel
-from contextlib import asynccontextmanager
-import edge_tts
-import asyncio
-import os
-import uuid
-from typing import Optional, List
-import aiofiles
-import json
-from urllib.parse import urlparse
-import tempfile
-import shutil
-
-# 創建臨時輸出目錄
-OUTPUT_DIR = tempfile.mkdtemp(prefix="edge_tts_")
-os.makedirs(OUTPUT_DIR, exist_ok=True)
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    # 啟動時執行
-    print("===== Application Startup =====")
-    yield
-    # 關閉時執行
-    print("===== Application Shutdown =====")
-    try:
-        if os.path.exists(OUTPUT_DIR):
-            shutil.rmtree(OUTPUT_DIR)
-            print("✅ 清理臨時文件完成")
-    except Exception as e:
-        print(f"⚠️ 清理臨時文件時發生錯誤: {e}")
-
-app = FastAPI(
-    title="Edge TTS API",
-    description="A web service for text-to-speech using Microsoft Edge TTS",
-    version="1.0.0",
-    lifespan=lifespan
-)
-
-# 添加 CORS 中間件（允許所有來源以方便測試）
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# 允許的來源網址清單
-ALLOWED_ORIGINS = [
-    "https://n-m2azgihk7rtcqqaxywntv75at3thf3bgps4xdlq-0lu-script.googleusercontent.com",
-    "https://www.dfes.ntpc.edu.tw",
-    "https://www.dfes.ntpc.edu.tw/"
-]
-
-# 如果希望允許所有 GAS，取消下面兩行的註解
-# "https://script.google.com",  # 所有 Google Apps Script
-
-# 掛載靜態文件（如果存在）
-if os.path.exists("static"):
-    app.mount("/static", StaticFiles(directory="static"), name="static")
-
-class TTSRequest(BaseModel):
-    text: str
-    voice: Optional[str] = "zh-TW-HsiaoChenNeural"
-    rate: Optional[str] = "+0%"
-    volume: Optional[str] = "+0%"
-    pitch: Optional[str] = "+0Hz"
-
-class TTSResponse(BaseModel):
-    success: bool
-    message: str
-    audio_url: Optional[str] = None
-    error: Optional[str] = None
-
-# 來源檢查函數
-def _is_origin_allowed(request: Request) -> bool:
-    """檢查請求來源是否被允許"""
-    origin = request.headers.get("origin")
-    referer = request.headers.get("referer")
-    user_agent = request.headers.get("user-agent", "")
-    
-    print(f"請求來源檢查 - Origin: {origin}, Referer: {referer}, User-Agent: {user_agent}")
-    
-    # 檢查允許的來源（一般網頁）
-    if origin:
-        for allowed_origin in ALLOWED_ORIGINS:
-            if origin.startswith(allowed_origin):
-                print(f"✅ Origin 匹配: {origin} 匹配 {allowed_origin}")
-                return True
-    
-    if referer:
-        for allowed_origin in ALLOWED_ORIGINS:
-            if referer.startswith(allowed_origin):
-                print(f"✅ Referer 匹配: {referer} 匹配 {allowed_origin}")
-                return True
-    
-    # 允許來自 Hugging Face Spaces 的直接調用
-    if "huggingface" in user_agent.lower():
-        print("✅ 檢測到 Hugging Face Spaces 環境")
-        return True
-    
-    print(f"❌ 請求被拒絕 - Origin: {origin}, Referer: {referer}")
-    return False
-
-@app.get("/")
-async def root():
-    """根路徑，返回 API 信息"""
-    return {
-        "message": "Edge TTS API Service - Hugging Face Spaces",
-        "version": "1.0.0",
-        "endpoints": {
-            "GET /voices": "獲取所有可用語音",
-            "POST /tts": "文字轉語音",
-            "GET /tts": "文字轉語音 (GET 方法)",
-            "GET /health": "健康檢查",
-            "GET /allowed-origins": "查看允許的來源網址",
-            "GET /debug-request": "調試請求信息"
-        },
-        "note": "此服務部署在 Hugging Face Spaces 上，僅允許特定來源網址訪問"
-    }
-
-@app.get("/health")
-async def health_check():
-    """健康檢查端點"""
-    return {"status": "healthy", "service": "edge-tts-api", "platform": "huggingface-spaces"}
-
-@app.get("/allowed-origins")
-async def get_allowed_origins():
-    """獲取允許的來源列表（僅供管理員查看）"""
-    return {
-        "allowed_origins": ALLOWED_ORIGINS,
-        "count": len(ALLOWED_ORIGINS),
-        "description": "允許的來源網址清單，支援擴充匹配"
-    }
-
-@app.get("/debug-request")
-async def debug_request(request: Request):
-    """調試端點：顯示請求的詳細信息"""
-    headers = dict(request.headers)
-    return {
-        "method": request.method,
-        "url": str(request.url),
-        "headers": headers,
-        "origin": headers.get("origin"),
-        "referer": headers.get("referer"),
-        "user_agent": headers.get("user-agent"),
-        "is_allowed": _is_origin_allowed(request),
-        "client_ip": request.client.host if request.client else "unknown",
-        "request_info": {
-            "has_origin": bool(headers.get("origin")),
-            "has_referer": bool(headers.get("referer")),
-            "is_huggingface": "huggingface" in headers.get("user-agent", "").lower(),
-            "api_key_provided": "x-api-key" in headers
-        }
-    }
-
-
-@app.get("/voices")
-async def get_voices():
-    """獲取所有可用的語音列表"""
-    try:
-        voices = await edge_tts.list_voices()
-        return {
-            "success": True,
-            "voices": voices,
-            "count": len(voices)
-        }
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"獲取語音列表失敗: {str(e)}")
-
-@app.post("/tts", response_model=TTSResponse)
-async def text_to_speech(request: TTSRequest, http_request: Request):
-    """文字轉語音 API (POST 方法)"""
-    try:
-        # 檢查來源是否被允許
-        if not _is_origin_allowed(http_request):
-            return TTSResponse(
-                success=False, 
-                message="來源網站未被允許使用此 API", 
-                error="forbidden"
-            )
-        
-        # 生成唯一文件名
-        file_id = str(uuid.uuid4())
-        output_file = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
-        
-        # 創建 TTS 通信對象
-        communicate = edge_tts.Communicate(
-            text=request.text,
-            voice=request.voice,
-            rate=request.rate,
-            volume=request.volume,
-            pitch=request.pitch
-        )
-        
-        # 生成語音文件
-        await communicate.save(output_file)
-        
-        # 檢查文件是否成功創建
-        if not os.path.exists(output_file):
-            raise Exception("語音文件生成失敗")
-        
-        return TTSResponse(
-            success=True,
-            message="語音生成成功",
-            audio_url=f"/audio/{file_id}.mp3"
-        )
-        
-    except Exception as e:
-        return TTSResponse(
-            success=False,
-            message="語音生成失敗",
-            error=str(e)
-        )
-
-@app.get("/tts")
-async def text_to_speech_get(
-    text: str = Query(..., description="要轉換的文字"),
-    voice: str = Query("zh-TW-HsiaoChenNeural", description="語音名稱"),
-    rate: str = Query("+0%", description="語速調整"),
-    volume: str = Query("+0%", description="音量調整"),
-    pitch: str = Query("+0Hz", description="音調調整"),
-    http_request: Request = None
-):
-    """文字轉語音 API (GET 方法)"""
-    try:
-        print(f"TTS 請求: text={text}, voice={voice}, rate={rate}, volume={volume}, pitch={pitch}")
-        
-        # 檢查來源是否被允許
-        if http_request and not _is_origin_allowed(http_request):
-            raise HTTPException(
-                status_code=403, 
-                detail="來源網站未被允許使用此 API"
-            )
-        
-        # 生成唯一文件名
-        file_id = str(uuid.uuid4())
-        output_file = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
-        print(f"輸出文件路徑: {output_file}")
-        
-        # 創建 TTS 通信對象
-        communicate = edge_tts.Communicate(
-            text=text,
-            voice=voice,
-            rate=rate,
-            volume=volume,
-            pitch=pitch
-        )
-        
-        # 生成語音文件
-        print("開始生成語音文件...")
-        await communicate.save(output_file)
-        print("語音文件生成完成")
-        
-        # 檢查文件是否成功創建
-        if not os.path.exists(output_file):
-            print(f"文件不存在: {output_file}")
-            raise HTTPException(status_code=500, detail="語音文件生成失敗")
-        
-        file_size = os.path.getsize(output_file)
-        print(f"文件大小: {file_size} bytes")
-        
-        # 返回音頻文件
-        return FileResponse(
-            output_file,
-            media_type="audio/mpeg",
-            filename=f"tts_{file_id}.mp3"
-        )
-        
-    except Exception as e:
-        print(f"TTS 錯誤: {str(e)}")
-        import traceback
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"語音生成失敗: {str(e)}")
-
-@app.get("/audio/{file_id}.mp3")
-async def get_audio_file(file_id: str):
-    """獲取生成的音頻文件"""
-    file_path = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
-    
-    if not os.path.exists(file_path):
-        raise HTTPException(status_code=404, detail="音頻文件不存在")
-    
-    return FileResponse(
-        file_path,
-        media_type="audio/mpeg",
-        filename=f"tts_{file_id}.mp3"
-    )
-
-@app.delete("/audio/{file_id}.mp3")
-async def delete_audio_file(file_id: str):
-    """刪除音頻文件"""
-    file_path = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
-    
-    if not os.path.exists(file_path):
-        raise HTTPException(status_code=404, detail="音頻文件不存在")
-    
-    try:
-        os.remove(file_path)
-        return {"success": True, "message": "文件刪除成功"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"文件刪除失敗: {str(e)}")
-
-# 清理函數已移至 lifespan 處理器中
-
-if __name__ == "__main__":
-    import uvicorn
+"""
+Hugging Face Spaces 入口點
+基於 main.py 但針對 Hugging Face Spaces 進行優化
+"""
+
+from fastapi import FastAPI, HTTPException, Query, Request
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+from contextlib import asynccontextmanager
+import edge_tts
+import asyncio
+import os
+import uuid
+from typing import Optional, List
+import aiofiles
+import json
+from urllib.parse import urlparse
+import tempfile
+import shutil
+
+# 創建臨時輸出目錄
+OUTPUT_DIR = tempfile.mkdtemp(prefix="edge_tts_")
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # 啟動時執行
+    print("===== Application Startup =====")
+    yield
+    # 關閉時執行
+    print("===== Application Shutdown =====")
+    try:
+        if os.path.exists(OUTPUT_DIR):
+            shutil.rmtree(OUTPUT_DIR)
+            print("✅ 清理臨時文件完成")
+    except Exception as e:
+        print(f"⚠️ 清理臨時文件時發生錯誤: {e}")
+
+app = FastAPI(
+    title="Edge TTS API",
+    description="A web service for text-to-speech using Microsoft Edge TTS",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# 添加 CORS 中間件（允許所有來源以方便測試）
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 允許的來源網址清單
+ALLOWED_ORIGINS = [
+    "https://n-m2azgihk7rtcqqaxywntv75at3thf3bgps4xdlq-0lu-script.googleusercontent.com",
+    "https://www.dfes.ntpc.edu.tw",
+    "https://demo.dfes.ntpc.edu.tw"
+]
+
+# 如果希望允許所有 GAS，取消下面兩行的註解
+# "https://script.google.com",  # 所有 Google Apps Script
+
+# 掛載靜態文件（如果存在）
+if os.path.exists("static"):
+    app.mount("/static", StaticFiles(directory="static"), name="static")
+
+class TTSRequest(BaseModel):
+    text: str
+    voice: Optional[str] = "zh-TW-HsiaoChenNeural"
+    rate: Optional[str] = "+0%"
+    volume: Optional[str] = "+0%"
+    pitch: Optional[str] = "+0Hz"
+
+class TTSResponse(BaseModel):
+    success: bool
+    message: str
+    audio_url: Optional[str] = None
+    error: Optional[str] = None
+
+# 來源檢查函數
+def _is_origin_allowed(request: Request) -> bool:
+    """檢查請求來源是否被允許"""
+    origin = request.headers.get("origin")
+    referer = request.headers.get("referer")
+    user_agent = request.headers.get("user-agent", "")
+    
+    print(f"請求來源檢查 - Origin: {origin}, Referer: {referer}, User-Agent: {user_agent}")
+    
+    # 檢查允許的來源（一般網頁）
+    if origin:
+        for allowed_origin in ALLOWED_ORIGINS:
+            if origin.startswith(allowed_origin):
+                print(f"✅ Origin 匹配: {origin} 匹配 {allowed_origin}")
+                return True
+    
+    if referer:
+        for allowed_origin in ALLOWED_ORIGINS:
+            if referer.startswith(allowed_origin):
+                print(f"✅ Referer 匹配: {referer} 匹配 {allowed_origin}")
+                return True
+    
+    # 允許來自 Hugging Face Spaces 的直接調用
+    if "huggingface" in user_agent.lower():
+        print("✅ 檢測到 Hugging Face Spaces 環境")
+        return True
+    
+    print(f"❌ 請求被拒絕 - Origin: {origin}, Referer: {referer}")
+    return False
+
+@app.get("/")
+async def root():
+    """根路徑，返回 API 信息"""
+    return {
+        "message": "Edge TTS API Service - Hugging Face Spaces",
+        "version": "1.0.0",
+        "endpoints": {
+            "GET /voices": "獲取所有可用語音",
+            "POST /tts": "文字轉語音",
+            "GET /tts": "文字轉語音 (GET 方法)",
+            "GET /health": "健康檢查",
+            "GET /allowed-origins": "查看允許的來源網址",
+            "GET /debug-request": "調試請求信息"
+        },
+        "note": "此服務部署在 Hugging Face Spaces 上，僅允許特定來源網址訪問"
+    }
+
+@app.get("/health")
+async def health_check():
+    """健康檢查端點"""
+    return {"status": "healthy", "service": "edge-tts-api", "platform": "huggingface-spaces"}
+
+@app.get("/allowed-origins")
+async def get_allowed_origins():
+    """獲取允許的來源列表（僅供管理員查看）"""
+    return {
+        "allowed_origins": ALLOWED_ORIGINS,
+        "count": len(ALLOWED_ORIGINS),
+        "description": "允許的來源網址清單，支援擴充匹配"
+    }
+
+@app.get("/debug-request")
+async def debug_request(request: Request):
+    """調試端點：顯示請求的詳細信息"""
+    headers = dict(request.headers)
+    return {
+        "method": request.method,
+        "url": str(request.url),
+        "headers": headers,
+        "origin": headers.get("origin"),
+        "referer": headers.get("referer"),
+        "user_agent": headers.get("user-agent"),
+        "is_allowed": _is_origin_allowed(request),
+        "client_ip": request.client.host if request.client else "unknown",
+        "request_info": {
+            "has_origin": bool(headers.get("origin")),
+            "has_referer": bool(headers.get("referer")),
+            "is_huggingface": "huggingface" in headers.get("user-agent", "").lower(),
+            "api_key_provided": "x-api-key" in headers
+        }
+    }
+
+
+@app.get("/voices")
+async def get_voices():
+    """獲取所有可用的語音列表"""
+    try:
+        voices = await edge_tts.list_voices()
+        return {
+            "success": True,
+            "voices": voices,
+            "count": len(voices)
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"獲取語音列表失敗: {str(e)}")
+
+@app.post("/tts", response_model=TTSResponse)
+async def text_to_speech(request: TTSRequest, http_request: Request):
+    """文字轉語音 API (POST 方法)"""
+    try:
+        # 檢查來源是否被允許
+        if not _is_origin_allowed(http_request):
+            return TTSResponse(
+                success=False, 
+                message="來源網站未被允許使用此 API", 
+                error="forbidden"
+            )
+        
+        # 生成唯一文件名
+        file_id = str(uuid.uuid4())
+        output_file = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
+        
+        # 創建 TTS 通信對象
+        communicate = edge_tts.Communicate(
+            text=request.text,
+            voice=request.voice,
+            rate=request.rate,
+            volume=request.volume,
+            pitch=request.pitch
+        )
+        
+        # 生成語音文件
+        await communicate.save(output_file)
+        
+        # 檢查文件是否成功創建
+        if not os.path.exists(output_file):
+            raise Exception("語音文件生成失敗")
+        
+        return TTSResponse(
+            success=True,
+            message="語音生成成功",
+            audio_url=f"/audio/{file_id}.mp3"
+        )
+        
+    except Exception as e:
+        return TTSResponse(
+            success=False,
+            message="語音生成失敗",
+            error=str(e)
+        )
+
+@app.get("/tts")
+async def text_to_speech_get(
+    text: str = Query(..., description="要轉換的文字"),
+    voice: str = Query("zh-TW-HsiaoChenNeural", description="語音名稱"),
+    rate: str = Query("+0%", description="語速調整"),
+    volume: str = Query("+0%", description="音量調整"),
+    pitch: str = Query("+0Hz", description="音調調整"),
+    http_request: Request = None
+):
+    """文字轉語音 API (GET 方法)"""
+    try:
+        print(f"TTS 請求: text={text}, voice={voice}, rate={rate}, volume={volume}, pitch={pitch}")
+        
+        # 檢查來源是否被允許
+        if http_request and not _is_origin_allowed(http_request):
+            raise HTTPException(
+                status_code=403, 
+                detail="來源網站未被允許使用此 API"
+            )
+        
+        # 生成唯一文件名
+        file_id = str(uuid.uuid4())
+        output_file = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
+        print(f"輸出文件路徑: {output_file}")
+        
+        # 創建 TTS 通信對象
+        communicate = edge_tts.Communicate(
+            text=text,
+            voice=voice,
+            rate=rate,
+            volume=volume,
+            pitch=pitch
+        )
+        
+        # 生成語音文件
+        print("開始生成語音文件...")
+        await communicate.save(output_file)
+        print("語音文件生成完成")
+        
+        # 檢查文件是否成功創建
+        if not os.path.exists(output_file):
+            print(f"文件不存在: {output_file}")
+            raise HTTPException(status_code=500, detail="語音文件生成失敗")
+        
+        file_size = os.path.getsize(output_file)
+        print(f"文件大小: {file_size} bytes")
+        
+        # 返回音頻文件
+        return FileResponse(
+            output_file,
+            media_type="audio/mpeg",
+            filename=f"tts_{file_id}.mp3"
+        )
+        
+    except Exception as e:
+        print(f"TTS 錯誤: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"語音生成失敗: {str(e)}")
+
+@app.get("/audio/{file_id}.mp3")
+async def get_audio_file(file_id: str):
+    """獲取生成的音頻文件"""
+    file_path = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
+    
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="音頻文件不存在")
+    
+    return FileResponse(
+        file_path,
+        media_type="audio/mpeg",
+        filename=f"tts_{file_id}.mp3"
+    )
+
+@app.delete("/audio/{file_id}.mp3")
+async def delete_audio_file(file_id: str):
+    """刪除音頻文件"""
+    file_path = os.path.join(OUTPUT_DIR, f"{file_id}.mp3")
+    
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="音頻文件不存在")
+    
+    try:
+        os.remove(file_path)
+        return {"success": True, "message": "文件刪除成功"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"文件刪除失敗: {str(e)}")
+
+# 清理函數已移至 lifespan 處理器中
+
+if __name__ == "__main__":
+    import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)