|
|
@@ -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() |
yujqiao revised this gist