InvoiceScraper.py

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By

from pathlib import Path
from dotenv import dotenv_values

import paho.mqtt.client as mqtt
import pyotp
import json
import time

class Invoice:
    """
    Represents a single invoice with price and url.
    """
    def __init__(self, price, link):
        self.price  = price
        self.url    = link
        
    def to_json(self):
        """
        Converts the Invoice object to a JSON string.
        Returns:
            str: JSON representation of the Invoice.
        """
        return json.dumps({
            "price": self.price.replace(",", ".") if self.price else None,
            "link": self.url
        }, indent=4)

    def __str__(self):
        return f"Price: {self.price}, URL: {self.url}"

class MqttPublisher:
    """
    Publishes messages to an MQTT broker.
    """
    def __init__(self, broker, port, topic, username, password):
        self.client = mqtt.Client()                         # Create a new MQTT client instance
        self.client.username_pw_set(username, password)     # Set username and password
        self.client.connect(broker, port)                   # Connect to the MQTT broker
        self.topic = topic                                  # Set the topic to publish to
        
    def publish(self, message):
        """
        Publishes a message to the specified MQTT topic.
        Args:
            message (str): The message to publish.
        """
        self.client.publish(self.topic, message)            # Publish the message to the specified topic

def read_env_variables():
    """
    Reads environment variables from a .env file for credentials and settings.
    Returns:
        dict: Dictionary of environment variables.
    """
    env_path = Path(__file__)                       # Get the path of the current script
    env_file = f"{env_path.parent}/config.env"      # Path to the .env file
    return dotenv_values(env_file)                  # Load environment variables

base_url = "https://www.amazon.de"                  # Base URL for Amazon.de (adjust if you're in a different region)
env_variables = read_env_variables()                # Load credentials/settings

def page_load_complete(driver):
    """
    Checks if the page has completely loaded.
    Args:
        driver (webdriver): Selenium WebDriver instance.
    Returns:
        bool: True if page is fully loaded, False otherwise.
    """
    # Use JavaScript to check if document is ready
    return driver.execute_script("return document.readyState") == "complete"

def generate_otp(otp_secret):
    """
    Generates a one-time password (OTP) using the provided secret.
    Args:
        otp_secret (str): The OTP secret.
    Returns:
        str: The generated OTP code.
    """
    totp = pyotp.TOTP(otp_secret)   # Create TOTP object
    return totp.now()               # Generate current OTP and return it

def login(driver):
    """
    Logs into Amazon using credentials and OTP from environment variables.
    Args:
        driver (webdriver): Selenium WebDriver instance.
    """
    try:
        driver.get(f"{base_url}/gp/css/order-history?ref_=nav_orders_first")    # Open Amazon order history
        WebDriverWait(driver, 10).until(page_load_complete)                     # Wait for page to load
        
        email = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "ap_email")))
        email.send_keys(env_variables["EMAIL"])                                 # Enter email
        
        WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.ID, "continue"))).click()            # Click continue to proceed to the password page
        
        password = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "ap_password")))
        password.send_keys(env_variables["PASSWORD"])                           # Enter password
        
        WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.ID, "signInSubmit"))).click()        # Click sign in button
        
        otp_input   = WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "auth-mfa-otpcode")))
        otp_code    = generate_otp(env_variables["OTP_SECRET"])                 # Generate OTP code
        otp_input.send_keys(otp_code)                                           # Enter OTP in the input field
        
        WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.ID, "auth-signin-button"))).click()  # Click sign in button
        print("--> Login successful!")
        
    except Exception as e:
        print(f"--> Login process failed: {e}. Closing driver.")
        driver.quit()

def get_invoice(driver):
    invoice = Invoice(price = None, link = None)                # Initialize an Invoice object
    
    try:
        driver.get(f"{base_url}/cpe/myinvoices")                # Navigate to the monthly bill page
        WebDriverWait(driver, 10).until(page_load_complete)     # Wait for the page to load
        
        invoiceContainer        = driver.find_element("id", "myInvoicesPageContainer")                                      # Find the invoice container which contains all invoices
        headerContainer         = invoiceContainer.find_element(By.CLASS_NAME, "a-box-title")                               # Find the header container (current year)   
        currentInvoiceContainer = headerContainer.find_element(By.XPATH, 'following::div[contains(@class, "a-box")][1]')    # Find the first invoice container after the header (current month)
        
        url             = currentInvoiceContainer.find_element(By.TAG_NAME, "a")    # Find the html link in the invoice container
        invoice.url     = url.get_attribute("href")                                 # Extract the URL from the link element
        invoice.price   = currentInvoiceContainer.text.strip()[-5:]                 # Extract the price from the text
        
        return invoice  # Return the Invoice object with price and link
    
    except Exception as e:
        print(f"Could not retrieve monthly bill: {e}")

def main():
    # Run Chrome in headless mode (no GUI)
    options = Options()
    options.add_argument("--headless")
    options.add_argument("--disable-gpu")
    options.add_argument("--window-size=1920,1080") 
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")
    
    service = Service(executable_path="/usr/bin/chromedriver")
    driver  = webdriver.Chrome(service=service, options=options)

    publisher = MqttPublisher(
        broker      = env_variables["MQTT_BROKER"],
        port        = int(env_variables["MQTT_PORT"]),
        topic       = env_variables["MQTT_TOPIC"],
        username    = env_variables["MQTT_USERNAME"],
        password    = env_variables["MQTT_PASSWORD"]
    )
    
    try:
        login(driver)                       # Perform login
        while True:
            invoice = get_invoice(driver)   # Retrieve the current invoice
            
            if invoice and invoice.price and invoice.url:  # Check if invoice is valid
                print(f"--> Monthly bill found: {invoice}")
                publisher.publish(invoice.to_json())       # Publish the invoice as JSON
            else:
                print("--> No monthly bill found or price is empty.")

            time.sleep(120)                 # Wait for 2 minutes before repeating

    finally:
        driver.quit()                       # Close the browser when done

if __name__ == "__main__":
    main()