Spaces:
Running
Running
| from fastapi import FastAPI, Request, Response, Form, Header, HTTPException, BackgroundTasks | |
| from fastapi.responses import JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.middleware.cors import CORSMiddleware # 匯入 FastAPI 的 CORS 中介軟體 | |
| import requests | |
| from typing import Annotated # 推薦用於 Pydantic v2+ | |
| from linebot.exceptions import InvalidSignatureError # 匯入 Line 簽章無效的例外 | |
| from services.linebot import line_handle | |
| from services.deblur import deblur_image_tiled | |
| from services.agents import run_agent | |
| import json | |
| from PIL import Image | |
| import io | |
| import os | |
| from datetime import datetime | |
| import uvicorn | |
| from dotenv import load_dotenv # 匯入 dotenv 以載入 .env 環境變數檔案 | |
| STATIC_DIR = "static" | |
| os.environ["TORCH_HOME"] = "./.cache" | |
| os.environ["HF_HOME"] = "./.cache" | |
| os.environ["TRANSFORMERS_CACHE"] = "./.cache" | |
| os.makedirs("./.cache", exist_ok=True) | |
| os.makedirs(STATIC_DIR, exist_ok=True) | |
| load_dotenv() | |
| # ===================== | |
| # 初始化 FastAPI | |
| # ===================== | |
| app = FastAPI(title="DeblurGANv2 API") | |
| app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") | |
| # 設定 CORS (跨來源資源共用) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], # 允許所有來源 | |
| allow_credentials=True, # 允許憑證 | |
| allow_methods=["*"], # 允許所有 HTTP 方法 | |
| allow_headers=["*"], # 允許所有 HTTP 標頭 | |
| ) | |
| # ===================== | |
| # API 路由 | |
| # ===================== | |
| def root(): | |
| return {"message": "DeblurGANv2 API ready!"} | |
| async def webhook( | |
| request: Request, | |
| background_tasks: BackgroundTasks, | |
| x_line_signature=Header(None), # 從標頭獲取 Line 的簽章 | |
| ): | |
| """ | |
| Line Bot 的 Webhook 路由。 | |
| """ | |
| # 獲取請求的原始內容 (body) | |
| body = await request.body() | |
| try: | |
| # 使用背景任務來處理 Webhook，這樣可以立即回傳 200 OK 給 Line 伺服器 | |
| background_tasks.add_task( | |
| line_handle, body.decode("utf-8"), x_line_signature | |
| ) | |
| except InvalidSignatureError: | |
| # 如果簽章無效，拋出 400 錯誤 | |
| raise HTTPException(status_code=400, detail="Invalid signature") | |
| return "ok" | |
| async def ai_agent( | |
| input_text: Annotated[str, Form(description="輸入文字")], | |
| # 將您的 form-data 欄位定義為函數參數，並使用 Form() | |
| file_name: Annotated[str, Form(description="檔案名稱")] = None, | |
| file_format: Annotated[str, Form(description="檔案格式")] = None, | |
| file_url: Annotated[str, Form(description="檔案下載網址")] = None, # 可選參數，有預設值 | |
| # 對於需要轉換類型 (例如 int) 的欄位，直接在類型提示中指定 | |
| file_width: Annotated[int, Form(description="檔案寬度")] = 0, | |
| file_height: Annotated[int, Form(description="檔案高度")] = 0, | |
| file_created_at: Annotated[str, Form(description="檔案建立時間")] = None, | |
| ): | |
| out_str = "" | |
| image_url = "" | |
| text_result = "" | |
| try: | |
| agent_input = input_text | |
| if file_url: | |
| agent_input = f"請根據這張圖片回答問題。圖片的路徑是 {file_url}，我的問題是：{input_text}" | |
| # 運行 LangChain 代理人 | |
| response = run_agent(agent_input) | |
| # Agent 的輸出是 JSON 字串 (無論成功或失敗) | |
| out_str = response["output"] | |
| #print(f"out_str:{out_str}") | |
| tool_result = out_str.get("tool_result", {}) | |
| if tool_result is None: | |
| tool_result = {} | |
| #print(f"tool_result:{tool_result}") | |
| final_response = out_str.get("final_response") | |
| #print(f"final_response:{final_response}") | |
| if "error" in tool_result: | |
| # 3A. 處理錯誤情況 (例如：圖片下載失敗、API 錯誤) | |
| error_msg = tool_result["error"] | |
| reply_text = f"🚫 圖片工具執行失敗：\n{error_msg}" | |
| return JSONResponse( | |
| {"status": "error", "message": reply_text}, | |
| status_code=500 | |
| ) | |
| elif "image_url" in tool_result: | |
| # 3B. 處理成功情況 (帶有圖片 URL，例如：圖片生成或去模糊工具) | |
| image_url = tool_result["image_url"] | |
| text_result = tool_result.get("text_result", "圖片處理完成。") | |
| # Line 要求圖片 URL 必須是 HTTPS | |
| # 由於 HF_SPACE 預設是 https，這裡做一個保險轉換 | |
| if image_url.startswith("http://"): | |
| image_url = image_url.replace("http://", "https://") | |
| elif "text_result" in tool_result: | |
| # 3C. 處理純文字結果 (例如：多模態分析工具) | |
| text_result = tool_result["text_result"] | |
| elif final_response: | |
| text_result = final_response | |
| else: | |
| # 3D. 處理意外的 JSON 結構 | |
| text_result = "代理人回覆格式無法識別，請聯繫管理員。" | |
| return JSONResponse( | |
| { | |
| "status": "success", | |
| "message": text_result, | |
| "image_url" : image_url | |
| }, status_code=200 | |
| ) | |
| except json.JSONDecodeError: | |
| # 4. 處理 Agent 返回了無法解析的純文字 (非 JSON) | |
| # 這通常發生在 Agent 決定不使用工具，直接回覆純文字，或者推理過程出錯。 | |
| return JSONResponse( | |
| {"status": "error", "message": out_str }, status_code=500 | |
| ) | |
| except Exception as e: | |
| # 處理代理人執行時的錯誤 | |
| print(f"代理人執行出錯: {e}") | |
| out = f"代理人執行出錯!錯誤訊息：{e}" | |
| import traceback | |
| traceback.print_exc() | |
| return JSONResponse( | |
| {"status": "error", "message": out }, status_code=500 | |
| ) | |
| async def predict( | |
| request: Request, | |
| # 將您的 form-data 欄位定義為函數參數，並使用 Form() | |
| file_name: Annotated[str, Form(description="檔案名稱")], | |
| file_format: Annotated[str, Form(description="檔案格式")], | |
| file_url: Annotated[str, Form(description="檔案下載網址")], # 可選參數，有預設值 | |
| # 對於需要轉換類型 (例如 int) 的欄位，直接在類型提示中指定 | |
| file_width: Annotated[int, Form(description="檔案寬度")] = 0, | |
| file_height: Annotated[int, Form(description="檔案高度")] = 0, | |
| file_created_at: Annotated[str, Form(description="檔案建立時間")] = None, | |
| ): | |
| try: | |
| print("### start /predict !!") | |
| if not file_url: | |
| return JSONResponse( | |
| {"status": "error", "message": "file_url is required"}, | |
| status_code=400 | |
| ) | |
| # 2️⃣ 從 URL 下載圖片 | |
| resp = requests.get(file_url) | |
| resp.raise_for_status() | |
| img = Image.open(io.BytesIO(resp.content)).convert("RGB") | |
| # 3️⃣ 去模糊 | |
| result = deblur_image_tiled(img) | |
| # 4️⃣ 產生檔名 | |
| base_name = f"{file_name}_{file_width}_{file_height}_{file_created_at}.jpg" | |
| file_path = os.path.join(STATIC_DIR, base_name) | |
| # 5️⃣ 儲存到 static | |
| result.save(file_path, format="JPEG") | |
| # 6️⃣ 回傳前端可取用的 URL | |
| file_url_return = str(request.base_url) + f"static/{base_name}" | |
| return { | |
| "status": "success", | |
| "file_url": file_url_return, | |
| "file_name": base_name, | |
| "file_format": "jpg", | |
| "file_width": result.width, | |
| "file_height": result.height, | |
| "file_created_at": datetime.now().strftime("%Y%m%d%H%M%S") | |
| } | |
| except Exception as e: | |
| import traceback | |
| traceback.print_exc() | |
| return JSONResponse( | |
| {"status": "error", "message": str(e)}, status_code=500 | |
| ) | |
| if __name__ == "__main__": | |
| uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False) |