""" 台灣停車場即時資料收集器 支援台北市、新北市、桃園市、台中市、台南市、高雄市 使用方式： python -X utf8 collector.py # 執行一次收集 python -X utf8 collector.py --daemon # 持續執行（每 30 分鐘） python -X utf8 collector.py --test # 測試模式（只抓一次，不存 DB） CHANGELOG: ----------- 2026-02-04: 資料庫優化 - 移除冗餘欄位 - parking_availability 不再寫入 park_name, city（可從 parking_static 查詢） - 原因：減少儲存空間約 30%（每筆省 ~25 bytes） - 向後相容：schema 不變，舊資料保留，新資料寫入 NULL - 查詢時需 JOIN parking_static 取得名稱 2026-02-04: Bug 修復 - 跨日排程錯誤 - 修復 get_next_scheduled_time() 在 23:30 時 hour+1=24 的錯誤 - 改用 timedelta(hours=1) 處理跨日 2026-02-04: 新增統一監控回報功能 - 整合 shared/status.py 模組 - 每次收集後自動向雲端監控服務回報狀態 - 需設定環境變數 COLLECTOR_MONITOR_API 和 COLLECTOR_MONITOR_TOKEN TODO: - 雲林縣資料異常：僅 1 個停車場，每小時使用率固定 68.2%（一條直線）， 疑似靜態/無感測器資料。Dashboard 已預設隱藏，需調查該 API 是否有效。 - 新北市 API 不回傳停車場名稱，需另找資料來源補充。 """ import sys sys.stdout.reconfigure(encoding='utf-8') import os import json import sqlite3 import requests import logging import random import time import argparse from datetime import datetime from pathlib import Path # 狀態回報模組（選用） try: sys.path.insert(0, str(Path(__file__).parent.parent / "shared")) from status import StatusReporter HAS_STATUS_REPORTER = True except ImportError: HAS_STATUS_REPORTER = False StatusReporter = None # ============================================================ # 設定 # ============================================================ # 資料儲存在 D 槽，避免影響 Research_zoo 版本控制 DATA_DIR = Path("D:/data-collectors/parking") DB_PATH = DATA_DIR / "data" / "parking.db" LOG_PATH = DATA_DIR / "logs" / "collector.log" # 抓取間隔（秒）：30 分鐘 ± 60 秒 BASE_INTERVAL = 1800 # 30 分鐘 INTERVAL_JITTER = 60 # ± 60 秒 # 請求之間的延遲（秒） REQUEST_DELAY_MIN = 2 REQUEST_DELAY_MAX = 5 # 請求逾時（秒） REQUEST_TIMEOUT = 30 # 重試設定 MAX_RETRIES = 3 RETRY_BACKOFF = 5 # 重試間隔基數（秒） # User-Agent（模擬一般瀏覽器） USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" # ============================================================ # API 端點設定 # ============================================================ CITY_APIS = { "taipei": { "name": "台北市", "static_url": "https://tcgbusfs.blob.core.windows.net/blobtcmsv/TCMSV_alldesc.json", "dynamic_url": "https://tcgbusfs.blob.core.windows.net/blobtcmsv/TCMSV_allavailable.json", "parser": "taipei", }, "newtaipei": { "name": "新北市", "dynamic_url": "https://data.ntpc.gov.tw/api/datasets/e09b35a5-a738-48cc-b0f5-570b67ad9c78/json?size=2000", "parser": "newtaipei", "update_freq": "每 3 分鐘", }, "taoyuan": { "name": "桃園市", "dynamic_url": "https://data.tycg.gov.tw/api/v1/rest/datastore/0daad6e6-0632-44f5-bd25-5e1de1e9146f?format=json", "backup_url": "http://data.tycg.gov.tw/api/v1/rest/datastore/0daad6e6-0632-44f5-bd25-5e1de1e9146f?format=json", "parser": "taoyuan", "update_freq": "每 1 分鐘", }, "taichung": { "name": "台中市", "dynamic_url": "https://motoretag.taichung.gov.tw/DataAPI/api/ParkingSpotListAPIV2", "parser": "taichung", "update_freq": "每 10 分鐘", }, "tainan": { "name": "台南市", "dynamic_url": "https://parkweb.tainan.gov.tw/api/parking.php", "parser": "tainan", }, "kaohsiung": { "name": "高雄市", "dynamic_url": "https://openapi.kcg.gov.tw/Api/Service/Get/897e552a-2887-4f6f-a6ee-709f7fbe0ee3", "parser": "kaohsiung", }, } # ============================================================ # 日誌設定 # ============================================================ def setup_logging(): """設定日誌""" LOG_PATH.parent.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_PATH, encoding="utf-8"), logging.StreamHandler(sys.stdout), ], ) return logging.getLogger(__name__) logger = setup_logging() # ============================================================ # 資料庫 # ============================================================ def init_db(): """初始化資料庫""" DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH) c = conn.cursor() # 停車場即時資料表 c.execute(""" CREATE TABLE IF NOT EXISTS parking_availability ( id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT NOT NULL, park_id TEXT NOT NULL, park_name TEXT, total_car INTEGER, available_car INTEGER, occupancy_rate REAL, status TEXT, collect_time DATETIME NOT NULL, source_time DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # 建立索引加速查詢 c.execute(""" CREATE INDEX IF NOT EXISTS idx_parking_city_time ON parking_availability (city, collect_time) """) c.execute(""" CREATE INDEX IF NOT EXISTS idx_parking_park_time ON parking_availability (park_id, collect_time) """) # 停車場靜態資料表（名稱、地址、總車位等） c.execute(""" CREATE TABLE IF NOT EXISTS parking_static ( id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT NOT NULL, park_id TEXT NOT NULL, park_name TEXT, area TEXT, address TEXT, total_car INTEGER, total_motor INTEGER, lat REAL, lng REAL, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(city, park_id) ) """) # 收集記錄表（追蹤每次收集的狀態） c.execute(""" CREATE TABLE IF NOT EXISTS collection_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT NOT NULL, status TEXT NOT NULL, record_count INTEGER, error_message TEXT, duration_ms INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() return conn # ============================================================ # HTTP 請求 # ============================================================ def make_request(url: str, retries: int = MAX_RETRIES, verify_ssl: bool = True) -> dict | None: """發送 HTTP 請求，含重試機制""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json", "Accept-Language": "zh-TW,zh;q=0.9,en;q=0.8", } for attempt in range(retries): try: response = requests.get( url, headers=headers, timeout=REQUEST_TIMEOUT, verify=verify_ssl, ) response.raise_for_status() return response.json() except requests.exceptions.SSLError as e: # SSL 錯誤時，嘗試關閉 SSL 驗證 if verify_ssl: logger.warning(f"SSL 驗證失敗，嘗試關閉 SSL 驗證...") return make_request(url, retries=1, verify_ssl=False) else: logger.error(f"SSL 請求失敗: {e}") return None except requests.exceptions.RequestException as e: wait_time = RETRY_BACKOFF * (attempt + 1) + random.uniform(0, 2) logger.warning(f"請求失敗 (嘗試 {attempt + 1}/{retries}): {e}") if attempt < retries - 1: logger.info(f"等待 {wait_time:.1f} 秒後重試...") time.sleep(wait_time) else: logger.error(f"請求最終失敗: {url}") return None return None # ============================================================ # 資料解析器 # ============================================================ def parse_taipei(data: dict) -> list[dict]: """解析台北市資料""" records = [] parks = data.get("data", {}).get("park", []) for park in parks: available = park.get("availablecar", -9) if available == -9: available = None records.append({ "park_id": park.get("id"), "park_name": park.get("name"), "total_car": None, # 動態資料不含總車位 "available_car": available, "status": str(park.get("availablecarstatus", "")), "source_time": park.get("updatedTime"), }) return records def parse_newtaipei(data: list) -> list[dict]: """解析新北市資料（即時資料只有 ID 和剩餘車位）""" records = [] for park in data: available_str = park.get("AVAILABLECAR", "-9") try: available = int(available_str) if available_str and available_str != "-9" else None except (ValueError, TypeError): available = None records.append({ "park_id": park.get("ID"), "park_name": park.get("NAME"), # 即時資料可能沒有名稱 "total_car": park.get("TOTALCAR"), # 即時資料可能沒有總車位 "available_car": available, "status": None, "source_time": None, }) return records def parse_taoyuan(data: dict) -> list[dict]: """解析桃園市資料""" records = [] parks = data.get("result", {}).get("records", []) for park in parks: surplus = park.get("surplusSpace") try: available = int(surplus) if surplus and surplus != "開放中" else None except (ValueError, TypeError): available = None records.append({ "park_id": park.get("parkId"), "park_name": park.get("parkName"), "total_car": park.get("totalSpace"), "available_car": available, "status": surplus if not isinstance(available, int) else None, "source_time": None, }) return records def parse_taichung(data: list) -> list[dict]: """解析台中市資料（巢狀結構：區域 > ParkingLots）""" records = [] for area in data: parking_lots = area.get("ParkingLots", []) for park in parking_lots: records.append({ "park_id": str(park.get("ID")), "park_name": park.get("Position"), "total_car": park.get("TotalCar"), "available_car": park.get("AvailableCar"), "status": park.get("AvailableCarRGB"), "source_time": park.get("Updatetime"), }) return records def parse_tainan(data: dict) -> list[dict]: """解析台南市資料""" records = [] parks = data if isinstance(data, list) else data.get("data", []) for park in parks: records.append({ "park_id": park.get("id") or park.get("ID"), "park_name": park.get("name") or park.get("NAME"), "total_car": park.get("car_total"), # 總車位 "available_car": park.get("car"), # 剩餘車位 "status": None, "source_time": park.get("update_time"), }) return records def parse_kaohsiung(data: dict) -> list[dict]: """解析高雄市資料（注意：目前只有靜態資料，無即時車位）""" records = [] parks = data.get("data", []) if isinstance(data, dict) else data for park in parks: # 高雄市 API 使用中文欄位名 park_id = park.get("Seq") or park.get("parkno") or park.get("id") park_name = park.get("臨時停車處所") or park.get("parkname") or park.get("name") total_car = park.get("可提供小型車停車位") or park.get("totalcar") # 嘗試轉換為整數 try: total_car = int(total_car) if total_car else None except (ValueError, TypeError): total_car = None records.append({ "park_id": str(park_id) if park_id else None, "park_name": park_name, "total_car": total_car, "available_car": None, # 高雄市目前無即時車位資料 "status": "static_only", # 標記為靜態資料 "source_time": None, }) return records PARSERS = { "taipei": parse_taipei, "newtaipei": parse_newtaipei, "taoyuan": parse_taoyuan, "taichung": parse_taichung, "tainan": parse_tainan, "kaohsiung": parse_kaohsiung, } # ============================================================ # 靜態資料收集 # ============================================================ def fetch_static_data(city_key: str, city_config: dict) -> list[dict]: """取得城市的靜態停車場資料""" static_url = city_config.get("static_url") if not static_url: return [] data = make_request(static_url) if not data: return [] if city_key == "taipei": return parse_taipei_static(data) return [] def parse_taipei_static(data: dict) -> list[dict]: """解析台北市靜態資料""" records = [] parks = data.get("data", {}).get("park", []) for park in parks: # 取得座標（優先用 WGS84） lat, lng = None, None entrance = park.get("EntranceCoord", {}).get("EntrancecoordInfo", []) if entrance: lat = float(entrance[0].get("Xcod", 0)) if entrance[0].get("Xcod") else None lng = float(entrance[0].get("Ycod", 0)) if entrance[0].get("Ycod") else None records.append({ "park_id": park.get("id"), "park_name": park.get("name"), "area": park.get("area"), "address": park.get("address"), "total_car": park.get("totalcar"), "total_motor": park.get("totalmotor"), "lat": lat, "lng": lng, }) return records def update_static_data(conn: sqlite3.Connection, city_key: str, records: list[dict]): """更新靜態資料到資料庫""" if not records: return 0 c = conn.cursor() count = 0 for record in records: if not record.get("park_id"): continue c.execute(""" INSERT INTO parking_static (city, park_id, park_name, area, address, total_car, total_motor, lat, lng, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(city, park_id) DO UPDATE SET park_name = excluded.park_name, area = excluded.area, address = excluded.address, total_car = excluded.total_car, total_motor = excluded.total_motor, lat = excluded.lat, lng = excluded.lng, updated_at = CURRENT_TIMESTAMP """, ( city_key, record["park_id"], record.get("park_name"), record.get("area"), record.get("address"), record.get("total_car"), record.get("total_motor"), record.get("lat"), record.get("lng"), )) count += 1 conn.commit() return count def collect_all_static(conn: sqlite3.Connection): """收集所有城市的靜態資料""" logger.info("開始收集靜態資料...") for city_key, city_config in CITY_APIS.items(): if "static_url" not in city_config: continue city_name = city_config["name"] logger.info(f"收集 {city_name} 靜態資料...") records = fetch_static_data(city_key, city_config) if records: count = update_static_data(conn, city_key, records) logger.info(f" {city_name}: {count} 筆靜態資料") else: logger.warning(f" {city_name}: 無法取得靜態資料") time.sleep(random.uniform(1, 2)) # ============================================================ # 資料收集 # ============================================================ def collect_city(city_key: str, city_config: dict, conn: sqlite3.Connection, test_mode: bool = False) -> dict: """收集單一縣市的資料""" city_name = city_config["name"] start_time = time.time() collect_time = datetime.now() logger.info(f"開始收集 {city_name} 資料...") # 取得資料 url = city_config.get("dynamic_url") data = make_request(url) # 若主要 URL 失敗，嘗試備用 URL if data is None and "backup_url" in city_config: logger.info(f"嘗試備用 URL...") data = make_request(city_config["backup_url"]) if data is None: duration = int((time.time() - start_time) * 1000) error_msg = "無法取得資料" if not test_mode: c = conn.cursor() c.execute( "INSERT INTO collection_log (city, status, error_message, duration_ms) VALUES (?, ?, ?, ?)", (city_key, "error", error_msg, duration), ) conn.commit() return {"status": "error", "error": error_msg, "count": 0} # 解析資料 parser = PARSERS.get(city_config["parser"]) if not parser: return {"status": "error", "error": f"找不到解析器: {city_config['parser']}", "count": 0} try: records = parser(data) except Exception as e: logger.error(f"解析失敗: {e}") return {"status": "error", "error": str(e), "count": 0} # 儲存資料 if not test_mode: c = conn.cursor() saved_count = 0 for record in records: # 跳過無效記錄 park_id = record.get("park_id") if not park_id: continue # 計算使用率 # available < 0 是 API 特殊值（-1, -2），表示無資料 # available = 0 且 total 無值時也視為無資料（無法判斷是否真的滿位） total = record.get("total_car") available = record.get("available_car") # 即時 API 沒有 total_car 時，從 parking_static 補充 if not total and available is not None: row = c.execute( "SELECT total_car FROM parking_static WHERE city = ? AND park_id = ?", (city_key, str(park_id)) ).fetchone() if row and row[0]: total = row[0] if (total and available is not None and available >= 0 and total > 0 and available <= total): occupancy_rate = min((1 - available / total) * 100, 100.0) else: occupancy_rate = None c.execute(""" INSERT INTO parking_availability (city, park_id, total_car, available_car, occupancy_rate, status, collect_time, source_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( city_key, park_id, total, available, occupancy_rate, record.get("status"), collect_time.isoformat(), record.get("source_time"), )) saved_count += 1 # 順帶從即時資料更新靜態資料（名稱等） park_name = record.get("park_name") if park_name: c.execute(""" INSERT INTO parking_static (city, park_id, park_name, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(city, park_id) DO UPDATE SET park_name = CASE WHEN excluded.park_name IS NOT NULL AND excluded.park_name != '' THEN excluded.park_name ELSE parking_static.park_name END, updated_at = CURRENT_TIMESTAMP """, (city_key, park_id, park_name)) duration = int((time.time() - start_time) * 1000) c.execute( "INSERT INTO collection_log (city, status, record_count, duration_ms) VALUES (?, ?, ?, ?)", (city_key, "success", saved_count, duration), ) conn.commit() records = [r for r in records if r.get("park_id")] # 更新 records 供後續使用 logger.info(f" {city_name}: {len(records)} 筆資料") return {"status": "success", "count": len(records)} def collect_all(conn: sqlite3.Connection, test_mode: bool = False) -> dict: """收集所有縣市的資料""" logger.info("=" * 50) logger.info(f"開始收集所有縣市資料 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info("=" * 50) results = {} cities = list(CITY_APIS.items()) # 隨機打亂順序，避免每次都是同樣的請求模式 random.shuffle(cities) for i, (city_key, city_config) in enumerate(cities): # 請求之間加入隨機延遲 if i > 0: delay = random.uniform(REQUEST_DELAY_MIN, REQUEST_DELAY_MAX) logger.debug(f"等待 {delay:.1f} 秒...") time.sleep(delay) results[city_key] = collect_city(city_key, city_config, conn, test_mode) # 統計 success = sum(1 for r in results.values() if r["status"] == "success") total_records = sum(r.get("count", 0) for r in results.values()) logger.info("-" * 50) logger.info(f"收集完成: {success}/{len(results)} 縣市成功, 共 {total_records} 筆資料") return results def get_next_scheduled_time(): """計算下一個整點或半點時間（:00 或 :30）""" from datetime import timedelta now = datetime.now() minute = now.minute if minute < 30: # 下一個是 :30 next_time = now.replace(minute=30, second=0, microsecond=0) else: # 下一個是下一小時的 :00（使用 timedelta 處理跨日） next_time = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) return next_time def wait_until_scheduled_time(): """等待到下一個排程時間""" next_time = get_next_scheduled_time() now = datetime.now() wait_seconds = (next_time - now).total_seconds() if wait_seconds > 0: logger.info(f"等待至 {next_time.strftime('%H:%M:%S')}（{wait_seconds:.0f} 秒後）") time.sleep(wait_seconds) return next_time def run_daemon(): """持續執行模式（在整點和半點收集）""" logger.info("啟動持續收集模式...") logger.info("排程時間: 每小時的 :00 和 :30") # 初始化狀態回報器 reporter = None if HAS_STATUS_REPORTER: reporter = StatusReporter("parking-collector") logger.info("狀態回報模組已啟用") conn = init_db() # 啟動時先收集靜態資料 collect_all_static(conn) # 立即執行一次（帶狀態回報） _collect_with_report(conn, reporter) try: while True: # 等待到下一個整點或半點 next_time = wait_until_scheduled_time() logger.info(f"=== 排程收集 {next_time.strftime('%H:%M')} ===") _collect_with_report(conn, reporter) except KeyboardInterrupt: logger.info("收到中斷信號，停止收集") finally: conn.close() def _collect_with_report(conn: sqlite3.Connection, reporter=None): """收集資料並回報狀態""" start_time = time.time() results = collect_all(conn) duration = time.time() - start_time # 統計結果 success_count = sum(1 for r in results.values() if r["status"] == "success") total_records = sum(r.get("count", 0) for r in results.values()) failed_cities = [k for k, r in results.items() if r["status"] != "success"] # 回報狀態 if reporter: reporter.report( success=len(failed_cities) == 0, records=total_records, duration=duration, metrics={ "cities_success": success_count, "cities_total": len(results), }, errors=[f"{c}: {results[c].get('error', 'unknown')}" for c in failed_cities] if failed_cities else None, ) def run_once(test_mode: bool = False): """執行一次收集""" conn = init_db() if not test_mode else None # 初始化狀態回報器（非測試模式） reporter = None if HAS_STATUS_REPORTER and not test_mode: reporter = StatusReporter("parking-collector") try: start_time = time.time() results = collect_all(conn, test_mode) duration = time.time() - start_time if test_mode: print("\n=== 測試結果 ===") for city_key, result in results.items(): city_name = CITY_APIS[city_key]["name"] status = "OK" if result["status"] == "success" else "FAIL" count = result.get("count", 0) error = result.get("error", "") print(f" {city_name}: {status} ({count} 筆) {error}") else: # 回報狀態 if reporter: success_count = sum(1 for r in results.values() if r["status"] == "success") total_records = sum(r.get("count", 0) for r in results.values()) failed_cities = [k for k, r in results.items() if r["status"] != "success"] reporter.report( success=len(failed_cities) == 0, records=total_records, duration=duration, metrics={ "cities_success": success_count, "cities_total": len(results), }, errors=[f"{c}: {results[c].get('error', 'unknown')}" for c in failed_cities] if failed_cities else None, ) finally: if conn: conn.close() # ============================================================ # 主程式 # ============================================================ def run_static_only(): """只收集靜態資料""" conn = init_db() collect_all_static(conn) conn.close() logger.info("靜態資料收集完成") def main(): global BASE_INTERVAL parser = argparse.ArgumentParser(description="台灣停車場即時資料收集器") parser.add_argument("--daemon", "-d", action="store_true", help="持續執行模式") parser.add_argument("--test", "-t", action="store_true", help="測試模式（不存 DB）") parser.add_argument("--static", "-s", action="store_true", help="只收集靜態資料") parser.add_argument("--interval", "-i", type=int, default=1800, help="收集間隔（秒），預設 30 分鐘") args = parser.parse_args() BASE_INTERVAL = args.interval if args.static: run_static_only() elif args.daemon: run_daemon() else: run_once(test_mode=args.test) if __name__ == "__main__": main()