new

example01.py

@@ -0,0 +1,288 @@
+# ===東吳大學資料系 2025 年 LINEBOT ===
+import logging
+import os
+import tempfile
+import uuid
+from io import BytesIO
+
+import markdown
+from bs4 import BeautifulSoup
+from flask import Flask, abort, request, send_from_directory
+
+from google import genai
+from google.genai import types
+from google.genai.types import Tool, GenerateContentConfig, GoogleSearch
+
+from linebot.v3 import WebhookHandler
+from linebot.v3.exceptions import InvalidSignatureError
+from linebot.v3.messaging import (
+    ApiClient,
+    Configuration,
+    ImageMessage,
+    MessagingApi,
+    MessagingApiBlob,
+    ReplyMessageRequest,
+    TextMessage,
+)
+from linebot.v3.webhooks import (
+    ImageMessageContent,
+    MessageEvent,
+    TextMessageContent,
+)
+
+from PIL import Image
+from linebot.v3.webhooks import VideoMessageContent
+import requests
+
+# === 初始化 Google Gemini ===
+GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY")
+client = genai.Client(api_key=GOOGLE_API_KEY)
+
+google_search_tool = Tool(
+    google_search=GoogleSearch()
+)
+
+chat = client.chats.create(
+    model="gemini-2.5-pro-preview-05-06",
+    config=GenerateContentConfig(
+        system_instruction="你是一個中文的AI助手，關於所有問題，請用繁體中文文言文回答",
+        tools=[google_search_tool],
+        response_modalities=["TEXT"],
+    )
+)
+
+# === 初始設定 ===
+static_tmp_path = tempfile.gettempdir()
+os.makedirs(static_tmp_path, exist_ok=True)
+base_url = os.getenv("SPACE_HOST")  # e.g., "your-space-name.hf.space"
+
+# === Flask 應用初始化 ===
+app = Flask(__name__)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s"
+)
+app.logger.setLevel(logging.INFO)
+
+channel_secret = os.environ.get("YOUR_CHANNEL_SECRET")
+channel_access_token = os.environ.get("YOUR_CHANNEL_ACCESS_TOKEN")
+
+configuration = Configuration(access_token=channel_access_token)
+handler = WebhookHandler(channel_secret)
+
+
+# === AI Query 包裝 ===
+def query(payload):
+    response = chat.send_message(message=payload)
+    return response.text
+
+
+# === 靜態圖檔路由 ===
+@app.route("/images/<filename>")
+def serve_image(filename):
+    return send_from_directory(static_tmp_path, filename)
+
+
+# === LINE Webhook 接收端點 ===
+@app.route("/")
+def home():
+    return {"message": "Line Webhook Server"}
+
+
+@app.route("/", methods=["POST"])
+def callback():
+    signature = request.headers.get("X-Line-Signature")
+    body = request.get_data(as_text=True)
+    app.logger.info(f"Request body: {body}")
+
+    try:
+        handler.handle(body, signature)
+    except InvalidSignatureError:
+        app.logger.warning("Invalid signature. Please check channel credentials.")
+        abort(400)
+
+    return "OK"
+
+
+# === 處理文字訊息 ===
+@handler.add(MessageEvent, message=TextMessageContent)
+def handle_text_message(event):
+    user_input = event.message.text.strip()
+    if user_input.startswith("AI "):
+        prompt = user_input[3:].strip()
+        try:
+            # 使用 Gemini 生成圖片
+            response = client.models.generate_content(
+                model="gemini-2.0-flash-exp-image-generation",
+                contents=prompt,
+                config=types.GenerateContentConfig(
+                    response_modalities=["TEXT", "IMAGE"]
+                ),
+            )
+
+            # 處理回應中的圖片
+            for part in response.candidates[0].content.parts:
+                if part.inline_data is not None:
+                    image = Image.open(BytesIO(part.inline_data.data))
+                    filename = f"{uuid.uuid4().hex}.png"
+                    image_path = os.path.join(static_tmp_path, filename)
+                    image.save(image_path)
+
+                    # 建立圖片的公開 URL
+                    image_url = f"https://{base_url}/images/{filename}"
+                    app.logger.info(f"Image URL: {image_url}")
+
+                    # 回傳圖片給 LINE 使用者
+                    with ApiClient(configuration) as api_client:
+                        line_bot_api = MessagingApi(api_client)
+                        line_bot_api.reply_message(
+                            ReplyMessageRequest(
+                                reply_token=event.reply_token,
+                                messages=[
+                                    ImageMessage(
+                                        original_content_url=image_url,
+                                        preview_image_url=image_url,
+                                    )
+                                ],
+                            )
+                        )
+
+        except Exception as e:
+            app.logger.error(f"Gemini API error: {e}")
+            with ApiClient(configuration) as api_client:
+                line_bot_api = MessagingApi(api_client)
+                line_bot_api.reply_message(
+                    ReplyMessageRequest(
+                        reply_token=event.reply_token,
+                        messages=[TextMessage(text="抱歉，生成圖片時發生錯誤。")],
+                    )
+                )
+    else:
+        with ApiClient(configuration) as api_client:
+            line_bot_api = MessagingApi(api_client)
+            response = query(event.message.text)
+            html_msg = markdown.markdown(response)
+            soup = BeautifulSoup(html_msg, "html.parser")
+
+            line_bot_api.reply_message_with_http_info(
+                ReplyMessageRequest(
+                    reply_token=event.reply_token,
+                    messages=[TextMessage(text=soup.get_text())],
+                )
+            )
+
+
+# === 處理圖片訊息 ===
+@handler.add(MessageEvent, message=ImageMessageContent)
+def handle_image_message(event):
+    # === 以下是處理圖片回傳部分 === #
+    with ApiClient(configuration) as api_client:
+        blob_api = MessagingApiBlob(api_client)
+        content = blob_api.get_message_content(message_id=event.message.id)
+
+    # Step 4：將圖片存到本地端
+    with tempfile.NamedTemporaryFile(
+        dir=static_tmp_path, suffix=".jpg", delete=False
+    ) as tf:
+        tf.write(content)
+        filename = os.path.basename(tf.name)
+
+    image_url = f"https://{base_url}/images/{filename}"
+
+    app.logger.info(f"Image URL: {image_url}")
+
+    # === 以下是解釋圖片 === #
+    image = Image.open(tf.name)
+    response = client.models.generate_content(
+        model="gemini-2.0-flash",
+        config=types.GenerateContentConfig(
+            system_instruction="你是一個資深的面相命理師，如果有人上手掌的照片，就幫他解釋手相，如果上傳正面臉部的照片，就幫他解釋面相，照片要先去背，如果是一般的照片，就正常說明照片不用算命，請用繁體中文回答",
+            response_modalities=["TEXT"],
+            tools=[google_search_tool],
+        ),
+        contents=[image, "用繁體中文描述這張圖片"],
+    )
+    app.logger.info(response.text)
+
+    # === 以下是回傳圖片部分 === #
+    with ApiClient(configuration) as api_client:
+        line_bot_api = MessagingApi(api_client)
+        line_bot_api.reply_message(
+            ReplyMessageRequest(
+                reply_token=event.reply_token,
+                messages=[
+                    ImageMessage(
+                        original_content_url=image_url, preview_image_url=image_url
+                    ),
+                    TextMessage(text=response.text),
+                ],
+            )
+        )
+
+# === 處理影片訊息 ===
+
+@handler.add(MessageEvent, message=VideoMessageContent)
+def handle_video_message(event):
+    # 下載影片內容
+    with ApiClient(configuration) as api_client:
+        blob_api = MessagingApiBlob(api_client)
+        video_data = blob_api.get_message_content(message_id=event.message.id)
+
+    # 儲存影片到本地
+    if video_data is None:
+        err_msg = "抱歉，無法取得影片內容。"
+        app.logger.error(err_msg)
+        with ApiClient(configuration) as api_client:
+            line_bot_api = MessagingApi(api_client)
+            line_bot_api.reply_message(
+                ReplyMessageRequest(
+                    reply_token=event.reply_token,
+                    messages=[TextMessage(text=err_msg)]
+                )
+            )
+        return
+
+    with tempfile.NamedTemporaryFile(
+        dir=static_tmp_path, suffix=".mp4", delete=False
+    ) as tf:
+        tf.write(video_data)
+        filename = os.path.basename(tf.name)
+
+    video_url = f"https://{base_url}/images/{filename}"
+    app.logger.info(f"Video URL: {video_url}")
+
+    # 影片說明
+    try:
+        video_response = requests.get(video_url)
+        video_response.raise_for_status()
+        video_data = video_response.content
+
+        response = client.models.generate_content(
+            model="gemini-2.5-flash-preview-05-20",
+            config=types.GenerateContentConfig(
+                system_instruction="你是一個專業的影片解說員，請用繁體中文簡要說明這段影片的內容。",
+                response_modalities=["TEXT"],
+                tools=[google_search_tool],
+            ),
+            contents=[{"mime_type": "video/mp4", "data": video_data}, "用繁體中文描述這段影片"],
+        )
+        description = response.text
+    except Exception as e:
+        app.logger.error(f"Gemini API error (video): {e}")
+        description = "抱歉，無法解釋這段影片內容。"
+    except Exception as e:
+        app.logger.error(f"Gemini API error (video): {e}")
+        description = "抱歉，無法解釋這段影片內容。"
+
+    # 回傳影片連結與說明
+    with ApiClient(configuration) as api_client:
+        line_bot_api = MessagingApi(api_client)
+        line_bot_api.reply_message(
+            ReplyMessageRequest(
+                reply_token=event.reply_token,
+                messages=[
+                    TextMessage(text=f"影片連結：{video_url}"),
+                    TextMessage(text=description),
+                ],
+            )
+        )

