Linebot / main.py
alanchen1115's picture
Update main.py
da72bf2 verified
raw
history blame
8.17 kB
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI, Request, Header, BackgroundTasks, HTTPException, status
import google.generativeai as genai_gen
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, AudioMessage, ImageMessage
import json, os
import io
import PIL.Image
from Image_text_generation import Image_text_Generator
from Uploading_images_file import get_image_url,get_file_url, store_user_message, analyze_with_gemini, get_previous_message
#==========================
# API 金鑰
#==========================
# 設定 Google AI API 金鑰
google_api = os.environ["GOOGLE_API_KEY"]
genai_gen.configure(api_key=google_api)
# 設定生成文字的參數
generation_config = genai_gen.types.GenerationConfig(max_output_tokens=1000, temperature=0.2, top_p=0.5, top_k=16) #2048
# 使用 gemini-2.0-flash-exp 模型
model = genai_gen.GenerativeModel('gemini-2.0-flash', system_instruction="主要用繁體中文回答,但如果用戶使用詢問英文問題,就用英文回應。你現在是個專業助理,職稱為OPEN小助理,個性活潑、樂觀,願意回答所有問題", generation_config=generation_config)
# 設定 Line Bot 的 API 金鑰和秘密金鑰
line_bot_api = LineBotApi(os.environ["CHANNEL_ACCESS_TOKEN"])
line_handler = WebhookHandler(os.environ["CHANNEL_SECRET"])
# 設定是否正在與使用者交談
working_status = os.getenv("DEFALUT_TALKING", default = "true").lower() == "true"
# 建立 FastAPI 應用程式
app = FastAPI()
# 設定 CORS,允許跨域請求
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 處理根路徑請求
@app.get("/")
def root():
return {"title": "Line Bot"}
# 處理 Line Webhook 請求
@app.post("/webhook")
async def webhook(
request: Request,
background_tasks: BackgroundTasks,
x_line_signature=Header(None),
):
# 取得請求內容
body = await request.body()
try:
# 將處理 Line 事件的任務加入背景工作
background_tasks.add_task(
line_handler.handle, body.decode("utf-8"), x_line_signature
)
except InvalidSignatureError:
# 處理無效的簽章錯誤
raise HTTPException(status_code=400, detail="Invalid signature")
return "ok"
#==========================
# 主程式(圖片與文字)
#==========================
# 建立 chat_sessions 字典
chat_sessions = {}
@line_handler.add(MessageEvent, message=(ImageMessage, TextMessage))
def handle_image_message(event):
user_id = event.source.user_id
chat = chat_sessions.get(user_id) or model.start_chat(history=[])
chat_sessions[user_id] = chat
# ========
# 生成圖片
# ========
user_text = event.message.text if event.message.type == "text" else None
image_url = None
if user_text and user_text.startswith("生成圖片"):
prompt = user_text.replace("生成圖片", "").strip()
# 先立即回覆避免token過期
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="圖片生成中~ 請稍候.....✨"))
image_generator = Image_text_Generator(user_id)
# 生成圖片
image_binary = image_generator.generate_image_with_gemini(prompt)
if image_binary:
image_url = image_generator.upload_image_to_imgur()
if image_url:
# 使用 push message 發送圖片,避免 reply token 超時
line_bot_api.push_message(
event.source.user_id,
[
TextSendMessage(text="✨ 這是我為你生成的圖片喔~"),
ImageSendMessage(original_content_url=image_url, preview_image_url=image_url)
]
)
else:
line_bot_api.push_message(
event.source.user_id,
TextSendMessage(text="⚠️ 圖片上傳失敗,請稍後再試~")
)
else:
line_bot_api.push_message(
event.source.user_id,
TextSendMessage(text="⚠️ 圖片生成失敗,請稍後再試~")
)
return
# ========
# 傳文件
# ========
if event.message.type == "file":
file_name = event.message.file_name
file_path = get_file_url(event.message.id, file_name, file_type="pdf")
if file_path:
file_generator = Image_text_Generator() # 確保這個類別已正確初始化
store_user_message(user_id, "file", file_path)
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text="📄 文件已接收成功囉,幫我輸入你想詢問的問題喔~")
)
else:
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text="❌ 沒有成功接收到文件,請再試一次。")
)
return
# 使用者在傳完文件後輸入問題
previous_message = get_previous_message(user_id)
if previous_message and previous_message["type"] == "file" and event.message.type == "text":
file_path = previous_message["content"]
user_text = event.message.text
store_user_message(user_id, "text", user_text)
try:
if not os.path.exists(file_path):
raise FileNotFoundError(f"📂 檔案路徑無效:{file_path}")
file_generator = Image_text_Generator(imgur)
completion = file_generator.generate_txt_with_gemini([user_text, file_path])
# 若是 Gemini 回傳的內容需取 text
out = completion.text if hasattr(completion, "text") else str(completion)
except Exception as e:
out = f"⚠️ 發生錯誤:{e}"
# ========
# 上傳圖片
# ========
if event.message.type == "image":
image_path = get_image_url(event.message.id)
if image_path:
store_user_message(user_id, "image", image_path)
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="圖片已接收成功囉,幫我輸入你想詢問的問題喔~"))
else:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="沒有接收到圖片~"))
return
previous_message = get_previous_message(user_id)
if previous_message and previous_message["type"] == "image" and event.message.type == "text":
image_path = previous_message["content"]
user_text = event.message.text
store_user_message(user_id, "text", user_text)
try:
if not os.path.exists(image_path):
raise FileNotFoundError(f"圖片路徑無效:{image_path}")
organ = PIL.Image.open(image_path)
completion = chat.send_message([user_text, organ])
out = completion.text
except Exception as e:
out = f"發生錯誤: {e}"
# ========
# 純文字
# ========
else:
if event.message.type != "text":
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="請輸入文字或圖片~"))
return
if event.message.text == "再見":
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="Bye!"))
return
if working_status:
try:
prompt = event.message.text
store_user_message(user_id, "text", prompt)
completion = chat.send_message(prompt)
out = completion.text if completion.text else "我不太懂什麼意思也~"
except:
out = "執行出錯!請換個說法!"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=out))
if __name__ == "__main__":
# 啟動 FastAPI 應用程式
uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True)