-
-
Save Hackl0us/0394e9a941a4c9b200c9b83d3d686a8d to your computer and use it in GitHub Desktop.
ICA Application Status Watcher
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| ''' | |
| 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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment