yujqiao · September 22, 2025 14:55 · Sep 22, 2025 · Sep 22, 2025 · Sep 22, 2025
diff --git a/ica_watch.py b/ica_watch.py
@@ -1,5 +1,7 @@
 '''
-This is a helper script to watch application status. Use at your own risk
+This is a helper script to watch application status and send notification on any state change.
+
+Use at your own risk.
 
 Set env var APPLICATION_NO to your case number e.g. ISC25000000000
 Set env var CHECK_INTERVAL to refresh interval in seconds

diff --git a/ica_watch.py b/ica_watch.py
@@ -3,6 +3,7 @@
 
 Set env var APPLICATION_NO to your case number e.g. ISC25000000000
 Set env var CHECK_INTERVAL to refresh interval in seconds
+Set env var INITIAL_COOKIES to the cookie after log in to ICA website, in http header format
 Implement your own notification logic in send_notification
 '''
 

diff --git a/ica_watch.py b/ica_watch.py
@@ -0,0 +1,201 @@
+'''
+This is a helper script to watch application status. Use at your own risk
+
+Set env var APPLICATION_NO to your case number e.g. ISC25000000000
+Set env var CHECK_INTERVAL to refresh interval in seconds
+Implement your own notification logic in send_notification
+'''
+
+import requests
+import time
+import os
+from datetime import datetime
+import json
+import shutil
+import logging
+from http.cookies import SimpleCookie
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+# Create a persistent session with cookie jar
+session = requests.Session()
+
+APPLICATION_NO = os.getenv("APPLICATION_NO")
+if not APPLICATION_NO:
+    raise ValueError("No APPLICATION_NO found in environment variables")
+
+REFRESH_URL = (
+    "https://eservices.ica.gov.sg/ipses/api/AuthenticationAuthorization/token/refresh"
+)
+APPLICATION_URL = "https://eservices.ica.gov.sg/ipses/api/EligibilityAndApplication/RetrieveByGrpBundle"
+
+HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
+    "Accept": "application/json",
+    "Accept-Language": "en-US,en;q=0.5",
+    "Accept-Encoding": "deflate",
+    "Content-Type": "application/json",
+    "Origin": "https://eservices.ica.gov.sg",
+    "Connection": "keep-alive",
+    "Referer": f"https://eservices.ica.gov.sg/ipses/app/{APPLICATION_NO}/case-summary",
+    "Sec-Fetch-Dest": "empty",
+    "Sec-Fetch-Mode": "cors",
+    "Sec-Fetch-Site": "same-origin",
+    "Pragma": "no-cache",
+    "Cache-Control": "no-cache",
+}
+
+REFRESH_HEADERS = {
+    "Accept": "*/*",
+    "Accept-Language": "en-US,en;q=0.5",
+    "Accept-Encoding": "gzip, deflate, br, zstd",
+    "Connection": "keep-alive",
+    "Host": "eservices.ica.gov.sg",
+    "Priority": "u=4",
+    "Referer": f"https://eservices.ica.gov.sg/ipses/app/{APPLICATION_NO}/case-summary",
+    "Sec-Fetch-Dest": "empty",
+    "Sec-Fetch-Mode": "cors",
+    "Sec-Fetch-Site": "same-origin",
+    "Pragma": "no-cache",
+    "Cache-Control": "no-cache",
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0",
+}
+
+
+def load_initial_cookies():
+    """Load initial cookies from environment variable INITIAL_COOKIES"""
+    cookie_string = os.getenv("INITIAL_COOKIES", "")
+    if cookie_string:
+        logger.info("Loading initial cookies from environment variable")
+        # Parse cookie string and add to session
+        cookie = SimpleCookie()
+        cookie.load(cookie_string)
+        for key, morsel in cookie.items():
+            session.cookies.set(key, morsel.value)
+        logger.info(f"Loaded {len(cookie)} initial cookies")
+    else:
+        raise ValueError(
+            "No initial cookies found in INITIAL_COOKIES environment variable"
+        )
+
+
+def send_notification(message):
+    # TODO: your notification impl here
+    return
+
+def refresh_token():
+    try:
+        response = session.get(REFRESH_URL, headers=REFRESH_HEADERS)
+        response.raise_for_status()
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error refreshing token: {e}")
+
+def fetch_ica_data():
+    payload = {"eServiceGroupId": APPLICATION_NO}
+
+    try:
+        response = session.post(APPLICATION_URL, headers=HEADERS, json=payload)
+        response.raise_for_status()
+        logger.info(response.text)
+        return response.json()
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error fetching data: {e}")
+        return None
+
+
+def load_previous_data():
+    try:
+        with open("./data/previous_response.json", "r", encoding="utf-8") as f:
+            data = json.load(f)
+            return data
+    except FileNotFoundError:
+        return None
+
+
+def save_current_data(data):
+    # Move current to previous
+    if os.path.exists("./data/current_response.json"):
+        shutil.copyfile("./data/current_response.json", "./data/previous_response.json")
+
+    with open("./data/current_response.json", "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+
+
+def monitor_ica():
+    check_interval = int(os.getenv("CHECK_INTERVAL", "300"))
+
+    # Load initial cookies from environment variable
+    load_initial_cookies()
+
+    logger.info(f"Starting ICA monitor with {check_interval}s interval...")
+    send_notification("🤖 ICA Watch Started")
+
+    while True:
+        try:
+            logger.info("Checking for changes...")
+
+            current_data = fetch_ica_data()
+            if current_data is None:
+                send_notification("Failed to fetch ICA data. Exiting")
+                logger.error("Failed to fetch ICA data. Exiting")
+                break
+
+            previous_data = load_previous_data()
+
+            if previous_data is None:
+                save_current_data(current_data)
+                message = f"""# 🤖 ICA Watch Started
+
+Initial data captured at {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
+
+Monitoring for changes..."""
+                send_notification(message)
+                logger.info(
+                    "Initial data saved to current_response.json and previous_response.json"
+                )
+            else:
+                from deepdiff import DeepDiff
+
+                diff = DeepDiff(previous_data, current_data, ignore_order=True)
+                if len(diff) != 0:
+                    logger.warning("CHANGE DETECTED!")
+
+                    message = f"""# 🚨 ICA Data Changed!
+
+**Time:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
+
+## Diff:
+```
+{diff.pretty()}
+```"""
+
+                    send_notification(message)
+                else:
+                    logger.info("No changes detected")
+
+            save_current_data(current_data)
+
+            logger.info("refreshing token...")
+            refresh_token()
+
+            logger.info(f"Sleeping for {check_interval} seconds...")
+            time.sleep(check_interval)
+
+        except KeyboardInterrupt:
+            logger.info("Monitoring stopped by user")
+            break
+        except Exception as e:
+            logger.error(f"Unexpected error: {e}")
+            time.sleep(check_interval)
+
+
+def main():
+    monitor_ica()
+
+
+if __name__ == "__main__":
+    main()
No results found