Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update LINE Bot SDK from deprecated to latest v3 #50

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Update LINE Bot SDK from deprecated to latest v3 #50

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 53 additions & 42 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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)
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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'])
Expand Down