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 路由 # ===================== @app.get("/") def root(): return {"message": "DeblurGANv2 API ready!"} @app.post("/webhook") 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" @app.post("/ai_agent") 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 ) @app.post("/predict") 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)