旅行時間分析專案系列 — 02: 資料爬蟲

source: https://pixabay.com/photos/code-html-digital-coding-web-1076536/

[English / 中文] 如果你不是會員，你可以透過此鏈結瀏覽文章

這是旅行時間分析專案系列文章第二篇，專案目標是利用事故、道路施工、節日等相關資料進行時間序列分析與預測，找出適合台灣特定國道方向下各路段的旅行時間預測的神經網路模型。如果你想了解更多專案的起源想法，請參考系列文章第一篇 旅行時間分析專案系列 — 01: 簡介 。

資料爬蟲

國道旅行時間資料以及專案分析上用來輔助建模或是觀測的資料都需要從線上取得；本篇文章會介紹如何從相關來源抓取數據，並提供核心爬蟲程式的應用方式。以下幾點將逐一說明：

• 資料來源：資料從哪裡來，其內容與特點是什麼？
• 核心爬蟲程式: 簡要介紹個人撰寫的爬蟲程式。
• 資料抓取過程: 實際抓取資料的步驟與應用。
• 其他參考資料: 在未必必要的情況下，也可以考量的補充來源。

資料有哪些?

官方推薦

競賽官方建議與提供的資料集相當廣泛，從來源可以先推演一遍潛在的應用:

• 高速公路局交通資料庫: 主要的旅行時間、車速、流量等資料，是必備的基礎資料
• 易壅塞路段資料: 整理過的表格文件，也許可以用來設計簡易的壅塞判斷標籤
• 連續假期間國道疏導措施檢討報告: 措施會被檢討，一定程度可從文字報告中推論特殊節日下的交通狀況，或許也可用來設計以強化連續假期的嚴重性?
• 交通部觀光資訊網: 可用來檢視各時間的訂房率、統計地區性重要景點，越多景點在節日下也許影響會很大?
• 氣象資料開放平台: 常見的氣象資料對於車速會有影響
• 台灣中油油價: 這通常是一個較長遠的經濟因素，也許可以發揮創意應用在模型上?

