Reverse Engineering UCSD's Parking Ticket System (and 100+ other universities)

def citation_exists(citation_id):
    base_url = "https://ucsd-transportation.t2hosted.com/Account/Citations/Search"

    headers = {
        "accept": "*/*",
        "accept-language": "en-US,en;q=0.9",
        "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
        "sec-ch-ua": '"Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"macOS"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "x-requested-with": "XMLHttpRequest",
        "cookie": "ASP.NET_SessionId=aaaaaaaaaaaaaaaaaaaaaaaa; __RequestVerificationToken=erzxiBeAmnBMHSa8dDdF7s5cw5E7xKdz_Aoskm4fEq2k-kou-t4G1drplCPO4wT85Nr_i58jEckw2635H8M-J1TiJz4vb8Ai14FZywLzfuQ1; apiToken={%22secure%22:true%2C%22sameSite%22:%22none%22}",
        "Referer": "https://ucsd-transportation.t2hosted.com/Account/Portal",
        "Referrer-Policy": "strict-origin-when-cross-origin",
    }

    data = {
        "__RequestVerificationToken": "7bOTSiled4jlS4xdGJLj8gZqm9Sn4wYaxhTXEC4B_Peh6mqPrXBAR0eEmh90IXJ077Nqxq4NWY7tk-xQmH7t_AhlzNiwdo1hgLy0PifrHAw1",
        "CitationNumber": str(citation_id),
        "StateId": "5",
        "PlateNumber": "",
        "X-Requested-With": "XMLHttpRequest",
    }

    response = requests.post(base_url, headers=headers, data=data)

    # Check if '/Account/Citations/Results' is in the response. If so, the ticket exists.
    return "/Account/Citations/Results" in response.text

def find_latest_citation(start_id = 15207400000, max_increment = 10000):
    low = start_id
    high = start_id + max_increment

    while low <= high:
        mid = (low + high) // 2

        # Check if the certain citation number exists
        if citation_exists(mid):
            print(mid, "does exist. Moving forward...")

            # If exists, try a larger number
            low = mid + 1
        else:
            print(mid, "does not exist. Moving backward...")

            # If does not exist, try a smaller number
            high = mid - 1

    # When the loop finishes, `high` should be the latest valid citation number
    return high

def check_for_error_message(driver):
    # This function checks if there is an error message that appeared in the HTML after looking up a citation.
    try:
        error_message_element = WebDriverWait(driver, 3).until(
            EC.presence_of_element_located((By.CLASS_NAME, "message"))
        )
        return error_message_element.text
    except TimeoutException:
        return None

def check_for_redirect(driver):
    # This function checks if, after searching for a citation, a redirect occurs. If there is a redirect, then we know for a fact that the website returned valid ticket data.
    try:
        WebDriverWait(driver, 3).until(
            EC.url_contains("/Account/Citations/Results")
        )
        return True
    except TimeoutException:
        return False

def extract_citation_details(driver):
    # This function extracts the ticket data from the table in the HTML
    try:
        # Wait until the table containing the citation details appears
        WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.ID, "citations-list-table"))
        )

        # Extract the citation details based on the XPath selectors
        citation_number = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[1]").text
        status = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[2]").text
        balance = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[3]").text
        issue_date = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[4]").text
        license_plate = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[5]").text
        location = driver.find_element(By.XPATH, "//table[@id='citations-list-table']//tr/td[6]").text

        # Convert the extracted data into a dictionary (valid response)
        citation_data = {
            "citation_number": citation_number,
            "status": status,
            "balance": balance,
            "issue_date": issue_date,
            "license_plate": license_plate,
            "location": location
        }
        return citation_data

    except TimeoutException:
        return None
    except NoSuchElementException:
        return None

def get_citation_status(citation_id):
    # This function performs a lookup for a certain citation id. It then parses the parking ticket data or error message

    if not citation_exists(citation_id):
        return "Citation does not exist"

    driver = get_webdriver() # We are using chromedriver by Selenium
    try:
        # Navigate to the URL
        driver.get("https://ucsd-transportation.t2hosted.com/Account/Portal")

        # Wait for the citation input field to appear
        WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.ID, "citationSearchBox"))
        )

        # Enter the citation ID
        citation_input = driver.find_element(By.ID, "citationSearchBox")
        citation_input.clear()
        citation_input.send_keys(citation_id)

        # Click the search button
        search_button = WebDriverWait(driver, 10).until(
            EC.element_to_be_clickable((By.XPATH, "//button[@type='submit' and contains(text(), 'Search Citations')]"))
        )
        search_button.click()

        # Run checks for redirection or errors
        with concurrent.futures.ThreadPoolExecutor() as executor:
            error_check = executor.submit(check_for_error_message, driver)
            redirect_check = executor.submit(check_for_redirect, driver)

            # Wait for either result to return
            done, _ = concurrent.futures.wait(
                [error_check, redirect_check],
                return_when=concurrent.futures.FIRST_COMPLETED
            )

            for future in done:
                result = future.result()
                if result == True:
                    # If redirection is detected, extract citation details
                    return extract_citation_details(driver)
                elif isinstance(result, str):
                    return f"Error message found: {result}"

        return "No result found."

    except TimeoutException:
        return "Timeout occurred while fetching citation data"
    finally:
        driver.quit()

