import argparse import datetime import io import json import os import re import sys import subprocess import zipfile # ========================================== # šŸ› ļø RUNTIME DEPENDENCY INSTALLATION # ========================================== try: import geopandas as gpd except ImportError: print("šŸ“¦ Installing geopandas dynamically...", flush=True) subprocess.check_call([ sys.executable, "-m", "pip", "install", "geopandas==0.10.2", "shapely<2.0.0", "fiona<1.9.0" ]) import geopandas as gpd import pandas as pd import requests DEFAULT_OUTPUT_DIR = "raw_data" # ========================================== # šŸ”‘ ENVIRONMENT API KEYS # ========================================== # Keys are read directly from CI/CD Environment Variables DATA_GOV_API_KEY = os.getenv("DATA_GOV_API_KEY", "") DATAMALL_API_KEY = os.getenv("DATAMALL_API_KEY", "") ONEMAP_TOKEN = os.getenv("ONEMAP_TOKEN", "") ONEMAP_EMAIL = os.getenv("ONEMAP_EMAIL", "") ONEMAP_PASSWORD = os.getenv("ONEMAP_PASSWORD", "") # Dynamic Headers DATA_GOV_HEADERS = {} DATAMALL_HEADERS = {} ONEMAP_HEADERS = {} # ========================================== # šŸ“¦ DATASETS CONFIGURATION # ========================================== DATA_GOV_DATASETS = [ { "id": "d_8594ae9ff96d0c708bc2af633048edfb", "name": "subzone_boundary_mp19", "file": "subzone_boundary_mp19.geojson", }, { "id": "d_d95ae740c0f8961a0b10435836660ce0", "name": "pop_age_sex_census2020", "file": "pop_age_sex_census2020.csv", }, { "id": "d_7f243956483d5901f237e6f87b096636", "name": "pop_dwelling_census2020", "file": "pop_dwelling_census2020.csv", }, { "id": "d_9a6bdc9d93bd041eb0cfbb6a8cb3248f", "name": "rail_stations_mp19", "file": "rail_stations_mp19.geojson", }, { "id": "d_90d86daa5bfaa371668b84fa5f01424f", "name": "land_use_mp19", "file": "land_use_mp19.geojson", }, { "id": "d_688b934f82c1059ed0a6993d2a829089", "name": "general_info_schools", "file": "general_info_schools.csv", }, ] ONEMAP_THEMES = [ "ssot_hawkercentres", "museum", "votg_phpc", "libraries", "communityclubs", "monuments", "historicsites", "hotels", "tourism", "nationalparks", ] # ========================================== # šŸ” INITIALIZE HEADERS & AUTH # ========================================== def initialize_headers(): global DATA_GOV_HEADERS, DATAMALL_HEADERS, ONEMAP_HEADERS token = ONEMAP_TOKEN # Generate fresh JWT Token if credentials exist if ONEMAP_EMAIL and ONEMAP_PASSWORD: print("šŸ” Authenticating with OneMap API...") try: auth_res = requests.post( "https://www.onemap.gov.sg/api/auth/post/getToken", json={"email": ONEMAP_EMAIL, "password": ONEMAP_PASSWORD}, headers={"Content-Type": "application/json"}, timeout=15 ) if auth_res.status_code == 200: token = auth_res.json().get("access_token") print("āœ… Fresh OneMap JWT Token generated successfully!") except Exception as err: print(f"āš ļø OneMap Auth failed ({err}). Falling back to ONEMAP_TOKEN variable.") clean_datamall_key = re.sub(r"[^\x00-\x7F]+", "", str(DATAMALL_API_KEY)).strip() clean_onemap_token = str(token).replace("Bearer", "").strip() DATA_GOV_HEADERS = {"x-api-key": DATA_GOV_API_KEY, "Content-Type": "application/json"} DATAMALL_HEADERS = {"AccountKey": clean_datamall_key, "accept": "application/json"} ONEMAP_HEADERS = {"Authorization": f"Bearer {clean_onemap_token}", "accept": "application/json"} # ========================================== # šŸ› ļø HELPER FUNCTIONS FOR API FETCHING # ========================================== def fetch_poll_download_dataset(dataset_id, filepath): poll_url = f"https://api-open.data.gov.sg/v1/public/api/datasets/{dataset_id}/poll-download" try: response = requests.get(poll_url, headers=DATA_GOV_HEADERS, timeout=15) json_data = response.json() if json_data.get("code") != 0: print(f"āŒ Error polling dataset {dataset_id}: {json_data.get('errMsg')}") return False download_url = json_data["data"]["url"] data_response = requests.get(download_url, timeout=30) with open(filepath, "w", encoding="utf-8") as f: f.write(data_response.text) print(f"āœ… Saved to {filepath}") return True except Exception as e: print(f"āŒ Failed to process dataset {dataset_id}: {e}") return False def fetch_lta_od_csv(endpoint_type, filepath): url = f"https://datamall2.mytransport.sg/ltaodataservice/{endpoint_type}" try: response = requests.get(url, headers=DATAMALL_HEADERS, timeout=15) if response.status_code != 200: return None download_url = response.json()["value"][0]["Link"] zip_res = requests.get(download_url, timeout=60) if zip_res.status_code == 200: with zipfile.ZipFile(io.BytesIO(zip_res.content)) as z: with z.open(z.namelist()[0]) as f: df = pd.read_csv(f) df.to_csv(filepath, index=False) print(f"āœ… Saved {len(df):,} records to: {filepath}") return len(df) return None except Exception as e: print(f"āŒ Error fetching {endpoint_type}: {e}") return None def fetch_lta_bus_stops(filepath): url = "https://datamall2.mytransport.sg/ltaodataservice/BusStops" all_stops, skip = [], 0 try: while True: res = requests.get(f"{url}?$skip={skip}", headers=DATAMALL_HEADERS, timeout=15) if res.status_code != 200: break data = res.json().get("value", []) if not data: break all_stops.extend(data) skip += 500 if all_stops: df = pd.DataFrame(all_stops) df.to_csv(filepath, index=False) print(f"āœ… Saved {len(df):,} bus stop locations to: {filepath}") return len(df) return None except Exception as e: print(f"āŒ Error fetching bus stops: {e}") return None def fetch_onemap_theme(theme_name, filepath): url = f"https://www.onemap.gov.sg/api/public/themesvc/retrieveTheme?queryName={theme_name}" try: res = requests.get(url, headers=ONEMAP_HEADERS, timeout=15) if res.status_code == 200: data = res.json().get("SrchResults", []) pois = data[1:] if len(data) > 1 else [] if pois: df = pd.DataFrame(pois) df.to_csv(filepath, index=False) print(f"āœ… Saved {len(pois):,} POIs to: {filepath}") return len(pois) return None except Exception as e: print(f"āŒ Exception fetching theme '{theme_name}': {e}") return None # ========================================== # šŸš€ MAIN PIPELINE EXECUTION # ========================================== if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--output-dir", type=str, default=DEFAULT_OUTPUT_DIR, help="Local directory to store fetched raw data") args, _ = parser.parse_known_args() print(f"šŸš€ Starting Local Raw Data Fetch -> Output Directory: '{args.output_dir}'") initialize_headers() raw_data_dir = args.output_dir os.makedirs(raw_data_dir, exist_ok=True) # 1. Data.gov.sg Batch Fetch print("\nšŸ“¦ Fetching Data.gov.sg Datasets...") for item in DATA_GOV_DATASETS: out_file = os.path.join(raw_data_dir, item["file"]) fetch_poll_download_dataset(item["id"], out_file) # 2. LTA DataMall Endpoints print("\nšŸ“¦ Fetching LTA DataMall Endpoints...") fetch_lta_od_csv("PV/ODBus", os.path.join(raw_data_dir, "lta_od_bus.csv")) fetch_lta_od_csv("PV/ODTrain", os.path.join(raw_data_dir, "lta_od_train.csv")) fetch_lta_bus_stops(os.path.join(raw_data_dir, "lta_bus_stops.csv")) # 3. OneMap POI Themes print(f"\nšŸ“¦ Fetching {len(ONEMAP_THEMES)} OneMap POI Themes...") for theme_api_name in ONEMAP_THEMES: out_file = os.path.join(raw_data_dir, f"onemap_{theme_api_name}.csv") fetch_onemap_theme(theme_api_name, out_file) print(f"\nšŸŽ‰ Local Data Ingestion Complete! All raw data stored in '{raw_data_dir}'")