DrOlu RPA: Enterprise Robotic Process Automation Framework
Open-Source UiPath Alternative
Enterprise-grade web automation for sites without APIs using Playwright + Python + mitmproxy + open-source RPA frameworks.
What I Do
I am an enterprise-grade RPA framework that automates web-based business processes without requiring APIs. Unlike commercial tools like UiPath or Automation Anywhere, I leverage powerful open-source technologies:
- Browser Automation: Playwright for reliable cross-browser automation with auto-waiting
- Traffic Analysis: mitmproxy for understanding hidden APIs and network requests
- Data Extraction: BeautifulSoup, lxml, and Playwright for structured data scraping
- Workflow Orchestration: Multi-step process automation with state management
- Visual Automation: OCR and image recognition for legacy systems
- Scheduling: Cron-based and event-driven task execution
- Error Recovery: Intelligent retry mechanisms and failure handling
When to Use Me
Use this skill when you need to:
- Automate repetitive web-based tasks (data entry, form filling, report generation)
- Extract data from websites without APIs
- Automate legacy web applications
- Build unattended automation bots
- Create scheduled data collection workflows
- Automate multi-step business processes across multiple websites
- Handle file uploads/downloads automatically
- Automate authentication flows (login, MFA, SSO)
Core Stack Components
1. Playwright - Browser Automation Engine
Purpose: Reliable cross-browser automation with modern web support
Why Playwright over Selenium:
- Auto-waiting for elements (no explicit waits needed)
- Native support for shadow DOM, iframes, and web components
- Built-in screenshot, video, and trace recording
- Network interception and mocking
- Multi-browser support (Chromium, Firefox, WebKit)
- Mobile emulation
- Persistent contexts for session reuse
Installation:
# Install with uv
uv add playwright
# Install browsers
uv run playwright install chromium
uv run playwright install firefox # Optional
uv run playwright install webkit # Optional
2. mitmproxy - Traffic Analysis
Purpose: Discover hidden APIs and understand network behavior
Use Cases:
- Capture API endpoints the website uses internally
- Understand request/response patterns
- Replay captured requests with curl
- Identify authentication mechanisms
Installation:
uv tool install mitmproxy
3. Python Libraries
Core Libraries:
# Browser automation
uv add playwright
# Data extraction
uv add beautifulsoup4 lxml html5lib
# HTTP requests
uv add httpx aiohttp requests
# Data processing
uv add pandas openpyxl xlrd
# OCR and vision
uv add pytesseract pillow opencv-python
# Scheduling
uv add schedule apscheduler
# Workflow orchestration
uv add prefect # or temporalio
# Configuration
uv add pydantic python-dotenv
# Logging
uv add structlog loguru
4. Additional RPA Frameworks
RPA Framework (Robot Framework):
uv add rpaframework rpaframework-browser
Botcity:
uv add botcity-framework-core botcity-framework-web
Core Automation Patterns
Pattern 1: Basic Page Automation
#!/usr/bin/env python3
"""Basic page automation - run with: uv run script.py"""
from playwright.sync_api import sync_playwright, Page, Browser
from typing import Optional
import structlog
log = structlog.get_logger()
class PageAutomator:
"""Base class for page automation."""
def __init__(self, headless: bool = True, slow_mo: int = 0):
self.headless = headless
self.slow_mo = slow_mo
self.browser: Optional[Browser] = None
self.page: Optional[Page] = None
self.playwright = None
def __enter__(self):
self.playwright = sync_playwright().start()
self.browser = self.playwright.chromium.launch(
headless=self.headless,
slow_mo=self.slow_mo
)
self.page = self.browser.new_page()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self.browser:
self.browser.close()
if self.playwright:
self.playwright.stop()
def navigate(self, url: str, wait_until: str = "networkidle"):
"""Navigate to URL and wait for page load."""
log.info("navigating", url=url)
self.page.goto(url, wait_until=wait_until)
def click(self, selector: str, timeout: int = 30000):
"""Click element with auto-wait."""
log.info("clicking", selector=selector)
self.page.click(selector, timeout=timeout)
def fill(self, selector: str, value: str):
"""Fill input field."""
log.info("filling", selector=selector)
self.page.fill(selector, value)
def get_text(self, selector: str) -> str:
"""Get text content of element."""
return self.page.text_content(selector)
def screenshot(self, path: str):
"""Take screenshot."""
self.page.screenshot(path=path, full_page=True)
def wait_for_selector(self, selector: str, timeout: int = 30000):
"""Wait for element to appear."""
self.page.wait_for_selector(selector, timeout=timeout)
# Usage
if __name__ == "__main__":
with PageAutomator(headless=False, slow_mo=100) as bot:
bot.navigate("https://example.com")
bot.screenshot("example.png")
Pattern 2: Async Automation for Performance
#!/usr/bin/env python3
"""Async automation for parallel tasks - run with: uv run script.py"""
import asyncio
from playwright.async_api import async_playwright, Page
from typing import Any
import structlog
log = structlog.get_logger()
class AsyncPageAutomator:
"""Async page automation for parallel processing."""
def __init__(self, headless: bool = True):
self.headless = headless
self.browser = None
self.playwright = None
async def __aenter__(self):
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(headless=self.headless)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.browser:
await self.browser.close()
if self.playwright:
await self.playwright.stop()
async def new_page(self) -> Page:
"""Create new page context."""
context = await self.browser.new_context()
return await context.new_page()
async def process_url(self, url: str) -> dict[str, Any]:
"""Process a single URL."""
page = await self.new_page()
try:
await page.goto(url, wait_until="networkidle")
title = await page.title()
content = await page.content()
return {"url": url, "title": title, "content_length": len(content)}
finally:
await page.close()
async def process_urls_parallel(self, urls: list[str], max_concurrent: int = 5) -> list[dict]:
"""Process multiple URLs in parallel."""
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_process(url: str):
async with semaphore:
return await self.process_url(url)
tasks = [limited_process(url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
async def main():
urls = [
"https://example.com",
"https://httpbin.org/html",
"https://quotes.toscrape.com",
]
async with AsyncPageAutomator(headless=True) as bot:
results = await bot.process_urls_parallel(urls)
for result in results:
if isinstance(result, dict):
log.info("processed", **result)
if __name__ == "__main__":
asyncio.run(main())
Pattern 3: Session Persistence
#!/usr/bin/env python3
"""Session persistence for login state - run with: uv run script.py"""
from playwright.sync_api import sync_playwright
from pathlib import Path
import json
class SessionManager:
"""Manage browser sessions with persistence."""
def __init__(self, storage_dir: str = "./sessions"):
self.storage_dir = Path(storage_dir)
self.storage_dir.mkdir(exist_ok=True)
def save_session(self, page, session_name: str):
"""Save browser session (cookies + storage)."""
context = page.context
# Save storage state (cookies + localStorage)
storage_path = self.storage_dir / f"{session_name}_storage.json"
context.storage_state(path=str(storage_path))
print(f"Session saved: {storage_path}")
def load_session(self, browser, session_name: str):
"""Load browser session."""
storage_path = self.storage_dir / f"{session_name}_storage.json"
if storage_path.exists():
context = browser.new_context(storage_state=str(storage_path))
print(f"Session loaded: {storage_path}")
return context.new_page()
else:
print("No saved session found, creating new context")
return browser.new_page()
def clear_session(self, session_name: str):
"""Clear saved session."""
storage_path = self.storage_dir / f"{session_name}_storage.json"
if storage_path.exists():
storage_path.unlink()
def login_and_save_session():
"""Login once and save session for reuse."""
with sync_playwright() as p:
browser = p.chromium.launch(headless=False)
page = browser.new_page()
session_mgr = SessionManager()
# Navigate to login page
page.goto("https://example.com/login")
# Perform login
page.fill("#username", "your_username")
page.fill("#password", "your_password")
page.click("button[type=submit]")
# Wait for login to complete
page.wait_for_url("**/dashboard**")
# Save session
session_mgr.save_session(page, "my_account")
browser.close()
def use_saved_session():
"""Use previously saved session."""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
session_mgr = SessionManager()
# Load session - no login needed!
page = session_mgr.load_session(browser, "my_account")
# Navigate directly to protected page
page.goto("https://example.com/dashboard")
# You're already logged in
print(page.title())
browser.close()
Pattern 4: Network Request Interception
#!/usr/bin/env python3
"""Network interception for API discovery - run with: uv run script.py"""
from playwright.sync_api import sync_playwright, Route, Request
import json
class NetworkInterceptor:
"""Intercept and analyze network requests."""
def __init__(self):
self.captured_requests = []
self.captured_responses = []
def capture_request(self, route: Route, request: Request):
"""Capture request details."""
self.captured_requests.append({
"url": request.url,
"method": request.method,
"headers": dict(request.headers),
"post_data": request.post_data,
})
route.continue_()
def capture_api_calls(self, page, url_pattern: str = "**/api/**"):
"""Set up API call interception."""
page.route(url_pattern, self.capture_request)
def block_resources(self, page, resource_types: list[str] = None):
"""Block specified resource types for faster loading."""
if resource_types is None:
resource_types = ["image", "stylesheet", "font", "media"]
def block_handler(route: Route, request: Request):
if request.resource_type in resource_types:
route.abort()
else:
route.continue_()
page.route("**/*", block_handler)
def mock_api_response(self, page, url_pattern: str, mock_data: dict):
"""Mock API response for testing."""
def mock_handler(route: Route):
route.fulfill(
status=200,
content_type="application/json",
body=json.dumps(mock_data)
)
page.route(url_pattern, mock_handler)
def export_as_curl(self, request_data: dict) -> str:
"""Convert captured request to curl command."""
cmd = f"curl -X {request_data['method']} '{request_data['url']}'"
for key, value in request_data.get("headers", {}).items():
if key.lower() not in ["host", "content-length"]:
cmd += f" \\\n -H '{key}: {value}'"
if request_data.get("post_data"):
cmd += f" \\\n -d '{request_data['post_data']}'"
return cmd
def discover_hidden_apis():
"""Discover hidden API endpoints used by a website."""
interceptor = NetworkInterceptor()
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
# Capture all API calls
interceptor.capture_api_calls(page, "**/api/**")
interceptor.capture_api_calls(page, "**/graphql**")
# Navigate and interact
page.goto("https://example.com")
page.click("#load-more") # Trigger API calls
# Wait for network to settle
page.wait_for_load_state("networkidle")
# Export discovered APIs
print("Discovered API Endpoints:")
for req in interceptor.captured_requests:
print(f"\n{req['method']} {req['url']}")
print(interceptor.export_as_curl(req))
browser.close()
Selector Strategies
Priority Order (Most Reliable First)
Test IDs (most stable):
page.click("[data-testid='submit-button']") page.click("[data-cy='login-form']")Accessible Roles:
page.get_by_role("button", name="Submit") page.get_by_role("textbox", name="Email") page.get_by_role("link", name="Learn more")Labels and Text:
page.get_by_label("Email address") page.get_by_text("Sign in") page.get_by_placeholder("Enter your email")CSS Selectors (stable if well-structured):
page.click("#login-button") page.click(".btn-primary") page.click("form.login button[type=submit]")XPath (use sparingly):
page.click("//button[contains(text(), 'Submit')]") page.click("//div[@class='container']//input[@name='email']")
Selector Chaining
# Scope selectors to parent elements
form = page.locator("form.login-form")
form.locator("input[name=email]").fill("user@example.com")
form.locator("input[name=password]").fill("password123")
form.locator("button[type=submit]").click()
Waiting Strategies
# Auto-wait (built-in)
page.click("#button") # Waits automatically
# Explicit wait for element
page.wait_for_selector("#dynamic-content", state="visible")
# Wait for specific conditions
page.wait_for_function("document.querySelector('#status').innerText === 'Ready'")
# Wait for network
page.wait_for_load_state("networkidle")
page.wait_for_response("**/api/data")
# Wait for navigation
page.wait_for_url("**/success**")
Data Extraction Patterns
Table Extraction
#!/usr/bin/env python3
"""Extract table data - run with: uv run script.py"""
from playwright.sync_api import sync_playwright
import pandas as pd
def extract_table(page, table_selector: str) -> pd.DataFrame:
"""Extract HTML table to DataFrame."""
# Get headers
headers = page.locator(f"{table_selector} thead th").all_text_contents()
# Get rows
rows = page.locator(f"{table_selector} tbody tr").all()
data = []
for row in rows:
cells = row.locator("td").all_text_contents()
data.append(cells)
return pd.DataFrame(data, columns=headers)
def extract_table_with_pagination(page, table_selector: str, next_button: str) -> pd.DataFrame:
"""Extract table data across multiple pages."""
all_data = []
while True:
# Extract current page
df = extract_table(page, table_selector)
all_data.append(df)
# Check for next page
next_btn = page.locator(next_button)
if next_btn.is_visible() and next_btn.is_enabled():
next_btn.click()
page.wait_for_load_state("networkidle")
else:
break
return pd.concat(all_data, ignore_index=True)
Structured Data Extraction
#!/usr/bin/env python3
"""Extract structured data - run with: uv run script.py"""
from playwright.sync_api import sync_playwright
from pydantic import BaseModel
from typing import Optional
import json
class Product(BaseModel):
"""Product data model."""
name: str
price: float
description: Optional[str] = None
image_url: Optional[str] = None
rating: Optional[float] = None
def extract_products(page, container_selector: str) -> list[Product]:
"""Extract product data from page."""
products = []
items = page.locator(container_selector).all()
for item in items:
try:
product = Product(
name=item.locator(".product-name").text_content().strip(),
price=float(item.locator(".price").text_content().replace("$", "").strip()),
description=item.locator(".description").text_content().strip()
if item.locator(".description").count() > 0 else None,
image_url=item.locator("img").get_attribute("src"),
rating=float(item.locator(".rating").get_attribute("data-rating"))
if item.locator(".rating").count() > 0 else None,
)
products.append(product)
except Exception as e:
print(f"Error extracting product: {e}")
return products
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto("https://example-shop.com/products")
products = extract_products(page, ".product-card")
# Export to JSON
with open("products.json", "w") as f:
json.dump([p.model_dump() for p in products], f, indent=2)
browser.close()
if __name__ == "__main__":
main()
Form Automation
Complete Form Filling
#!/usr/bin/env python3
"""Form automation - run with: uv run script.py"""
from playwright.sync_api import sync_playwright, Page
from pydantic import BaseModel
from typing import Optional
from pathlib import Path
class FormData(BaseModel):
"""Form data model."""
first_name: str
last_name: str
email: str
phone: Optional[str] = None
company: Optional[str] = None
country: str = "United States"
agree_terms: bool = True
newsletter: bool = False
file_upload: Optional[str] = None
class FormAutomator:
"""Automate form filling."""
def __init__(self, page: Page):
self.page = page
def fill_text_field(self, selector: str, value: str):
"""Fill text input field."""
self.page.fill(selector, value)
def select_dropdown(self, selector: str, value: str):
"""Select dropdown option by value or label."""
self.page.select_option(selector, value)
def check_checkbox(self, selector: str, checked: bool = True):
"""Check or uncheck checkbox."""
if checked:
self.page.check(selector)
else:
self.page.uncheck(selector)
def select_radio(self, selector: str):
"""Select radio button."""
self.page.check(selector)
def upload_file(self, selector: str, file_path: str):
"""Upload file."""
self.page.set_input_files(selector, file_path)
def fill_form(self, form_data: FormData, field_mapping: dict[str, str]):
"""Fill entire form from data model."""
data = form_data.model_dump()
for field_name, selector in field_mapping.items():
if field_name not in data or data[field_name] is None:
continue
value = data[field_name]
element = self.page.locator(selector)
# Determine field type and fill accordingly
tag = element.evaluate("el => el.tagName.toLowerCase()")
input_type = element.get_attribute("type") or "text"
if tag == "select":
self.select_dropdown(selector, value)
elif tag == "textarea":
self.fill_text_field(selector, value)
elif input_type == "checkbox":
self.check_checkbox(selector, bool(value))
elif input_type == "radio":
if value:
self.select_radio(selector)
elif input_type == "file":
if value and Path(value).exists():
self.upload_file(selector, value)
else:
self.fill_text_field(selector, str(value))
def submit_form(self, submit_selector: str = "button[type=submit]"):
"""Submit form."""
self.page.click(submit_selector)
def automate_registration():
"""Example: Automate registration form."""
with sync_playwright() as p:
browser = p.chromium.launch(headless=False)
page = browser.new_page()
page.goto("https://example.com/register")
form = FormAutomator(page)
form_data = FormData(
first_name="John",
last_name="Doe",
email="john.doe@example.com",
phone="+1-555-123-4567",
company="Acme Inc",
country="United States",
agree_terms=True,
newsletter=False,
)
field_mapping = {
"first_name": "#firstName",
"last_name": "#lastName",
"email": "#email",
"phone": "#phone",
"company": "#company",
"country": "select[name=country]",
"agree_terms": "#agreeTerms",
"newsletter": "#newsletter",
}
form.fill_form(form_data, field_mapping)
form.submit_form()
# Wait for success
page.wait_for_url("**/success**")
print("Registration completed!")
browser.close()
if __name__ == "__main__":
automate_registration()
Error Handling and Recovery
Retry Pattern
#!/usr/bin/env python3
"""Error handling with retry - run with: uv run script.py"""
import asyncio
from functools import wraps
from typing import Callable, Any
import structlog
log = structlog.get_logger()
def retry(
max_attempts: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,)
):
"""Retry decorator with exponential backoff."""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def async_wrapper(*args, **kwargs) -> Any:
last_exception = None
current_delay = delay
for attempt in range(1, max_attempts + 1):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
log.warning(
"attempt_failed",
attempt=attempt,
max_attempts=max_attempts,
error=str(e)
)
if attempt < max_attempts:
await asyncio.sleep(current_delay)
current_delay *= backoff
raise last_exception
@wraps(func)
def sync_wrapper(*args, **kwargs) -> Any:
import time
last_exception = None
current_delay = delay
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
log.warning(
"attempt_failed",
attempt=attempt,
max_attempts=max_attempts,
error=str(e)
)
if attempt < max_attempts:
time.sleep(current_delay)
current_delay *= backoff
raise last_exception
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decorator
class RobustAutomator:
"""Automator with built-in error recovery."""
def __init__(self, page):
self.page = page
@retry(max_attempts=3, delay=1.0, exceptions=(Exception,))
def safe_click(self, selector: str):
"""Click with retry on failure."""
self.page.click(selector, timeout=10000)
@retry(max_attempts=3, delay=1.0)
def safe_fill(self, selector: str, value: str):
"""Fill with retry on failure."""
self.page.fill(selector, value, timeout=10000)
def click_with_fallback(self, selectors: list[str]):
"""Try multiple selectors until one works."""
for selector in selectors:
try:
if self.page.locator(selector).is_visible():
self.page.click(selector)
return True
except Exception:
continue
raise Exception(f"None of the selectors worked: {selectors}")
def handle_popup(self):
"""Dismiss common popups."""
popup_selectors = [
"[aria-label='Close']",
".popup-close",
"#cookie-accept",
".modal-close",
"button:has-text('Accept')",
"button:has-text('Close')",
]
for selector in popup_selectors:
try:
if self.page.locator(selector).is_visible():
self.page.click(selector)
log.info("popup_dismissed", selector=selector)
except Exception:
pass
def wait_and_retry(self, action: Callable, timeout: int = 30):
"""Wait for page stability then perform action."""
self.page.wait_for_load_state("networkidle")
self.handle_popup()
return action()
Workflow Orchestration
Multi-Step Workflow
#!/usr/bin/env python3
"""Workflow orchestration - run with: uv run script.py"""
from playwright.sync_api import sync_playwright, Page
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any, Optional
import json
from datetime import datetime
import structlog
log = structlog.get_logger()
class StepStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class WorkflowStep:
"""Workflow step definition."""
name: str
action: Callable[[Page], Any]
depends_on: list[str] = field(default_factory=list)
retry_count: int = 3
status: StepStatus = StepStatus.PENDING
result: Any = None
error: Optional[str] = None
@dataclass
class WorkflowContext:
"""Shared context between workflow steps."""
data: dict = field(default_factory=dict)
variables: dict = field(default_factory=dict)
class Workflow:
"""Multi-step workflow orchestrator."""
def __init__(self, name: str):
self.name = name
self.steps: list[WorkflowStep] = []
self.context = WorkflowContext()
self.start_time: Optional[datetime] = None
self.end_time: Optional[datetime] = None
def add_step(
self,
name: str,
action: Callable[[Page], Any],
depends_on: list[str] = None,
retry_count: int = 3
):
"""Add step to workflow."""
step = WorkflowStep(
name=name,
action=action,
depends_on=depends_on or [],
retry_count=retry_count
)
self.steps.append(step)
return self
def get_step(self, name: str) -> Optional[WorkflowStep]:
"""Get step by name."""
for step in self.steps:
if step.name == name:
return step
return None
def can_run_step(self, step: WorkflowStep) -> bool:
"""Check if step dependencies are met."""
for dep_name in step.depends_on:
dep_step = self.get_step(dep_name)
if not dep_step or dep_step.status != StepStatus.COMPLETED:
return False
return True
def run_step(self, step: WorkflowStep, page: Page) -> bool:
"""Run a single step with retry."""
step.status = StepStatus.RUNNING
for attempt in range(1, step.retry_count + 1):
try:
log.info("step_running", step=step.name, attempt=attempt)
step.result = step.action(page)
step.status = StepStatus.COMPLETED
log.info("step_completed", step=step.name)
return True
except Exception as e:
log.warning("step_failed", step=step.name, attempt=attempt, error=str(e))
step.error = str(e)
if attempt == step.retry_count:
step.status = StepStatus.FAILED
return False
return False
def run(self, page: Page) -> bool:
"""Execute workflow."""
self.start_time = datetime.now()
log.info("workflow_started", name=self.name)
pending_steps = list(self.steps)
max_iterations = len(self.steps) * 2
iteration = 0
while pending_steps and iteration < max_iterations:
iteration += 1
made_progress = False
for step in list(pending_steps):
if step.status == StepStatus.PENDING and self.can_run_step(step):
success = self.run_step(step, page)
pending_steps.remove(step)
made_progress = True
if not success:
# Skip dependent steps
for other in pending_steps:
if step.name in other.depends_on:
other.status = StepStatus.SKIPPED
pending_steps.remove(other)
if not made_progress:
break
self.end_time = datetime.now()
# Check final status
failed_steps = [s for s in self.steps if s.status == StepStatus.FAILED]
if failed_steps:
log.error("workflow_failed", failed_steps=[s.name for s in failed_steps])
return False
log.info("workflow_completed", name=self.name, duration=str(self.end_time - self.start_time))
return True
def get_report(self) -> dict:
"""Generate workflow execution report."""
return {
"name": self.name,
"start_time": self.start_time.isoformat() if self.start_time else None,
"end_time": self.end_time.isoformat() if self.end_time else None,
"duration": str(self.end_time - self.start_time) if self.end_time and self.start_time else None,
"steps": [
{
"name": s.name,
"status": s.status.value,
"error": s.error,
}
for s in self.steps
]
}
# Example workflow
def create_data_extraction_workflow():
"""Create a data extraction workflow."""
def login(page: Page):
page.goto("https://example.com/login")
page.fill("#username", "user")
page.fill("#password", "pass")
page.click("button[type=submit]")
page.wait_for_url("**/dashboard**")
return True
def navigate_to_reports(page: Page):
page.click("a[href='/reports']")
page.wait_for_selector(".reports-table")
return True
def download_report(page: Page):
page.click("#download-csv")
# Wait for download
with page.expect_download() as download_info:
page.click("#confirm-download")
download = download_info.value
download.save_as("./report.csv")
return download.path()
def logout(page: Page):
page.click("#logout")
page.wait_for_url("**/login**")
return True
workflow = Workflow("Daily Report Download")
workflow.add_step("login", login)
workflow.add_step("navigate", navigate_to_reports, depends_on=["login"])
workflow.add_step("download", download_report, depends_on=["navigate"])
workflow.add_step("logout", logout, depends_on=["download"])
return workflow
if __name__ == "__main__":
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
workflow = create_data_extraction_workflow()
success = workflow.run(page)
report = workflow.get_report()
print(json.dumps(report, indent=2))
browser.close()
Scheduling
Cron-Based Scheduling
#!/usr/bin/env python3
"""Scheduled automation - run with: uv run scheduler.py"""
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from playwright.sync_api import sync_playwright
import structlog
from datetime import datetime
log = structlog.get_logger()
def run_daily_report():
"""Run daily report automation."""
log.info("job_started", job="daily_report", time=datetime.now().isoformat())
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
try:
page.goto("https://example.com/reports")
# ... automation logic
log.info("job_completed", job="daily_report")
except Exception as e:
log.error("job_failed", job="daily_report", error=str(e))
finally:
browser.close()
def run_hourly_check():
"""Run hourly status check."""
log.info("job_started", job="hourly_check", time=datetime.now().isoformat())
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
try:
page.goto("https://example.com/status")
status = page.locator("#system-status").text_content()
log.info("status_check", status=status)
except Exception as e:
log.error("job_failed", job="hourly_check", error=str(e))
finally:
browser.close()
def main():
scheduler = BlockingScheduler()
# Daily at 6 AM
scheduler.add_job(
run_daily_report,
CronTrigger(hour=6, minute=0),
id="daily_report",
name="Daily Report Download"
)
# Every hour
scheduler.add_job(
run_hourly_check,
CronTrigger(minute=0),
id="hourly_check",
name="Hourly Status Check"
)
# Every weekday at 9 AM
scheduler.add_job(
lambda: log.info("weekday_job"),
CronTrigger(day_of_week="mon-fri", hour=9),
id="weekday_job"
)
log.info("scheduler_started", jobs=len(scheduler.get_jobs()))
try:
scheduler.start()
except KeyboardInterrupt:
log.info("scheduler_stopped")
if __name__ == "__main__":
main()
Module Reference
Core Modules
- rpa-browser-automation.md - Advanced Playwright patterns, multi-browser, mobile emulation, stealth mode
- rpa-data-extraction.md - Scraping, parsing, structured data extraction, tables, infinite scroll
- rpa-form-automation.md - Form filling, dropdowns, file uploads, CAPTCHA handling
Workflow Modules
- rpa-workflows.md - Multi-step orchestration, state machines, checkpoints, Prefect/Temporal
- rpa-scheduling.md - Cron jobs, APScheduler, event-driven triggers, queue-based processing
Advanced Modules
- rpa-authentication.md - Login automation, MFA/TOTP, OAuth, SSO, session management
- rpa-error-handling.md - Retry patterns, circuit breaker, recovery, monitoring, alerting
- rpa-ocr-vision.md - Visual automation, OCR with Tesseract, image recognition
Enterprise Modules (v2.0)
- rpa-credentials.md - HashiCorp Vault, AWS Secrets Manager, Azure Key Vault, encrypted local storage
- rpa-cicd.md - GitHub Actions, GitLab CI, Docker containers, Kubernetes deployment
- rpa-observability.md - Structlog logging, Prometheus metrics, OpenTelemetry, Grafana dashboards
- rpa-desktop.md - Windows/macOS desktop automation with pyautogui and pywinauto
- rpa-documents.md - Excel (openpyxl), PDF (pypdf/reportlab), Word (python-docx), Email (IMAP/SMTP)
Best Practices
1. Selector Stability
- Prefer data-testid over CSS classes
- Use role-based selectors when possible
- Avoid XPath unless necessary
- Create selector constants/enums
2. Performance
- Block unnecessary resources (images, fonts, CSS)
- Use parallel processing for multiple pages
- Implement connection pooling
- Cache session state
3. Reliability
- Always use explicit waits
- Implement retry mechanisms
- Handle popups and overlays
- Save screenshots on failure
4. Maintainability
- Use Page Object Model pattern
- Separate selectors from logic
- Create reusable automation components
- Document automation workflows
5. Security
- Never hardcode credentials
- Use environment variables or secret managers
- Encrypt session storage
- Audit automation access
Installation Quick Start
# Create project
mkdir my-rpa-bot && cd my-rpa-bot
# Initialize with uv
uv init
# Add core dependencies
uv add playwright beautifulsoup4 pandas httpx pydantic structlog
# Install browsers
uv run playwright install chromium
# Create first bot
cat > bot.py << 'EOF'
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(headless=False)
page = browser.new_page()
page.goto("https://example.com")
print(page.title())
browser.close()
EOF
# Run
uv run bot.py
Enterprise Edition (v2.0) Features
Credentials & Secrets Management
# See rpa-credentials.md for full details
uv add hvac boto3 azure-identity azure-keyvault-secrets cryptography
- HashiCorp Vault integration
- AWS Secrets Manager
- Azure Key Vault
- Encrypted local storage with Fernet
CI/CD Integration
# See rpa-cicd.md for full details
# Docker, GitHub Actions, GitLab CI, Kubernetes
- Containerized bot deployment
- GitHub Actions workflows
- GitLab CI pipelines
- Kubernetes CronJobs and Deployments
Observability
# See rpa-observability.md for full details
uv add structlog prometheus-client opentelemetry-api opentelemetry-sdk
- Structured logging with structlog
- Prometheus metrics collection
- OpenTelemetry tracing
- Grafana dashboard templates
- PagerDuty/Slack alerting
Desktop Auto
…(truncated)