source: 113國道交管競賽-時間預測分析與探討 (https://freeway2024.tw/links#links)

實際採用資料

雖然資料種類繁多，考慮到時間及根據個人知識的理解，僅採用以下幾個資料來源作為本專案參考的資訊，應用在模型訓練與分析上:

• 112年1–10月交通事故簡訊通報狀況資料 (https://freeway2024.tw/112年1-10月交通事故簡訊通報資料.xlsx)
• 112 年 1–10 月施工路段資料 (https://freeway2024.tw/112年1-10月道路施工路段資料.xlsx)
• 2023年重要節日行事曆 (自行整理)
• 國道易壅塞路段 (https://data.gov.tw/dataset/33191)
• 交通部高速公路局交通資料庫 (https://tisvcloud.freeway.gov.tw/, M04A資料源為重點，實際採用的資料請見後續的專案notebook)

(資料爬蟲、清洗、儲存等都需要耗費大量時間，專案訂定前若能釐清架構，後續要橫向或垂直擴充就會變得比較容易)

核心爬蟲程式解說

主要的程式碼請參考 **data_scraper.py**

https://github.com/willim9313/Taiwan_Freeway5_TravelTime_Prediction/blob/main/hwttp/data_scraper.py

這邊僅介紹專案上主要採用的funcion:

• generate_date_strings: 生成開始與結尾日期用的字串生成式
• download_file: 下載單個文件，並存放到指定資料夾位置
• download_files: 一次性下載多個文件，並依序放置到指定資料夾位置
• scrape_etag_intergantry_traveltime: 指定一連串的時間資訊下進行ETag gantry間通過所花費的旅行時間為主的資料爬取(M04A 資料集)
• scrape_etag_info: 指定一連串的時間資訊下進行ETag資料的爬取，原始資料有區分動態與靜態資訊，可透過指定參數來決定下載目標

生成時間字串清單

這部份很簡單，我想要提供一個彈性的方法去生成指定日期範圍的清單，後面只需要將此清單pass給爬蟲程式，讓程式依據時間範圍去抓取資料即可

# ./hwttp/data_scraper.py
from datetime import datetime, timedelta

def generate_date_strings(start_date: str, 
                          end_date: str) -> list:
    '''
    datetime range generator

    Parameters
    ----------
    start_date : str
        e.g. '20230101'
    end_date : str
        e.g. '20230201'

    Returns
    -------
    date_list : list
        包含一連串像是'20230101'格式的日期

    '''
    start = datetime.strptime(start_date, '%Y%m%d')
    end = datetime.strptime(end_date, '%Y%m%d')
    date_list = []

    current_date = start
    while current_date <= end:
        date_list.append(current_date.strftime('%Y%m%d'))
        current_date += timedelta(days=1)

    return date_list

檔案下載

文件下載的方法不複雜，我們只要指定下載的URL跟目標存放的位置即可，這邊在實踐上是分成單檔與多檔處理，多檔的部分又會應用到單檔下載的程式碼，只是多了層迴圈去處理

def download_file(url: str, 
                  save_path: str) -> None:
    '''
    單一文件下載處理

    Parameters
    ----------
    url: str
        目標文件的url
    save_path: str
        文件下載後儲存的路徑

    Return
    -----
    None
    '''
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('content-length', 0))
    block_size = 1024  # 1 Kilobyte

    if response.status_code != 200:
        print(f"Failed to download {url}")
        return

    with open(save_path, 'wb') as file:
        for data in response.iter_content(block_size):
            file.write(data)

def download_files(base_url: str, 
                   file_names: list[str], 
                   save_dir: str) -> None:
    '''
    一次性下載多個文件用途，會根據url與file_name組出下載的目標
    並存放到save_dir中

    Parameters
    ----------
    base_url: str
        目標文件的主要路徑url
    file_names: list[str]
        使用list去包多個文件的名稱，['a1', 'a2', 'a3']
    save_dir: str
        儲存的目標資料夾路徑

    Return
    ------
    None

    '''
    os.makedirs(save_dir, exist_ok=True)
    for file_name in file_names:
        try:
            url = f"{base_url}{file_name}"
            save_path = os.path.join(save_dir, file_name)
            # print(f"Downloading {url} to {save_path}")
            download_file(url, save_path)
            # print(f"Downloaded {file_name} to {save_path}")
        except:
            print(f"Downloading {url} to {save_path} facing errors")

ETag旅行時間資料抓取

主要的資料都是在base_url 的路徑下，只需要依據我們需求的時間段去逐步下載就可以了，目前設定成下載資訊完畢後sleep 1秒，建議視實際狀況去彈性展延時間，在短時間內過於頻繁地去抓取有機會讓我們的請求被封殺。

def scrape_etag_intergantry_traveltime(date_list: list[str],
                                       save_dir: str='../data/raw/ETag_intergantry_traveltime') -> None:
    '''
    可指定一連串的時間資訊下進行ETag gantry間通過所花費的旅行時間為主資料的爬取

    Parameters
    ----------
    date_list: list[str]
        輸入日期串，像是['20230101', '20230102']
    save_dir: str
        資料下載後儲存的路徑，如果不存在會自動建立folder

    Return
    ------
    None
    '''

    base_url = f"<https://tisvcloud.freeway.gov.tw/history/TDCS/M04A/>"

    create_directory_if_not_exists(save_dir)

    for date in tqdm(date_list):
        file_name = f'M04A_{date}.tar.gz'
        if os.path.exists(save_dir+'/'+file_name): # 跳過已經有檔案的部分
            continue
        download_files(base_url, [file_name], save_dir)
        time.sleep(1)

ETag資訊抓取

ETag資訊是確認gantry所在座標用的資訊，正常狀況下不會有太多的變化；原始資料源又分靜態資訊跟配對的路徑資訊(動態與靜態皆有)，後者是因旅次產生而出現的路徑資訊，可視個人需求決定是否要納入分析範圍，function設計上增加一個select_data 來決定抓取的資料內容採用哪一種。

def scrape_etag_info(date_list: list[str],
                     save_dir_upper: str='../data/raw/ETag/',
                     select_data: str='all') -> None:
    '''
    可指定一連串的時間資訊下進行ETag資料的爬取，原始資料有區分動態與靜態資訊，可透過指定參數來決定下載目標

    Parameters
    ----------
    date_list: list[str]
        輸入日期串，像是['20230101', '20230102']
    save_dir_upper: str
        資料爬取存放的上層路徑，其下日期的子資料夾會自動生成
    select_data: str
        原始資料有區分靜態、動態，'all'= 全數下載, 'info'=靜態資訊下載, 'pair_static'= 配對靜態下載, 'pair_dynamic' = 配對動態下載  
        eTag 靜態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETag_0000.xml.gz  
        eTag 配對路徑靜態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETagPair_0000.xml.gz  
        eTag 配對路徑動態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETagPairLive_HHmm.xml.gz  

    Return
    ------
    None
    '''
    if select_data not in ['all', 'info', 'pair_static', 'pair_dynamic']:
        return 'select_data need to specify, please check the docstring.'

    for exact_date in tqdm(date_list):
        # 基本爬取的日期，以及預計要存放的路徑 
        base_url = f"<https://tisvcloud.freeway.gov.tw/history/motc20/ETag/{exact_date}/>"
        save_dir = f'{save_dir_upper}{exact_date}'

        create_directory_if_not_exists(save_dir)

        # 爬取該日期頁面下所有合理的
        response = requests.get(base_url)
        soup = BeautifulSoup(response.text, 'html.parser')
        file_paths = []
        for i in soup.find_all('a'):
            if i.get('href').endswith('.xml.gz'):
                file_paths.append(i.get('href'))

        # 這種抓法會有名稱重複，因此要去重、排序
        # 裡面會有3種的ETag資訊
        # eTag 靜態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETag_0000.xml.gz 
        # eTag 配對路徑靜態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETagPair_0000.xml.gz
        # eTag 配對路徑動態資訊(v2.0) = ./history/motc20/ETag/YYYYMMDD/ETagPairLive_HHmm.xml.gz

        file_paths = list(set(file_paths))
        file_paths.sort()

        # 排除或保留指定的資料類型
        if select_data =='info':
            file_paths = [file_path for file_path in file_paths if 'ETag_' in file_path]
        elif select_data == 'pair_static':
            file_paths = [file_path for file_path in file_paths if 'ETagPair_' in file_path]
        elif select_data == 'pair_dynamic':
            file_paths = [file_path for file_path in file_paths if 'ETagPairLive_' in file_path]
        else:
            pass

        for file_name in file_paths:
            if os.path.exists(save_dir+'/'+file_name): # 跳過已經有檔案的部分
                continue
            download_files(base_url, [file_name], save_dir)
            time.sleep(1)

實際資料爬取應用

在使用之前，要先import好對應的python file，再用ds 來調用上述的function。比方說我們需要抓取指定日期間地資料的話，只要決定好起訖的字串就可以生成日期清單，再把對應的清單塞給我們要爬取資訊的function即可；如果沒有提供資料存放的路徑，就會採用預設的路徑自動建立。

t sys
sys.path.append(os.path.abspath('..'))
import hwttp.data_scraper as ds

# 決定日期範圍
date_list = []
start_date = '20231101'
end_date = '20231102'
date_list = ds.generate_date_strings(start_date, end_date)

# ETag資訊抓取
ds.scrape_etag_info(date_list, select_data='pair_static')

# ETag旅行時間資料抓取
ds.scrape_etag_intergantry_traveltime(date_list)

更多詳細的應用方式與說明請參考以下的notebook

[embed]Taiwan_Freeway5_TravelTime_Prediction/notebooks/web_scraping.ipynb at main ·… A time series analysis and prediction project for travel time on Taiwan Freeway №5 using machine learning…github.com

其他可參考資料

VD資料的擷取沒有在本文章介紹，本專案實作上最後也沒有採納，但這部分的數據也有可發揮的空間，想知道怎麼樣獲取，可以參考 notebook的語法。那麼為何不取用VD的資料呢? 因為資料量比ETag更大，地理空間上所標記的位置更多，更具挑戰性；另外，VD資料的乾淨程度根據經驗更不穩定，採用VD資料意味著我們要花更多的心力在資料清洗與加工上。其他參考資料，比方說中油的歷史油價資訊，使用BeautifulSoup 跟requests作為我們的爬蟲工具，只需要幾行就可以把資料取下來轉換成dataframe了。

這個專案資料爬取的部分本身不難，多是TSIV提供的靜態資料，且採用結構化的路徑存放，對比其他像是從電商網站上取得特定資訊而言已經是小菜一碟了，接下來我們即將進入燒腦的環節 — Data Cleaning (資料清洗)

註: 本文是先用中文撰寫，再由ChatGPT翻譯後經由個人調整，語意上可能稍有出入

Travel Time Analysis Project Series

• Overview and Introduction
• Data Crawling (資料爬蟲)
• Data Cleaning (資料清洗)
• Data Exploration (資料探索)
• Preliminary EDA (初步探索性數據分析)
• EDA for National Freeway №5 (國道五號的探索性數據分析)
• Time Series Analysis and Modeling (時間序列分析與建模)
• Baseline Modeling (基線模型建模)
• Univariate Modeling (單變量建模)
• Feature Selection (特徵篩選)
• Multivariate Modeling (多變量建模)
• Cross-Validation and Performance Analysis (交叉驗證與成果分析)
• Closing Remarks

如果你喜歡這篇文章，你可以：

• 👏 拍個幾下手就好，不需要按到 50 下
• ✉️ 在 Medium 上追蹤我，這會給我更多動力繼續創作
• 😜 歡迎加我 LinkedIn，一起交流想法