def check_and_insert_citation(citation_id):
    # This function calls get_citation_status and inserts the appropriate data into the database

    citation_status = get_citation_status(citation_id)

    if isinstance(citation_status, dict):
        # There is valid ticket data. Insert it into the database
        insert_ticket(
            citation_id=citation_id,
            status=citation_status.get('status', None),
            issue_date=citation_status.get('issue_date', None),
            license_plate=citation_status.get('license_plate', None),
            balance=citation_status.get('balance', None),
            location=citation_status.get('location', None)
        )
        print(f"Citation ID {citation_id} data inserted/updated successfully.")
    elif "Error message found" in citation_status:
    # This likely means that the ticket has either been appealed or paid off. In either case, we cannot access the parking ticket data, so we just log the error message into our database.
        response = insert_error_ticket(citation_id, citation_status, should_try_again=True)
        print(f"Error: {citation_status} for citation id {citation_id}. This will not be retried.")
    elif citation_status == "Citation does not exist":
        print(f"Citation ID {citation_id} does not exist.")
    else:
        print(f"Unexpected response for citation ID {citation_id}: {citation_status}")

def process_all_citations(start_id = 15207422649):
    current_id = start_id
    while current_id > 15207400001:
        handle_citation(current_id)
        current_id -= 1  # Decrement citation ID to check the next one

def find_latest_citations_for_all_devices(region_num=152, max_increment=99999):
    results = {}
    
    # Loop through device numbers from 1 to 93
    for device_num in range(1, 94):
        ticket_num = 1
        citation_num = region_num * 100000000 + device_num * 100000 + ticket_num
        
        print(f"Searching for device_num {device_num} starting from {citation_num}")
        latest_citation = find_latest_citation(citation_num, max_increment)
        results[f"device_{device_num}"] = latest_citation
        
        print(f"Device {device_num}: Latest citation found is {latest_citation}")

    # Write the results to a JSON file
    with open('latest_citations.json', 'w') as json_file:
        json.dump(results, json_file, indent=4)
    
    return results

{
    "device_1": 15200119982,
    "device_2": 15200203670,
    "device_3": 15200316008,
    "device_4": 15200400001,
    "device_5": 15200500000,
    "device_6": 15200647499,
    "device_7": 15200700000,
    "device_8": 15200818437,
    "device_9": 15200906846,
    "device_10": 15201029610,
    // more here...
    "device_89": 15208900011,
    "device_90": 15209001322,
    "device_91": 15209100000,
    "device_92": 15209200000,
    "device_93": 15209300059
}

def load_latest_citations():
    """Load the latest citations from the JSON file."""
    with open("latest_citations.json", "r") as json_file:
        return json.load(json_file)


def save_latest_citations(latest_citations):
    """Save the updated latest citations to the JSON file."""
    with open("latest_citations.json", "w") as json_file:
        json.dump(latest_citations, json_file, indent=4)


def scrape_new_citations():
    """Continuously check for new citations on each device and process them."""
    latest_citations = (
        load_latest_citations()
    )  # Load the latest citations from the file

    for device, citation_id in latest_citations.items():
        next_citation_id = (
            citation_id + 1
        )  # Start with the next citation number

        while True:
            # Handle the citation and check if it was successfully inserted
            citation_handled = check_and_insert_citation(
                next_citation_id
            )

            if citation_handled:
                print("New citation found and inserted!")

                # Update the latest citation ID for this device
                latest_citations[device] = next_citation_id

                # Save the updated citation data back to the JSON file
                save_latest_citations(latest_citations)

                # Increment to try the next citation immediately
                next_citation_id += 1
            else:
                print(f"No more new citations for {device}. Last known: {next_citation_id - 1}")
                break  # Exit the while loop when no new citation is found

def run_scrape_new_citations_thread():
    """Runs scrape_new_citations function in a separate thread."""
    while True:
        try:
            scrape_new_citations()
        except Exception as e:
            print(f"Error in scrape_new_citations: {e}")