|
import os |
|
import io |
|
import PIL.Image |
|
from collections import defaultdict |
|
from linebot import LineBotApi |
|
|
|
|
|
|
|
|
|
line_bot_api = LineBotApi(os.environ["CHANNEL_ACCESS_TOKEN"]) |
|
def get_image_url(message_id): |
|
try: |
|
message_content = line_bot_api.get_message_content(message_id) |
|
file_path = f"/tmp/{message_id}.png" |
|
with open(file_path, "wb") as f: |
|
for chunk in message_content.iter_content(): |
|
f.write(chunk) |
|
print(f"✅ 圖片成功儲存到:{file_path}") |
|
return file_path |
|
except Exception as e: |
|
print(f"❌ 圖片取得失敗:{e}") |
|
return None |
|
|
|
|
|
user_message_history = defaultdict(list) |
|
def store_user_message(user_id, message_type, message_content): |
|
""" |
|
儲存用戶的訊息 |
|
""" |
|
user_message_history[user_id].append({ |
|
"type": message_type, |
|
"content": message_content}) |
|
|
|
def analyze_with_gemini(image_path, user_text): |
|
""" |
|
分析用戶問題和圖片,並返回 Gemini 的回應 |
|
""" |
|
try: |
|
|
|
if not os.path.exists(image_path): |
|
raise FileNotFoundError(f"圖片路徑無效:{image_path}") |
|
|
|
img_user = PIL.Image.open(image_path) |
|
response = chat.send_message([user_text, img_user]) |
|
|
|
|
|
return response.text |
|
|
|
except Exception as e: |
|
return f"發生錯誤: {e}" |
|
|
|
def get_previous_message(user_id): |
|
""" |
|
獲取用戶的上一則訊息 |
|
""" |
|
if user_id in user_message_history and len(user_message_history[user_id]) > 0: |
|
|
|
return user_message_history[user_id][-1] |
|
return { |
|
"type": 'text', |
|
"content": 'No message!'} |
|
|