import requests from bs4 import BeautifulSoup from datetime import datetime import os import time import random # CONFIG RETRY_LIMIT = 3 MIN_WAIT = 300 # 5 minutes MAX_WAIT = 600 # 10 minutes CSV_MIN_BYTES = 1000 # Anything smaller is probably header-only url = "https://members.payroll.ca/CPA21-EN/PubliciQA.aspx" BASE_DIR = os.path.dirname(os.path.abspath(__file__)) MAIN_CSV = os.path.join(BASE_DIR, "ResultsGrid_ExportData.csv") ERROR_CSV = os.path.join(BASE_DIR, "ResultsGrid_ExportData_error.csv") def is_error_html(csv_bytes): """Detect if the server returned an HTML error page instead of CSV.""" text = csv_bytes[:5000].decode("utf-8", errors="ignore").lower() if "error" in text: return True if "gwebsitekey" in text: return True if "imis" in text and "script" in text: return True return False def download_csv(): session = requests.Session() r = session.get(url, verify=False) soup = BeautifulSoup(r.text, "html.parser") # Extract hidden fields data = {} for hidden in soup.find_all("input", type="hidden"): data[hidden.get("name")] = hidden.get("value", "") # Simulate CSV export click data["__EVENTTARGET"] = ( "ctl00$TemplateBody$WebPartManager1$gwpciNewQueryMenuCommon$ciNewQueryMenuCommon$ResultsGrid$btnExportCSV" ) data["__EVENTARGUMENT"] = "" csv_response = session.post(url, data=data, verify=False) return csv_response.content def csv_is_empty(csv_bytes): """Detect the 'bad CSV' that only contains header row.""" if len(csv_bytes) < CSV_MIN_BYTES: return True content = csv_bytes.decode("utf-8-sig", errors="ignore") lines = content.strip().splitlines() return len(lines) <= 1 def save_file(path, csv_bytes): with open(path, "wb") as f: f.write(csv_bytes) # -------------------- # MAIN RETRY LOOP # -------------------- attempt = 1 while attempt <= RETRY_LIMIT: print(f"\n=== Attempt {attempt}/{RETRY_LIMIT} at {datetime.now()} ===") csv_bytes = download_csv() # ----- Detect HTML error page ----- if is_error_html(csv_bytes): print("❌ HTML Error detected. Keeping previous CSV unchanged.") # Save error file for debugging save_file(ERROR_CSV, csv_bytes) print(f"⚠️ Error version saved as {ERROR_CSV}") if attempt == RETRY_LIMIT: print("❌ Max retries reached. No changes made to main CSV.") break wait_time = random.randint(MIN_WAIT, MAX_WAIT) print(f"⏳ Waiting {wait_time} seconds before retrying...\n") time.sleep(wait_time) attempt += 1 continue # ----- Detect empty CSV (header only) ----- if csv_is_empty(csv_bytes): print("⚠️ Empty CSV detected. Keeping previous CSV unchanged.") save_file(ERROR_CSV, csv_bytes) print(f"⚠️ Empty version saved as {ERROR_CSV}") if attempt == RETRY_LIMIT: print("❌ Max retries reached. No changes made to main CSV.") break wait_time = random.randint(MIN_WAIT, MAX_WAIT) print(f"⏳ Waiting {wait_time} seconds before retrying...\n") time.sleep(wait_time) attempt += 1 continue # If CSV is valid → save it as the new main file print("✅ Valid CSV downloaded. Updating main CSV file...") save_file(MAIN_CSV, csv_bytes) print(f"📁 Updated: {MAIN_CSV}") break