Skip to content

Instantly share code, notes, and snippets.

@yujqiao

yujqiao/ica_watch.py

Last active September 22, 2025 14:55
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save yujqiao/38bae9dc0b2688f7c1de11f3ceaf52c6 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save yujqiao/38bae9dc0b2688f7c1de11f3ceaf52c6 to your computer and use it in GitHub Desktop.
Download ZIP
ICA Application Status Watcher
Raw
ica_watch.py
'''
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
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
'''
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment