爬取空氣品質指標 ( AQI )
這篇文章會使用 Python 的 Requests 函式庫,藉由政府資料開放平臺的空氣品質指標 ( AQI ) 的 API,實作一個可以自動抓取空氣品質指標數值的網路爬蟲,並進一步使用 CSV 儲存資料。
本篇使用的 Python 版本為 3.7.12,所有範例可使用 Google Colab 實作,不用安裝任何軟體 ( 參考:使用 Google Colab )
關於政府資料開放平臺
政府資料開放平臺是一個政府提供各種開放資料的管道,讓民眾可以在政府資源有限下,善用無限的創意,整合運用開放資料,推動政府資料開放加值應用,發展出各項跨機關便民服務。
政府資料開放平臺:https://data.gov.tw/
關於空氣品質指標 ( AQI )
從政府資料開放平臺裡,搜尋「空氣品質指標」,就能開啟空氣品質指標 ( AQI ) 的資料頁面,資料有提供 JSON 格式與 CSV 格式,這個範例中會使用 JSON 格式,點擊對應的按鈕,就可以開啟對應的 JSON API 內容。
- 空氣品質指標 ( AQI ) :https://data.gov.tw/dataset/40448
- JSON API:檔案連結
開啟 API 後,可以從內容架構裡,找到 records 的「鍵」,records 的內容由串列和字典 ( 或稱物件和陣列 ) 所組成,包含地點 SiteName、城市 County、AQI、PM2.5...等空氣品質指標的數值。
爬取空氣品質指標
使用 Requests 函式庫的 get 的方法,抓取空氣品質指標的內容,接著取出 records 的內容,由於 records 的內容為串列和字典所組成,直接使用 for 迴圈將結果印出。
import requests
# 2022/12 時氣象局有修改了 API 內容,將部份大小寫混合全改成小寫,因此程式碼也跟著修正
url = 'https://data.epa.gov.tw/api/v2/aqx_p_432?api_key=e8dd42e6-9b8b-43f8-991e-b3dee723a52d&limit=1000&sort=ImportDate%20desc&format=JSON'
data = requests.get(url) # 使用 get 方法透過空氣品質指標 API 取得內容
data_json = data.json() # 將取得的檔案轉換為 JSON 格式
for i in data_json['records']: # 依序取出 records 內容的每個項目
print(i['county'] + ' ' + i['sitename'], end=',') # 印出城市與地點名稱
print('AQI:' + i['aqi'], end=',') # 印出 AQI 數值
print('空氣品質' + i['status']) # 印出空氣品質狀態
使用 CSV 儲存資料
能取得空氣品質指標後,下一步可利用 Python 內建的 CSV 標準函式庫,將資料儲存為 CSV 檔案,下方的程式執行後,會先指定 Colab 執行的目錄,接著建立一個名為 csv-aqi 的 CSV 檔案,再將資料轉換為二維陣列,寫入 CSV。
參考:CSV 檔案操作、Colab 連動 Google Drive、檔案操作 os - chdir(path)、內建函式 ( 檔案讀寫 open )、for 迴圈
import requests
import csv
import os
os.chdir('/content/drive/MyDrive/Colab Notebooks') # 針對 Colab 改變路徑
csvfile = open('csv-aqi.csv', 'w') # 建立空白並可寫入的 CSV 檔案
csv_write = csv.writer(csvfile) # 設定 csv_write 為寫入
url = 'https://data.epa.gov.tw/api/v2/aqx_p_432?api_key=e8dd42e6-9b8b-43f8-991e-b3dee723a52d&limit=1000&sort=ImportDate%20desc&format=JSON'
data = requests.get(url)
data_json = data.json()
output = [['county','sitename','aqi','空氣品質']] # 設定 output 變數為二維串列,第一筆資料為開頭
for i in data_json['records']:
# 依序將取得的資料加入 output 中
output.append([i['county'],i['sitename'],i['aqi'],i['status']])
print(output)
csv_write.writerows(output) # 多行寫入 CSV
意見回饋
如果有任何建議或問題,可傳送「意見表單」給我,謝謝~