From 11d29075c27027a3e117c833f005396a12b835d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=98XGungo=E2=80=99?= <‘daniel2000890311g@gmail.com’> Date: Fri, 22 Sep 2023 16:08:11 +0800 Subject: [PATCH] update: linebot v3 method --- main.py | 95 ++++++++++++++++++++++++++++++++------------------------- 1 file changed, 53 insertions(+), 42 deletions(-) diff --git a/main.py b/main.py index 52680e7fc..6702661ab 100644 --- a/main.py +++ b/main.py @@ -1,14 +1,15 @@ from dotenv import load_dotenv from flask import Flask, request, abort -from linebot import ( - LineBotApi, WebhookHandler -) -from linebot.exceptions import ( - InvalidSignatureError -) -from linebot.models import ( - MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, AudioMessage -) + +from linebot.v3 import (WebhookHandler) +from linebot.v3.exceptions import (InvalidSignatureError) +from linebot.v3.messaging import (Configuration, ApiClient, MessagingApi, + ReplyMessageRequest, TextMessage, + ImageMessage, MessagingApiBlob) +from linebot.v3.webhooks import (MessageEvent, TextMessageContent, + AudioMessageContent) + + import os import uuid @@ -24,7 +25,11 @@ load_dotenv('.env') app = Flask(__name__) -line_bot_api = LineBotApi(os.getenv('LINE_CHANNEL_ACCESS_TOKEN')) +configuration = Configuration( + access_token=os.getenv('LINE_CHANNEL_ACCESS_TOKEN')) + +line_bot_api = MessagingApi(ApiClient(configuration)) +blob_api = MessagingApiBlob(ApiClient(configuration)) handler = WebhookHandler(os.getenv('LINE_CHANNEL_SECRET')) storage = None youtube = Youtube(step=4) @@ -49,7 +54,7 @@ def callback(): return 'OK' -@handler.add(MessageEvent, message=TextMessage) +@handler.add(MessageEvent, message=TextMessageContent) def handle_text_message(event): user_id = event.source.user_id text = event.message.text.strip() @@ -65,19 +70,21 @@ def handle_text_message(event): model_management[user_id] = model storage.save({ user_id: api_key - }) - msg = TextSendMessage(text='Token 有效，註冊成功') + }) + msg = TextMessage(text='Token 有效，註冊成功') elif text.startswith('/指令說明'): - msg = TextSendMessage(text="指令：\n/註冊 + API Token\n👉 API Token 請先到 https://platform.openai.com/ 註冊登入後取得\n\n/系統訊息 + Prompt\n👉 Prompt 可以命令機器人扮演某個角色，例如：請你扮演擅長做總結的人\n\n/清除\n👉 當前每一次都會紀錄最後兩筆歷史紀錄，這個指令能夠清除歷史訊息\n\n/圖像 + Prompt\n👉 會調用 DALL∙E 2 Model，以文字生成圖像\n\n語音輸入\n👉 會調用 Whisper 模型，先將語音轉換成文字，再調用 ChatGPT 以文字回覆\n\n其他文字輸入\n👉 調用 ChatGPT 以文字回覆") + msg = TextMessage( + text="指令：\n/註冊 + API Token\n👉 API Token 請先到 https://platform.openai.com/ 註冊登入後取得\n\n/系統訊息 + Prompt\n👉 Prompt 可以命令機器人扮演某個角色，例如：請你扮演擅長做總結的人\n\n/清除\n👉 當前每一次都會紀錄最後兩筆歷史紀錄，這個指令能夠清除歷史訊息\n\n/圖像 + Prompt\n👉 會調用 DALL∙E 2 Model，以文字生成圖像\n\n語音輸入\n👉 會調用 Whisper 模型，先將語音轉換成文字，再調用 ChatGPT 以文字回覆\n\n其他文字輸入\n👉 調用 ChatGPT 以文字回覆" + ) elif text.startswith('/系統訊息'): memory.change_system_message(user_id, text[5:].strip()) - msg = TextSendMessage(text='輸入成功') + msg = TextMessage(text='輸入成功') elif text.startswith('/清除'): memory.remove(user_id) - msg = TextSendMessage(text='歷史訊息清除成功') + msg = TextMessage(text='歷史訊息清除成功') elif text.startswith('/圖像'): prompt = text[3:].strip() @@ -86,10 +93,7 @@ def handle_text_message(event): if not is_successful: raise Exception(error_message) url = response['data'][0]['url'] - msg = ImageSendMessage( - original_content_url=url, - preview_image_url=url - ) + msg = ImageMessage(original_content_url=url, preview_image_url=url) memory.append(user_id, 'assistant', url) else: @@ -106,7 +110,7 @@ def handle_text_message(event): if not is_successful: raise Exception(error_message) role, response = get_role_and_content(response) - msg = TextSendMessage(text=response) + msg = TextMessage(text=response) else: chunks = website.get_content_from_url(url) if len(chunks) == 0: @@ -116,64 +120,71 @@ def handle_text_message(event): if not is_successful: raise Exception(error_message) role, response = get_role_and_content(response) - msg = TextSendMessage(text=response) + msg = TextMessage(text=response) else: is_successful, response, error_message = user_model.chat_completions(memory.get(user_id), os.getenv('OPENAI_MODEL_ENGINE')) if not is_successful: raise Exception(error_message) role, response = get_role_and_content(response) - msg = TextSendMessage(text=response) + msg = TextMessage(text=response) memory.append(user_id, role, response) except ValueError: - msg = TextSendMessage(text='Token 無效，請重新註冊，格式為 /註冊 sk-xxxxx') + msg = TextMessage(text='Token 無效，請重新註冊，格式為 /註冊 sk-xxxxx') except KeyError: - msg = TextSendMessage(text='請先註冊 Token，格式為 /註冊 sk-xxxxx') + msg = TextMessage(text='請先註冊 Token，格式為 /註冊 sk-xxxxx') except Exception as e: memory.remove(user_id) if str(e).startswith('Incorrect API key provided'): - msg = TextSendMessage(text='OpenAI API Token 有誤，請重新註冊。') + msg = TextMessage(text='OpenAI API Token 有誤，請重新註冊。') elif str(e).startswith('That model is currently overloaded with other requests.'): - msg = TextSendMessage(text='已超過負荷，請稍後再試') + msg = TextMessage(text='已超過負荷，請稍後再試') else: - msg = TextSendMessage(text=str(e)) - line_bot_api.reply_message(event.reply_token, msg) + msg = TextMessage(text=str(e)) + + line_bot_api.reply_message_with_http_info( + ReplyMessageRequest(reply_token=event.reply_token, messages=[msg])) -@handler.add(MessageEvent, message=AudioMessage) -def handle_audio_message(event): +@handler.add(MessageEvent, message=AudioMessageContent) +def handle_audio_message(event: MessageEvent): + print(event) user_id = event.source.user_id - audio_content = line_bot_api.get_message_content(event.message.id) + + audio_content = blob_api.get_message_content(event.message.id) + input_audio_path = f'{str(uuid.uuid4())}.m4a' with open(input_audio_path, 'wb') as fd: - for chunk in audio_content.iter_content(): - fd.write(chunk) + fd.write(audio_content) try: if not model_management.get(user_id): raise ValueError('Invalid API token') else: - is_successful, response, error_message = model_management[user_id].audio_transcriptions(input_audio_path, 'whisper-1') + is_successful, response, error_message = model_management[ + user_id].audio_transcriptions(input_audio_path, 'whisper-1') if not is_successful: raise Exception(error_message) memory.append(user_id, 'user', response['text']) - is_successful, response, error_message = model_management[user_id].chat_completions(memory.get(user_id), 'gpt-3.5-turbo') + is_successful, response, error_message = model_management[ + user_id].chat_completions(memory.get(user_id), 'gpt-3.5-turbo') if not is_successful: raise Exception(error_message) role, response = get_role_and_content(response) memory.append(user_id, role, response) - msg = TextSendMessage(text=response) + msg = TextMessage(text=response) except ValueError: - msg = TextSendMessage(text='請先註冊你的 API Token，格式為 /註冊 [API TOKEN]') + msg = TextMessage(text='請先註冊你的 API Token，格式為 /註冊 [API TOKEN]') except KeyError: - msg = TextSendMessage(text='請先註冊 Token，格式為 /註冊 sk-xxxxx') + msg = TextMessage(text='請先註冊 Token，格式為 /註冊 sk-xxxxx') except Exception as e: memory.remove(user_id) if str(e).startswith('Incorrect API key provided'): - msg = TextSendMessage(text='OpenAI API Token 有誤，請重新註冊。') + msg = TextMessage(text='OpenAI API Token 有誤，請重新註冊。') else: - msg = TextSendMessage(text=str(e)) + msg = TextMessage(text=str(e)) os.remove(input_audio_path) - line_bot_api.reply_message(event.reply_token, msg) + line_bot_api.reply_message_with_http_info( + ReplyMessageRequest(reply_token=event.reply_token, messages=[msg])) @app.route("/", methods=['GET'])