Automated Shift Agent Notifier using Web Scraping

Description

This Python script automates the process of monitoring shift pickups on ShiftAgent. It logs into the ShiftAgent website, navigates to the pickups page, and continuously checks for new shift pickup opportunities. When new shifts become available, it sends a text message / Email notification and updates a JSON file with the latest data.

Technical Breakdown

1. Setup and Dependencies
   • The script uses Playwright for web automation and interacts with SMTP to send email notifications.
   • It imports necessary libraries including playwright, json, datetime, and smtplib.
2. Text Message Notification Function
   • Defines a send_text_message() function that formats new shift data and sends it via email (which can be routed to SMS).
   • Uses SMTP to send the formatted message through a Gmail account.
3. Main Scraping Function
   • The scrape_shiftagent() function handles the core web scraping logic.
   • Initializes a Playwright browser instance (headless by default for background operation).
4. Login Process
   • Navigates to the ShiftAgent login page and waits for the login form to render.
   • Fills in login credentials and submits the form.
   • Waits for successful navigation to the user’s work periods page.
5. Shift Pickup Monitoring
   • Navigates to the pickups page and enters a continuous monitoring loop.
   • Refreshes the page every 30 seconds and checks for new shift pickup opportunities.
6. Data Processing and Notification
   • Compares newly scraped data with previous data to detect changes.
   • If new shifts are available, it triggers the text message notification.
7. Data Persistence
   • Updates a JSON file (shiftagent_data.json) with the latest shift data and timestamp.
8. Error Handling and Cleanup
   • Implements try-except blocks to catch and log any errors during the process.
   • Takes screenshots on errors for debugging purposes.
   • Ensures proper closure of browser contexts and instances.

from playwright.sync_api import sync_playwright, expect, TimeoutError
import time
import json
from datetime import datetime
import smtplib
from email.mime.text import MIMEText


def send_text_message(new_data):
    try:
        # Email credentials and setup
        sender_email = "john@example.com"
        sender_password = "password"
        recipient_email = "john@example.com" # Email
        # recipient_email = "YourPhoneNumber@text.att.net" # SMS message

        # Format message
        formatted_data = "\n\n".join([f"{line.split(' ')[0]}: {' '.join(line.split()[6:])}" for line in new_data])

        # Create the email (no subject)
        msg = MIMEText(formatted_data)
        msg['From'] = sender_email
        msg['To'] = recipient_email

        # Send email
        server = smtplib.SMTP('smtp.gmail.com', 587)
        server.starttls()
        server.login(sender_email, sender_password)
        server.sendmail(sender_email, recipient_email, msg.as_string())
        server.quit()

        print(f"Text message sent to {recipient_email}")

    except Exception as e:
        print(f"Failed to send text message: {str(e)}")


def scrape_shiftagent():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True) # False to view
        context = browser.new_context()
        page = context.new_page()

        try:
            print("Navigating to the login page...")
            page.goto("https://shiftagent.org/sa/#/login", wait_until="networkidle", timeout=60000)

            print("Waiting for login form to be rendered...")
            page.wait_for_selector('#fullscreen-overlay .login', state='visible', timeout=30000)

            print("Filling in login details...")
            page.fill('#fullscreen-overlay .login input[type="email"]', 'jbiskup-weise@student.linfield.com')
            page.fill('#fullscreen-overlay .login input[type="password"]', '43164316')

            print("Waiting for a moment...")
            time.sleep(1)

            print("Attempting to click submit button...")
            page.press('#fullscreen-overlay .login input[type="password"]', 'Enter')

            print("Waiting for navigation after login...")
            time.sleep(3)
            page.wait_for_url("https://shiftagent.org/sa/#/workPeriods/me", timeout=20000)

            if "/sa/#/login" in page.url:
                print("Login failed. Still on login page.")
                page.screenshot(path="login_failed.png")
                return

            print("Successfully logged in. Navigating to pickups page...")
            page.goto("https://shiftagent.org/sa/#/pickups", wait_until="networkidle", timeout=30000)

            previous_data = []
            while True:
                print("Refreshing the page...")
                page.reload(wait_until="networkidle")
                page.wait_for_selector('.msg_content', timeout=30000)

                elements = page.query_selector_all('.msg_content')
                current_data = [element.inner_text() for element in elements]

                if current_data != previous_data:
                    print("New data detected!")
                    new_items = list(set(current_data) - set(previous_data))
                    send_text_message(new_items)

                    # Update JSON file
                    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                    output = {
                        "data": current_data,
                        "last_updated": timestamp
                    }
                    with open("../shiftagent_data.json", "w") as json_file:
                        json.dump(output, json_file, indent=4)
                    print(f"Data successfully updated in shiftagent_data.json at {timestamp}")

                    previous_data = current_data

                else:
                    print("No new data detected.")

                time.sleep(30)

        except Exception as e:
            print(f"An error occurred: {str(e)}")
            page.screenshot(path="../error_screenshot.png")

        finally:
            context.close()
            browser.close()


if __name__ == "__main__":
    scrape_shiftagent()