gemini.py

@@ -225,13 +225,26 @@ def handle_video_message(event):
     # 下載影片內容
     with ApiClient(configuration) as api_client:
         blob_api = MessagingApiBlob(api_client)
-        content = blob_api.get_message_content(message_id=event.message.id)
+        video_data = blob_api.get_message_content(message_id=event.message.id)
 
     # 儲存影片到本地
+    if video_data is None:
+        err_msg = "抱歉，無法取得影片內容。"
+        app.logger.error(err_msg)
+        with ApiClient(configuration) as api_client:
+            line_bot_api = MessagingApi(api_client)
+            line_bot_api.reply_message(
+                ReplyMessageRequest(
+                    reply_token=event.reply_token,
+                    messages=[TextMessage(text=err_msg)]
+                )
+            )
+        return
+
     with tempfile.NamedTemporaryFile(
         dir=static_tmp_path, suffix=".mp4", delete=False
     ) as tf:
-        tf.write(content)
+        tf.write(video_data)
         filename = os.path.basename(tf.name)
 
     video_url = f"https://{base_url}/images/{filename}"
@@ -239,14 +252,16 @@ def handle_video_message(event):
 
     # 影片說明
     try:
+        from io import BytesIO
+        video_bytes = BytesIO(video_data)
         response = client.models.generate_content(
-            model="gemini-2.0-flash",
+            model="gemini-2.5-flash-preview-05-20",
             config=types.GenerateContentConfig(
                 system_instruction="你是一個專業的影片解說員，請用繁體中文簡要說明這段影片的內容。",
                 response_modalities=["TEXT"],
                 tools=[google_search_tool],
             ),
-            contents=[{"mime_type": "video/mp4", "data": content}, "用繁體中文描述這段影片"],
+            contents=[video_bytes, "用繁體中文描述這段影片"],
         )
         description = response.text
     except Exception as e: