import os import io import traceback import json # 解決 Matplotlib/Fontconfig 在無頭環境中的問題 os.environ["MPLCONFIGDIR"] = "/tmp/matplotlib" os.environ["XDG_CACHE_HOME"] = "/tmp" os.environ["USER_AGENT"] = "my-huggingface-space" import gradio as gr from fastapi import FastAPI, Request, Header, BackgroundTasks, HTTPException, status import requests # 確保導入 requests import requests_handler # 確保導入 requests_handler from linebot import LineBotApi, WebhookHandler from linebot.exceptions import InvalidSignatureError from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, ImageMessage, AudioMessage, AudioSendMessage import google.generativeai as genai from chatbot import ChatBot from line_bot_app import LineBotApp from system_instruction import system_instruction # --- 將 app 實例化移到頂層 --- app = FastAPI() # --- Gradio 的處理函數 --- def greet(name): """Gradio UI 的簡單測試函數""" return "Hello " + name + "!! 橘橘管理員介面測試中..." # --- FastAPI 路由定義 --- @app.post("/callback") async def callback(request: Request, background_tasks: BackgroundTasks, x_line_signature: str = Header(None)): """ Line Webhook 回調路由。 """ body = await request.body() # 從 app.state 獲取 bot_app 實例 bot_app = getattr(app.state, 'bot_app', None) if not bot_app: print("Error: bot_app not initialized in app state.") raise HTTPException(status_code=500, detail="Bot application not ready") try: # 使用背景任務處理 webhook，避免阻塞請求 background_tasks.add_task( bot_app.handle_webhook, body.decode("utf-8"), x_line_signature ) return {"status": "ok"} except InvalidSignatureError: print("Invalid signature error.") raise HTTPException(status_code=400, detail="Invalid signature") except Exception as e: print(f"處理 webhook 時發生未預期錯誤: {e}") print(traceback.format_exc()) # 即使內部處理出錯，也盡量回覆 Line "ok"，避免 Line 不斷重試 return {"status": "ok"} # 或者返回 500 錯誤，取決於你的策略 @app.get("/") async def read_root(): """ 根路由，用於確認服務正常運行。 """ return {"status": "ok", "message": "Line Bot Service is running"} @app.get("/healthcheck") async def health_check(): """ 健康檢查路由。 """ health_status = { "status": "ok", "details": {} } # 檢查 ChatBot 是否初始化 (通過 LineBotApp) bot_app = getattr(app.state, 'bot_app', None) if bot_app and hasattr(bot_app, 'chatbot') and bot_app.chatbot: health_status["details"]["chatbot_initialized"] = "ok" # 檢查 Gemini 模型是否初始化 if bot_app.chatbot.model: health_status["details"]["gemini_chat_model"] = "ok" # 可以加一個簡單的 API 測試，但避免重複配置 try: # 假設 chatbot 實例已配置好 API Key _ = bot_app.chatbot.model.generate_content("health check", generation_config={"max_output_tokens": 5}) # 限制輸出 except Exception as e: health_status["details"]["gemini_chat_model"] = f"error: {str(e)}" health_status["status"] = "error" else: health_status["details"]["gemini_chat_model"] = "error: model not loaded" health_status["status"] = "error" # 檢查圖片生成模型是否初始化 (如果需要) if hasattr(bot_app.chatbot, 'image_generator') and bot_app.chatbot.image_generator and hasattr(bot_app.chatbot.image_generator, 'image_model') and bot_app.chatbot.image_generator.image_model: health_status["details"]["gemini_image_model"] = "ok" else: health_status["details"]["gemini_image_model"] = "warning: model not loaded or check not implemented" # 可以考慮不將此設為 error else: health_status["details"]["chatbot_initialized"] = "error" health_status["status"] = "error" # 檢查 Line Bot API (簡單檢查實例是否存在) if bot_app and hasattr(bot_app, 'line_bot_api'): health_status["details"]["line_bot_api"] = "ok (instance exists)" else: health_status["details"]["line_bot_api"] = "error: instance not found" health_status["status"] = "error" return health_status # --- FastAPI 啟動事件 --- @app.on_event("startup") async def startup_event(): """ FastAPI 啟動事件。初始化 ChatBot、Line Bot 應用和 Gradio UI。 """ print("===== Application Startup =====") try: # 從環境變數獲取配置 google_api_key = os.getenv("GOOGLE_API_KEY") line_channel_access_token = os.getenv("CHANNEL_ACCESS_TOKEN") line_channel_secret = os.getenv("CHANNEL_SECRET") google_search_api_key = os.getenv("GOOGLE_CUSTOM_SEARCH_API_KEY") google_search_cse_id = os.getenv("GOOGLE_CUSTOM_SEARCH_CSE_ID") print("Checking required environment variables...") if not google_api_key: print("Warning: GOOGLE_API_KEY not set.") if not line_channel_access_token: print("Warning: CHANNEL_ACCESS_TOKEN not set.") if not line_channel_secret: print("Warning: CHANNEL_SECRET not set.") # 根據你的應用是否嚴格需要這些來決定是否 raise ValueError # if not all([google_api_key, line_channel_access_token, line_channel_secret]): # raise ValueError("缺少必要的環境變數設定 (Google API Key, Line Channel Access Token, Line Channel Secret)") print("Initializing ChatBot...") # 初始化 ChatBot chatbot = ChatBot( google_api_key=google_api_key, # 傳遞 API Key system_instruction=system_instruction, google_search_api_key=google_search_api_key, google_search_cse_id=google_search_cse_id ) print("ChatBot initialized.") print("Initializing LineBotApp...") # 初始化 Line Bot bot_app = LineBotApp( channel_access_token=line_channel_access_token, channel_secret=line_channel_secret, chatbot=chatbot ) print("LineBotApp initialized.") # 將 bot_app 保存到應用狀態中 (使用頂層 app) app.state.bot_app = bot_app print("LineBotApp stored in app state.") print("Setting up Gradio UI...") # 設置 Gradio UI gradio_ui = gr.Interface( fn=greet, inputs="text", outputs="text", flagging_options=None, # 使用 flagging_options title="Line Bot Admin", description="管理員介面 - 用於測試和監控 Line Bot" ) # 使用頂層 app 掛載 Gradio # mount_gradio_app 通常會修改傳入的 app 實例 gr.mount_gradio_app(app, gradio_ui, path="/ui") print("Gradio UI mounted at /ui.") print("===== Application Startup Complete =====") except Exception as e: print(f"啟動服務時發生錯誤: {e}") print(traceback.format_exc()) # 啟動失敗時，可以考慮讓應用程式退出或進入錯誤狀態 # raise e # 重新拋出異常，可能會讓 uvicorn 退出 # --- 主程式入口點 --- if __name__ == "__main__": import uvicorn # 從環境變數讀取端口，預設為 7860 (Hugging Face 通常用這個) port = int(os.getenv("PORT", 7860)) print(f"Starting Uvicorn on host 0.0.0.0, port {port}") # 使用頂層 app 啟動 uvicorn uvicorn.run(app, host="0.0.0.0", port=port)