Reverse Engineering the ReadyMode API for QA Automation: A 2-Day Sprint to 6x Faster Lead Delivery

# PATTERN: Use CDP mode for stable browser automation
from seleniumbase import SB

with SB(uc=True, headless=False, chromium_arg="--deny-permission-prompts") as sb:
    sb.activate_cdp_mode(readymode_config.login_url)

    # KEY: Wait for elements explicitly - no brittle XPaths or fixed sleeps
    sb.cdp.wait_for_element('input[id="login-account"]', timeout=10)
    sb.cdp.send_keys('input[id="login-account"]', username)
    # ... handle password, admin checkbox ...
    sb.cdp.click('input[type="submit"]')

    # PATTERN: Handle optional screens gracefully
    try:
        sb.cdp.click('input[value="Continue"]', timeout=5)
    except:
        pass  # Not all accounts have this screen

    # CRITICAL: Capture cookies for later HTTP calls
    cookies = {c["name"]: c["value"] for c in sb.get_cookies()}
    return cookies  # Hand off to faster HTTP operations

# PATTERN: SQLMesh model returns pandas DataFrame rows via yield
@model(
    "readymode_data.call_metadata",
    cron="@hourly",
    kind=dict(
        name=ModelKindName.INCREMENTAL_BY_UNIQUE_KEY,
        unique_key="call_log_id",
        lookback=3  # KEY: Re-check last 3 days for late updates
    )
)
def execute(context, start, end, **kwargs):
    # Modal functions run in isolated containers
    get_cookies = modal.Function.from_name("app", "get_readymode_cookies")
    fetch_page = modal.Function.from_name("app", "fetch_metadata_page")

    cookies = get_cookies.remote()

    # PATTERN: Process scheduler window incrementally
    for day in days_between(start, end):
        first_page = fetch_page.remote(cookies, day, page=0)
        total_pages = first_page.get("pages", 1)

        # Convert API response to DataFrame rows
        yield from parse_to_dataframe(first_page)

        if total_pages > 1:
            # KEY: Parallel fetch with Modal's .map()
            results = fetch_page.map(
                [cookies] * (total_pages - 1),
                # ... other params ...
            )
            for result in results:
                yield from parse_to_dataframe(result)

# PATTERN: Streaming download for large files
def download_recording(rec_id: str, cookies: dict):
    # Build authenticated URL
    download_url = f"{base_url}/{quote(rec_id, safe='/:')}"

    with httpx.Client(timeout=120) as client:
        # KEY: Stream to avoid memory issues
        with client.stream("GET", download_url, cookies=cookies) as r:
            r.raise_for_status()
            # ... write to file in chunks ...

    return destination_path  # Path for downstream processing

# PATTERN: GPU container with snapshot for instant warm starts
@app.cls(
    image=whisper_image,
    gpu="A10G",
    volumes={"/cache": cache_vol},
    enable_memory_snapshot=True,  # KEY: Pre-load models
    experimental_options={"enable_gpu_snapshot": True}
)
class WhisperX:
    @modal.enter(snap=True)
    def setup(self):
        # Models loaded once, reused across invocations
        self.model = whisperx.load_model(
            "large-v3-turbo",
            device="cuda",
            compute_type="float16"
        )
        self.align_model = whisperx.load_align_model("en")
        self.diarize_model = DiarizationPipeline(...)

    @modal.method()
    def transcribe(self, audio_path: str) -> pd.DataFrame:
        audio = whisperx.load_audio(audio_path)

        # Step 1: Transcribe
        result = self.model.transcribe(audio, batch_size=128)

        # Step 2: Align for word-level timestamps
        result = whisperx.align(result, self.align_model, audio)

        # Step 3: Diarize to identify speakers
        diarize_segments = self.diarize_model(audio)

        # CRITICAL: Return DataFrame for SQLMesh
        return pd.DataFrame({
            'start': [...],
            'end': [...],
            'speaker': [...],  # SPK_A, SPK_B, etc
            'text': [...]
        })

