forked from maso0310/linebot_openai
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp.py
209 lines (192 loc) · 9.56 KB
/
app.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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import *
import os
import json
import time
import requests
import traceback
app = Flask(__name__)
# Channel Access Token and Channel Secret (從環境變數取得或自行設定)
access_token = os.getenv('CHANNEL_ACCESS_TOKEN', '你的 Access Token')
channel_secret = os.getenv('CHANNEL_SECRET', '你的 Channel Secret')
weather_api = os.getenv('WEATHER_API', '你的 Weather API')
# Line API 初始化
line_bot_api = LineBotApi(access_token)
handler = WebhookHandler(channel_secret)
# 你的天氣 API 授權碼
code = weather_api
# 地震查詢功能，整合中央氣象局地震資料的 API
def earth_quake():
result = []
try:
# 小區域地震
url1 = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0016-001?Authorization={code}'
req1 = requests.get(url1)
data1 = req1.json()
eq1 = data1['records']['Earthquake'][0]
t1 = eq1['EarthquakeInfo']['OriginTime']
# 顯著有感地震
url2 = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={code}'
req2 = requests.get(url2)
data2 = req2.json()
eq2 = data2['records']['Earthquake'][0]
t2 = eq2['EarthquakeInfo']['OriginTime']
# 使用最新的地震資料
result = [eq1['ReportContent'], eq1['ReportImageURI']] # 使用小區域地震
if t2 > t1:
result = [eq2['ReportContent'], eq2['ReportImageURI']] # 如果顯著有感地震時間較近
except Exception as e:
print(e)
result = ['抓取失敗...', ''] # 如果發生錯誤，返回失敗訊息
return result
# 取得某個地點的氣象資訊，整合氣象預報與空氣品質
def weather(address):
result = {}
# 即時天氣
try:
url = [f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0001-001?Authorization={code}',
f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001?Authorization={code}']
for item in url:
req = requests.get(item) # 爬取目前天氣網址的資料
data = req.json()
station = data['records']['Station']
for i in station:
city = i['GeoInfo']['CountyName']
area = i['GeoInfo']['TownName']
if not f'{city}{area}' in result:
weather = i['WeatherElement']['Weather']
temp = i['WeatherElement']['AirTemperature']
humid = i['WeatherElement']['RelativeHumidity']
result[f'{city}{area}'] = f'目前天氣狀況「{weather}」，溫度 {temp} 度，相對濕度 {humid}%！'
except:
pass
# 氣象預報
api_list = {"宜蘭縣":"F-D0047-001","桃園市":"F-D0047-005","新竹縣":"F-D0047-009","苗栗縣":"F-D0047-013",
"彰化縣":"F-D0047-017","南投縣":"F-D0047-021","雲林縣":"F-D0047-025","嘉義縣":"F-D0047-029",
"屏東縣":"F-D0047-033","臺東縣":"F-D0047-037","花蓮縣":"F-D0047-041","澎湖縣":"F-D0047-045",
"基隆市":"F-D0047-049","新竹市":"F-D0047-053","嘉義市":"F-D0047-057","臺北市":"F-D0047-061",
"高雄市":"F-D0047-065","新北市":"F-D0047-069","臺中市":"F-D0047-073","臺南市":"F-D0047-077",
"連江縣":"F-D0047-081","金門縣":"F-D0047-085"}
for name in api_list:
if name in address:
city_id = api_list[name]
t = time.time()
t1 = time.localtime(t+28800)
t2 = time.localtime(t+28800+10800)
now = time.strftime('%Y-%m-%dT%H:%M:%S',t1)
now2 = time.strftime('%Y-%m-%dT%H:%M:%S',t2)
url = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/{city_id}?Authorization={code}&elementName=WeatherDescription&timeFrom={now}&timeTo={now2}'
req = requests.get(url) # 取得主要縣市預報資料
data = req.json() # json 格式化訊息內容
location = data['records']['locations'][0]['location']
city = data['records']['locations'][0]['locationsName']
for item in location:
try:
area = item['locationName']
note = item['weatherElement'][0]['time'][0]['elementValue'][0]['value']
if not f'{city}{area}' in result:
result[f'{city}{area}'] = ''
else:
result[f'{city}{area}'] = result[f'{city}{area}'] + '。\n\n'
result[f'{city}{area}'] = result[f'{city}{area}'] + '未來三小時' + note
except:
pass
# 空氣品質
try:
url = 'https://data.moenv.gov.tw/api/v2/aqx_p_432?api_key=e8dd42e6-9b8b-43f8-991e-b3dee723a52d&limit=1000&sort=ImportDate%20desc&format=JSON'
req = requests.get(url)
data = req.json()
records = data['records']
for item in records:
county = item['county'] # 縣市
sitename = item['sitename'] # 區域
name = f'{county}{sitename}'
aqi = int(item['aqi']) # AQI 數值
aqi_status = ['良好','普通','對敏感族群不健康','對所有族群不健康','非常不健康','危害']
msg = aqi_status[aqi//50] # 除以五十之後無條件捨去，取得整數
for k in result:
if name in k:
result[k] = result[k] + f'\n\nAQI：{aqi}，空氣品質{msg}。'
except:
pass
output = '找不到氣象資訊'
for i in result:
if i in address: # 如果地址裡存在 key 的名稱
output = f'「{address}」{result[i]}'
break
return output
def cctv(msg):
try:
output = ''
camera_list = {
'夢時代':'https://cctv1.kctmc.nat.gov.tw/27e5c086/',
'鼓山渡輪站':'https://cctv3.kctmc.nat.gov.tw/ddb9fc98/',
'中正交流道':'https://cctv3.kctmc.nat.gov.tw/166157d9/',
'五福愛河':'https://cctv4.kctmc.nat.gov.tw/335e2702/'
}
for item in camera_list:
if msg == item:
output = camera_list[msg]
except Exception as e:
print(e)
return output
# 主要的 LINE Bot 路由處理程式
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True) # 取得收到的訊息內容
try:
line_bot_api = LineBotApi(access_token) # 確認 token 是否正確
handler = WebhookHandler(channel_secret) # 確認 secret 是否正確
signature = request.headers['X-Line-Signature'] # 加入回傳的 headers
handler.handle(body, signature) # 綁定訊息回傳的相關資訊
json_data = json.loads(body) # 轉換內容為 json 格式
reply_token = json_data['events'][0]['replyToken'] # 取得回傳訊息的 Token ( reply message 使用 )
user_id = json_data['events'][0]['source']['userId'] # 取得使用者 ID ( push message 使用 )
print(json_data) # 印出內容
type = json_data['events'][0]['message']['type']
if type == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達':
#line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....'))
img_url = f'https://cwaopendata.s3.ap-northeast-1.amazonaws.com/Observation/O-A0058-001.png?{time.time_ns()}'
img_message = ImageSendMessage(original_content_url=img_url, preview_image_url=img_url)
line_bot_api.reply_message(reply_token,img_message)
elif text == '地震':
#line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....'))
reply = earth_quake()
text_message = TextSendMessage(text = reply[0])
line_bot_api.reply_message(reply_token, text_message)
line_bot_api.push_message(user_id, ImageSendMessage(original_content_url = reply[1], preview_image_url = reply[1]))
else:
reply = cctv(text)
if not reply == '':
text_message = TextSendMessage(text=reply)
line_bot_api.reply_message(reply_token,text_message)
sec = math.ceil(time.time())
reply = reply + f'snapshot?t={sec}'
line_bot_api.push_message(user_id, ImageSendMessage(original_content_url=reply, preview_image_url=reply))
else:
text_message = TextSendMessage(text=text)
#line_bot_api.reply_message(reply_token,text_message)
elif type == 'location':
#line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....'))
address = json_data['events'][0]['message']['address'].replace('台','臺') # 取出地址資訊，並將「台」換成「臺」
reply = weather(address)
text_message = TextSendMessage(text=reply)
line_bot_api.reply_message(reply_token,text_message)
except Exception as e:
print(e)
return 'OK' # 驗證 Webhook 使用，不能省略
# 歡迎新成員加入的事件
@handler.add(MemberJoinedEvent)
def welcome(event):
uid = event.joined.members[0].user_id
gid = event.source.group_id
profile = line_bot_api.get_group_member_profile(gid, uid)
name = profile.display_name
message = TextSendMessage(text=f'{name} 歡迎加入')
line_bot_api.reply_message(event.reply_token, message)
if __name__ == "__main__":
app.run()