import urllib.request import json import sqlite3 import os import sys from datetime import datetime, timezone, timedelta # Configuration URL = "https://eqms.doe.gov.my/api3/publicmapproxy/PUBLIC_DISPLAY/CAQM_MCAQM_Current_Reading/MapServer/0/query?f=json&outFields=*&returnGeometry=false&spatialRel=esriSpatialRelIntersects&where=1%3D1" REFERER = "https://eqms.doe.gov.my/APIMS/main" 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" DB_NAME = "/var/www/mypapit.net/public_html/pub/pollution/apims_data.db" def setup_database(): """Initializes the SQLite database and creates the readings table.""" conn = sqlite3.connect(DB_NAME) cursor = conn.cursor() # Create table with STATION_ID and DATETIME as composite primary key # to store historical records if the script is run multiple times. cursor.execute(""" CREATE TABLE IF NOT EXISTS station_readings ( STATION_ID TEXT, DATETIME INTEGER, DATETIMEMY TEXT, API INTEGER, API_PM10 INTEGER, PARAM_SELECTED TEXT, PARAM_SYMBOL TEXT, PARAM_SYMBOL_PM10 TEXT, CLASS TEXT, STATION_LOCATION TEXT, LONGITUDE REAL, LATITUDE REAL, PLACE TEXT, LOT_INFO TEXT, STATION_CATEGORY TEXT, STATE_NAME TEXT, REGION_NAME TEXT, PRIMARY KEY (STATION_ID, DATETIME) ) """) conn.commit() return conn def fetch_data(): """Downloads JSON from the APIMS query URL with appropriate headers.""" print(f"Fetching data from URL: {URL}") req = urllib.request.Request( URL, headers={ "User-Agent": USER_AGENT, "Referer": REFERER, "Accept": "application/json" } ) try: with urllib.request.urlopen(req, timeout=30) as response: if response.status != 200: print(f"Error: Server responded with status code {response.status}") return None data = response.read().decode('utf-8') return json.loads(data) except Exception as e: print(f"Error during network request: {e}") return None def epoch_to_iso8601_my(epoch_ms): """Converts a millisecond epoch timestamp to ISO8601 string in Malaysia (+08:00) timezone without applying timezone offset conversion. """ if epoch_ms is None: return None # Convert ms to seconds seconds = epoch_ms / 1000.0 # Interpret the UTC time directly as local time by changing the tzinfo # without shifting the hour value. dt_utc = datetime.fromtimestamp(seconds, tz=timezone.utc) dt_local = dt_utc.replace(tzinfo=timezone(timedelta(hours=8))) return dt_local.isoformat() def save_to_database(conn, data): """Parses features and saves attributes to the database.""" if not data or "features" not in data: print("No valid features found in the response.") return 0 features = data["features"] print(f"Found {len(features)} features in response.") cursor = conn.cursor() inserted_count = 0 updated_count = 0 sql = """ INSERT INTO station_readings ( STATION_ID, DATETIME, DATETIMEMY, API, API_PM10, PARAM_SELECTED, PARAM_SYMBOL, PARAM_SYMBOL_PM10, CLASS, STATION_LOCATION, LONGITUDE, LATITUDE, PLACE, LOT_INFO, STATION_CATEGORY, STATE_NAME, REGION_NAME ) VALUES ( :STATION_ID, :DATETIME, :DATETIMEMY, :API, :API_PM10, :PARAM_SELECTED, :PARAM_SYMBOL, :PARAM_SYMBOL_PM10, :CLASS, :STATION_LOCATION, :LONGITUDE, :LATITUDE, :PLACE, :LOT_INFO, :STATION_CATEGORY, :STATE_NAME, :REGION_NAME ) ON CONFLICT(STATION_ID, DATETIME) DO UPDATE SET DATETIMEMY=excluded.DATETIMEMY, API=excluded.API, API_PM10=excluded.API_PM10, PARAM_SELECTED=excluded.PARAM_SELECTED, PARAM_SYMBOL=excluded.PARAM_SYMBOL, PARAM_SYMBOL_PM10=excluded.PARAM_SYMBOL_PM10, CLASS=excluded.CLASS, STATION_LOCATION=excluded.STATION_LOCATION, LONGITUDE=excluded.LONGITUDE, LATITUDE=excluded.LATITUDE, PLACE=excluded.PLACE, LOT_INFO=excluded.LOT_INFO, STATION_CATEGORY=excluded.STATION_CATEGORY, STATE_NAME=excluded.STATE_NAME, REGION_NAME=excluded.REGION_NAME """ for feature in features: attributes = feature.get("attributes") if not attributes: continue # Ensure we have the minimum required identification fields if not attributes.get("STATION_ID") or attributes.get("DATETIME") is None: continue # Normalize fields to match our query parameter names # SQLite's named placeholder binds dictionary keys directly param_dict = { "STATION_ID": attributes.get("STATION_ID"), "DATETIME": attributes.get("DATETIME"), "DATETIMEMY": epoch_to_iso8601_my(attributes.get("DATETIME")), "API": attributes.get("API"), "API_PM10": attributes.get("API_PM10"), "PARAM_SELECTED": attributes.get("PARAM_SELECTED"), "PARAM_SYMBOL": attributes.get("PARAM_SYMBOL"), "PARAM_SYMBOL_PM10": attributes.get("PARAM_SYMBOL_PM10"), "CLASS": attributes.get("CLASS"), "STATION_LOCATION": attributes.get("STATION_LOCATION"), "LONGITUDE": attributes.get("LONGITUDE"), "LATITUDE": attributes.get("LATITUDE"), "PLACE": attributes.get("PLACE"), "LOT_INFO": attributes.get("LOT_INFO"), "STATION_CATEGORY": attributes.get("STATION_CATEGORY"), "STATE_NAME": attributes.get("STATE_NAME"), "REGION_NAME": attributes.get("REGION_NAME") } try: # Check if record already exists to track insert vs update stats cursor.execute( "SELECT 1 FROM station_readings WHERE STATION_ID = ? AND DATETIME = ?", (param_dict["STATION_ID"], param_dict["DATETIME"]) ) exists = cursor.fetchone() cursor.execute(sql, param_dict) if exists: updated_count += 1 else: inserted_count += 1 except sqlite3.Error as e: print(f"Database error inserting station {param_dict['STATION_ID']}: {e}") conn.commit() print(f"Database sync complete: {inserted_count} inserted, {updated_count} updated.") return inserted_count + updated_count def display_summary(conn): """Displays a summary of the data stored in the database.""" cursor = conn.cursor() try: # Total counts cursor.execute("SELECT COUNT(*) FROM station_readings") total_rows = cursor.fetchone()[0] # Unique stations cursor.execute("SELECT COUNT(DISTINCT STATION_ID) FROM station_readings") total_stations = cursor.fetchone()[0] print("\n--- SQLite Database Summary ---") print(f"Database Location: {os.path.abspath(DB_NAME)}") print(f"Total stored records: {total_rows}") print(f"Total unique stations: {total_stations}") if total_rows > 0: print("\nLatest 5 readings:") cursor.execute(""" SELECT STATION_ID, DATETIMEMY, API, CLASS, STATE_NAME FROM station_readings ORDER BY DATETIME DESC, STATION_ID LIMIT 5 """) print(f"{'Station ID':<12} | {'Time (Local/MY)':<25} | {'API':<5} | {'Class':<10} | {'State':<15}") print("-" * 75) for row in cursor.fetchall(): print(f"{row[0]:<12} | {row[1]:<25} | {row[2]:<5} | {row[3]:<10} | {row[4]:<15}") except sqlite3.Error as e: print(f"Failed to query database statistics: {e}") def main(): db_conn = setup_database() try: data = fetch_data() if data: save_to_database(db_conn, data) display_summary(db_conn) else: print("Failed to download data.") finally: db_conn.close() if __name__ == "__main__": main()