# PATTERN: SQLMesh model yielding DataFrame rows
@model("readymode_data.call_transcripts", cron="@hourly")
def execute(context, start, end, **kwargs) -> pd.DataFrame:
    # KEY: Find recordings without transcripts
    candidates = context.fetchdf("""
        SELECT r.* FROM call_recordings r
        WHERE NOT EXISTS (
            SELECT 1 FROM call_transcripts t
            WHERE t.call_log_id = r.call_log_id
        )
        AND r.call_time BETWEEN %(start)s AND %(end)s
    """, {"start": start, "end": end})

    if candidates.empty:
        return  # Nothing to process

    # PATTERN: Batch transcribe on GPU with Modal
    WhisperX = modal.Cls.from_name("app", "WhisperX")
    unique_paths = candidates["download_path"].unique()

    # KEY: .map() for parallel GPU processing
    transcript_dfs = WhisperX().transcribe.map(unique_paths)

    # Join transcripts back to call metadata
    for path, transcript_df in zip(unique_paths, transcript_dfs):
        call_ids = candidates[candidates.download_path == path]["call_log_id"]
        for call_id in call_ids:
            transcript_df["call_log_id"] = call_id
            yield transcript_df  # SQLMesh handles the rest

# PATTERN: Real-time queries in Marimo interactive notebooks
import marimo as mo
import pandas as pd

# KEY: Filter to calls with transcripts available
recent_calls_query = """
    SELECT
        call_log_id,
        agent_name,
        call_time,
        disposition,
        duration_seconds
        -- ... other fields ...
    FROM call_metadata cm
    WHERE call_time >= now() - interval '7 days'
        AND EXISTS (
            SELECT 1 FROM call_transcripts t
            WHERE t.call_log_id = cm.call_log_id
        )
    ORDER BY call_log_id DESC
"""

recent_calls = pd.read_sql(recent_calls_query, connection)

# PATTERN: Interactive table with row selection
selected = mo.ui.table(
    recent_calls,
    on_select=lambda df: load_call_details(df['call_log_id'])
)

# PATTERN: LLM scoring with structured output
import dspy

class QAScoring(dspy.Signature):
    """Score call quality on multiple dimensions"""
    transcript: str = dspy.InputField()

    # CRITICAL: Structured output for consistent scoring
    rapport_score: int = dspy.OutputField(desc="1-10")
    discovery_score: int = dspy.OutputField(desc="1-10")
    objections_handled: bool = dspy.OutputField()
    coaching_notes: str = dspy.OutputField()

# KEY: Chain-of-thought for reasoning transparency
program = dspy.ChainOfThought(QAScoring)

def generate_report(call_id: int) -> pd.DataFrame:
    # Fetch transcript
    transcript = fetch_transcript(call_id)

    # Generate scores with reasoning
    report = program(transcript=transcript)

    # PATTERN: Return DataFrame for display
    return pd.DataFrame({
        'Metric': ['Rapport', 'Discovery', ...],
        'Score': [report.rapport_score, ...],
        'Notes': [report.coaching_notes, ...]
    })

# Display in Marimo
mo.ui.table(generate_report(selected_call_id))

# Core components to implement:
modal_functions.py     # Browser automation, cookie capture, API calls
sqlmesh/models/        # Incremental data models (return DataFrames!)
  ├── call_metadata.py    # API → warehouse ingestion
  ├── call_recordings.py  # Download orchestration
  └── call_transcripts.py # GPU transcription pipeline
notebooks/qa_dashboard.py # Marimo interactive app

# Key patterns to replicate:
- SeleniumBase CDP for stable browser automation
- Cookie handoff from browser to HTTP clients
- SQLMesh incremental processing with unique keys
- Modal .map() for parallel execution
- GPU snapshots for warm starts
- DataFrame yields from SQLMesh models