-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathvercel_flask-linebot-chat_gemini_behavior.py
169 lines (137 loc) · 5.56 KB
/
vercel_flask-linebot-chat_gemini_behavior.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
"""
Author : ChungYi Fu (Kaohsiung, Taiwan) 2024/11/11 22:00
https://www.facebook.com/francefu
設定requirements.txt
Flask==3.0.3
"""
# 引入 Flask 框架及其相關模組
from flask import Flask, jsonify, request
# 引入 http.client 模組,用於發送 HTTP 請求
import http.client
# 引入 json 模組,用於處理 JSON 資料
import json
# 匯入 re 模組,用於正則表達式操作
import re
app = Flask(__name__)
# 設定專案標題
app.config['Project_title'] = '法蘭斯的ChatBot'
# Line Bot CHANNEL ACCESS TOKEN
CHANNEL_ACCESS_TOKEN = "iSHcOEyq3PM0Oe/PvgmCY69jAOdW6cWoj1Zn34VFgduEJzGlWWcZnAo6LjMt3L9EAldBB4erN2yt/E5tXQNQt7jSyOGKseY2jPD8czzI/RURL7jF/PBKtGm8PwIIymzXcqqdZLP2zaKRhliJUIMdJQdB04t89/1O/w1cDnyilFU=123"
# Gemini Key
GeminiKey = "DLcdVbFKegN3kbGnUidgBBMCaxg7MqC7deOwDP8123"
# Gemini Assistant Behavior
geminiBehavior = "請分析對話中的情境決定是否開關電燈。\n(1)若判斷情境可能需要開電燈回傳1、可能需要關電燈回傳0,無法判斷或無關或表達模糊回傳-1。\n(2)回傳內容只能是1、0、-1其中之一。\n(3)請不要多做解釋!\n\n\n\n"
# 定義一個路由 /about 觸發此函數
@app.route('/about')
def about():
return app.config['Project_title']
# 定義一個路由 / 觸發此函數
@app.route("/", methods=['POST'])
def home():
if request.method == 'POST':
msg = request.get_json()
try:
if msg and 'events' in msg and len(msg['events']) > 0:
for event in msg['events']:
reply_token = event['replyToken']
# 回傳request資料 reply_message_to_line_bot(CHANNEL_ACCESS_TOKEN, reply_token, str(msg))
linebot_response = handle_event(event)
if linebot_response:
reply_message = [
{
'type': 'text',
'text': linebot_response
}
]
reply_message_to_line_bot(CHANNEL_ACCESS_TOKEN, reply_token, reply_message)
except Exception as e:
# 捕捉並回應錯誤訊息
return jsonify({"data": msg, "error": str(e)}), 200
return "OK", 200
def reply_message_to_line_bot(access_token, reply_token, reply_message):
host = 'api.line.me'
endpoint = '/v2/bot/message/reply'
headers = {
'Content-Type': 'application/json; charset=UTF-8',
'Authorization': f'Bearer {access_token}'
}
payload = {
'replyToken': reply_token,
'messages': reply_message
}
# 建立 HTTP 連線
conn = http.client.HTTPSConnection(host)
# 發送 POST 請求
conn.request("POST", endpoint, body=json.dumps(payload), headers=headers)
# 獲取回應
response = conn.getresponse()
status_code = response.status
conn.close()
return status_code
def handle_event(event):
event_type = event['type']
if event_type == 'message':
return handle_message(event)
else:
return ""
def handle_message(event):
# 處理訊息事件
user_message = event['message']['text'].strip()
user_message_type = event['message']['type']
user_id = event['source']['userId']
group_id = event['source'].get('groupId', '請將Line bot加入群組後於群組輸入gid重新查詢')
room_id = event['source'].get('roomId', '請將Line bot加入聊天室後於聊天室輸入rid重新查詢')
if user_message_type == 'text':
if user_message.lower() == "help":
return f"我是Gemini聊天機器人。\n\n輸入id:查詢userId\n輸入gid:查詢groupId\n輸入rid:查詢roomId\n\n助手行為:\n{geminiBehavior}"
elif user_message.lower() == "id":
return user_id
elif user_message.lower() == "gid":
return group_id
elif user_message.lower() == "rid":
return room_id
else:
return handle_gemini(user_message, GeminiKey)
else:
return ""
def handle_gemini(message, key):
try:
# 設定請求的 URL
url = f"/v1beta/models/gemini-1.5-flash-latest:generateContent?key={key}"
host = "generativelanguage.googleapis.com"
# 構建請求的資料
payload = {
"contents": [
{
"parts": [
{
"text": f"{geminiBehavior}\n{message}"
}
]
}
]
}
# 設定 HTTP 請求
headers = {
'Content-Type': 'application/json; charset=utf-8'
}
# 建立 HTTP 連接
conn = http.client.HTTPSConnection(host)
conn.request("POST", url, body=json.dumps(payload), headers=headers)
# 獲取 HTTP 回應
response = conn.getresponse()
response_data = response.read().decode('utf-8')
conn.close()
# 解析 JSON 回應
json_response = json.loads(response_data)
response_text = json_response["candidates"][0]["content"]["parts"][0]["text"]
if response_text == "null":
response_text = json_response["error"]["message"]
return re.sub(r'[\n\r\t]', '', response_text).strip()
except Exception as e:
return str(e)
'''
# Vercel環境下不用加入以下程式碼
if __name__ == '__main__':
app.run(debug=True)
'''