# Scraper status on Ops — before and after (Feb 2026)

Simple guide to how the **worker** tells **Baskit Ops** what each retailer scraper is doing, and what we fixed when Checkers looked “stuck” while Woolworths looked active.

**Code touched:** `pipeline/v2/alerts.py`, `pipeline/checkers_scraper.py`, `pipeline/woolworths_scraper.py`, `pipeline/pnp_scraper.py`, `pipeline/v2/status_check.py`  
**Ops side (unchanged):** `POST /ingest/v1/status`, three-lamp board on **Scrape runs** (`PipelineHeartbeatModel` → one row per `scraper:{slug}`).

---

## Overview

The scraper **pipeline** (scrape → SQLite → push to Ops) did not change. Only **how live status is reported** to Ops changed.

| Piece | What it does |
|--------|----------------|
| **Worker** | Runs `run_pipeline_v2.py --rotate`, scrapes into `checkers.db` / `woolworths.db`, pushes via `push_to_ops.py` |
| **Ops ingest API** | Stores products/prices and updates `pipeline_heartbeat` rows for PnP, Checkers, Woolworths |
| **Ops console** | Shows three lamps (Ready / Busy / Error) plus ingest **run** history |
| **Customer app** | Reads catalogue from Ops API — not from the worker |

---

## Stakeholder summary — did we change how the worker works?

**Short answer:** No change to **what** the worker does (scrape catalogues, push products/prices to Ops, rotate schedule, error alerting). We fixed **how live progress is reported** to Ops (`POST /ingest/v1/status`) so each retailer’s lamp on **Scrape runs** stays accurate when two scrapers run in parallel.

**Customer-facing catalogue data:** Unchanged. Ingest still goes through `push_to_ops.py` → `/ingest/v1/products` and `/ingest/v1/prices`. The app does not read heartbeats.

### What `alerts.py` is responsible for

| Area | Changed in this work? |
|------|------------------------|
| **`report()`** — failures → local JSONL, Ops `/ingest/v1/alerts`, optional webhook | **No** |
| **`report_success()`** — cycle completed (local log only) | **No** |
| **`track_job()` / `clear_job()`** — track in-flight scrape/push per retailer | **Behaviour on Ops improved** (see below) |
| **`heartbeat_keepalive()`** — pulse every 5 min during long jobs | **Same schedule**; pulses now update the **correct** retailer row |
| **Scrape / push / rotate** (`run_pipeline_v2.py`, scrapers, `push_to_ops.py`) | **Not in this file** — unchanged |

Log line prefixes (`[checkers]`, `[woolworths]`, `[pnp]`) are changes in the **scraper scripts**, not in `alerts.py`. See [Container log prefixes](#container-log-prefixes-all-retailers) below.

### Why the `alerts.py` diff looks large

Git shows a refactor of **`set_status()`** plus two small helpers. Line count is similar; it is not a new pipeline feature.

| Change | Before | After |
|--------|--------|--------|
| **`_compose_status_message()`** | Built one combined string for all active jobs | **Removed** — each retailer gets its own message on its own row |
| **`_status_payload()` / `_post_status_body()`** | HTTP POST lived inside `set_status` | **Extracted** — same URL, headers, timeout; easier to read |
| **`set_status()` when `retailer=` is passed** | Could still update **`jobs[0]`’s row** with a combined message | Always updates **`scraper:{that retailer}`** |
| **`set_status()` with multiple jobs, no single retailer** | One POST, wrong or mixed `retailer` | POST `{ "scrapers": [ … ] }` — one object per active job (Ops already supported this) |
| **`clear_job("checkers")`** | Often **never** set Checkers to idle on Ops; refreshed another job instead | Always **`stage: idle`** for that retailer (`Finished checkers`) |
| **Comment on heartbeat rows** | Implied one row for the whole worker | Documents **one Ops row per retailer** (`scraper:pnp`, `scraper:checkers`, `scraper:woolworths`) |

### One sentence for management

*We fixed Ops dashboard telemetry for parallel scrapers; scraping, ingestion, and app catalogue behaviour are the same.*

---

## End-to-end flow (same before and after)

```mermaid
flowchart TB
  subgraph worker [baskit_worker]
    R[run_pipeline_v2 --rotate]
    S[Retailer scraper e.g. checkers_scraper.py]
    DB[(SQLite per retailer)]
    P[push_to_ops.py]
    A[alerts.py heartbeats]
    R --> S
    S --> DB
    R --> P
    P --> DB
    R --> A
    P --> A
  end

  subgraph ops [baskit_ops]
    ST[POST /ingest/v1/status]
    IN[POST /ingest/v1/products + prices]
    HB[(pipeline_heartbeat)]
    RUN[(ingest_runs)]
    ST --> HB
    IN --> RUN
  end

  subgraph app [baskit_app]
    API[GET /api/v1/catalogue/...]
  end

  A --> ST
  P --> IN
  API --> ops
```

**Example (one Checkers cycle):**

1. Worker: `track_job("checkers", "scrape", message="Scraping checkers catalogue")`
2. Worker runs `checkers_scraper.py scrape` → writes `data/checkers.db`
3. Worker: `track_job("checkers", "push", run_id=42, …)` → opens ingest run #42, sends batches
4. Worker: `clear_job("checkers")` → Ops should show Checkers **Ready**
5. App searches products that Ops already ingested — no direct worker call

---

## What Ops expects (same before and after)

Each retailer has its **own** heartbeat row:

| Retailer slug | Ops `host` key | Console label |
|---------------|----------------|---------------|
| `pnp` | `scraper:pnp` | Pick n Pay scraper |
| `checkers` | `scraper:checkers` | Checkers scraper |
| `woolworths` | `scraper:woolworths` | Woolworths scraper |

Ops also accepts a **batch** body so two scrapers can update in one request:

```json
{
  "scrapers": [
    { "retailer": "checkers", "stage": "scrape", "message": "Scraping checkers catalogue" },
    { "retailer": "woolworths", "stage": "push", "message": "Pushing 1200 products", "run_id": 43 }
  ]
}
```

Production often runs **two retailers at once** (`BASKIT_ROTATE_MAX_PARALLEL=2`, default retailers `checkers,woolworths`). Each lamp must update **its own** row.

---

## Before: one “lead” retailer stole the updates

### Idea (broken under parallel runs)

The worker kept a list of active jobs in memory, but when posting status it often:

1. Built one **combined message** (`checkers:scrape (...); woolworths:push (...)`).
2. Sent **one** `retailer` field taken from **`jobs[0]`** (whoever was registered first).
3. On `clear_job("checkers")`, refreshed only the remaining “lead” job — **did not** set Checkers to `idle`.

So pulses during Woolworths scrape could update **`scraper:checkers`**, and when Checkers finished its chip could stay **Busy** until it went stale (30+ minutes).

### Diagram — before (parallel Checkers + Woolworths)

```mermaid
sequenceDiagram
  participant W as Worker alerts.py
  participant O as Ops /ingest/v1/status
  participant CK as Row scraper:checkers
  participant WW as Row scraper:woolworths

  Note over W: Checkers starts first → jobs[0] = checkers
  W->>O: POST retailer=checkers, message="checkers:scrape..."
  O->>CK: Update ✓

  Note over W: Woolworths starts (parallel)
  W->>O: POST retailer=checkers, message="checkers:scrape...; woolworths:scrape..."
  Note over W: Still jobs[0] = checkers
  O->>CK: Update again (Woolworths text on Checkers row) ✗

  Note over W: Woolworths heartbeat pulse (retailer=woolworths)
  W->>O: POST retailer=checkers (lead), message=combined
  O->>CK: Update ✗
  Note over WW: Woolworths row may not update

  Note over W: Checkers finishes → clear_job(checkers)
  W->>O: POST retailer=woolworths only
  O->>WW: Update
  Note over CK: Still shows old Busy — never idle ✗
```

### Table — before vs expected behaviour

| Event | Expected on Ops board | What often happened before |
|--------|------------------------|----------------------------|
| Checkers scraping alone | Checkers **Busy · scrape** | OK |
| Woolworths starts while Checkers runs | Both **Busy**, each with own message | Checkers row got **mixed** messages; Woolworths lamp lagged |
| Woolworths keepalive every 5 min | Woolworths row `updated_at` moves | **Checkers** row updated instead |
| Checkers finishes | Checkers **Ready** | Checkers stayed **Busy** or stale |
| Ingest runs table | Checkers run #R-… closed | Usually still OK (push uses correct `retailer_id`) |

### Example — misleading console text (before)

**Single POST body (simplified):**

```json
{
  "host": "worker-container-abc",
  "retailer": "checkers",
  "stage": "scrape",
  "message": "checkers:scrape (Scraping checkers catalogue); woolworths:push (Pushing 800 products / 800 prices)"
}
```

Checkers lamp shows Woolworths push text. Woolworths lamp unchanged. Operator thinks “Checkers isn’t running.”

### Logs (before)

All scrapers printed the same prefix:

```text
[scrape] 12/240 'milk': +3 new ...
[scrape] 12/240 'milk': +3 new ...
```

In parallel, two streams looked identical — hard to grep for Checkers only.

---

## After: one POST per retailer (and batch when needed)

### Idea (matches Ops design)

| Rule | Behaviour |
|------|-----------|
| **`track_job(retailer, …)`** | Always `POST` with **`retailer` set to that slug** — updates `scraper:{slug}` only |
| **`heartbeat_keepalive(retailer=…)`** | Pulses the **same** retailer row every 5 minutes |
| **`clear_job(retailer)`** | `POST` **`stage: idle`** for **that** retailer (`Finished checkers`) |
| **Multiple jobs, no explicit retailer** | `POST` `{ "scrapers": [ … ] }` — one entry per active job |

### Diagram — after (parallel Checkers + Woolworths)

```mermaid
sequenceDiagram
  participant W as Worker alerts.py
  participant O as Ops /ingest/v1/status
  participant CK as Row scraper:checkers
  participant WW as Row scraper:woolworths

  W->>O: POST retailer=checkers, stage=scrape
  O->>CK: Update ✓

  W->>O: POST retailer=woolworths, stage=scrape
  O->>WW: Update ✓

  loop Every 5 min
    W->>O: POST retailer=checkers, stage=scrape
    O->>CK: Pulse ✓
    W->>O: POST retailer=woolworths, stage=scrape
    O->>WW: Pulse ✓
  end

  W->>O: POST retailer=checkers, stage=idle, message=Finished checkers
  O->>CK: Ready ✓
  Note over WW: Still Busy until Woolworths clear_job
```

### Table — before vs after (summary)

| Topic | Before | After |
|--------|--------|--------|
| Heartbeat `retailer` field | Often **first job in list**, not the job being reported | **Always the job** being updated |
| Combined status message | One string on **one** row | **Per-retailer** message on **each** row |
| Checkers finishes | Row often left **Busy** | Explicit **`idle`** on `scraper:checkers` |
| Parallel runs | Ops API batch supported but **unused** | Batch used when updating all jobs without a single `retailer` |
| Scraper stdout (scrape progress) | All used `[scrape] …` | Each retailer uses **`[checkers]`**, **`[woolworths]`**, **`[pnp]`** |
| CLI check | Single `status` object only | `status_check.py` also prints **scraper board** |

### Example — two correct POSTs (after)

**Checkers:**

```json
{
  "host": "worker-container-abc",
  "retailer": "checkers",
  "stage": "scrape",
  "message": "Scraping checkers catalogue"
}
```

**Woolworths (same minute):**

```json
{
  "host": "worker-container-abc",
  "retailer": "woolworths",
  "stage": "push",
  "message": "Pushing 800 products / 800 prices",
  "run_id": 43
}
```

**When Checkers completes:**

```json
{
  "host": "worker-container-abc",
  "retailer": "checkers",
  "stage": "idle",
  "message": "Finished checkers"
}
```

### Logs (after) — example

Orchestrator lines still use the label `scrape/{retailer}` (from `run_pipeline.py`). Scraper subprocesses use the retailer slug:

```text
2026-03-24 10:15:01 INFO [scrape/checkers] python checkers_scraper.py scrape ...
2026-03-24 10:15:02 INFO [checkers] 1/240 'milk': +3 new (saw 50/120), total_new=3
2026-03-24 10:20:01 INFO [status] stage=scrape retailer=woolworths Scraping woolworths catalogue
2026-03-24 10:20:02 INFO [woolworths] 1/180 'bread': +2 new (saw 40/80), total_new=2
2026-03-24 10:45:00 INFO [status] stage=idle retailer=checkers Finished checkers
```

---

## Container log prefixes (all retailers)

When `BASKIT_ROTATE_MAX_PARALLEL=2`, Docker/Dokploy merges stdout from two scrapers. Each retailer’s **scrape progress** lines now start with its slug so you can filter without guessing:

| Retailer | Script | Scrape progress prefix | Other prefixes (unchanged) |
|----------|--------|------------------------|----------------------------|
| Checkers | `checkers_scraper.py` | `[checkers]` | `[export]`, `[status]` |
| Woolworths | `woolworths_scraper.py` | `[woolworths]` | `[export]`, `[status]` |
| Pick n Pay | `pnp_scraper.py` | `[pnp]` | `[sitemap]` (seed step), `[export]`, `[status]` |

**Filter examples:**

```bash
# One retailer in merged worker logs
grep '\[checkers\]' 
grep '\[woolworths\]'
grep '\[pnp\]'

# Heartbeats from v2 orchestrator (all retailers)
grep '\[status\] retailer='
```

**Change log (documentation):**

| Date | Change |
|------|--------|
| 2026-02 | `alerts.py`: per-retailer Ops heartbeats + idle on `clear_job` |
| 2026-02 | `checkers_scraper.py`: `[scrape]` → `[checkers]` on scrape progress |
| 2026-10 | `woolworths_scraper.py`, `pnp_scraper.py`: `[scrape]` → `[woolworths]` / `[pnp]` on scrape progress (symmetric filtering) |

---

## Side-by-side: operator view on Scrape runs

| Lamp | Before (parallel run, Checkers done first) | After (same situation) |
|------|---------------------------------------------|-------------------------|
| **Checkers** | Busy · scrape · mixed message · stale possible | **Ready** · “Finished checkers” |
| **Woolworths** | Ready or stale (missed pulses) | **Busy · push** · run #R-43 |
| **Banner** (latest heartbeat) | One combined line, wrong `retailer` | Per-row chips are source of truth; banner is secondary |

---

## How to verify after deploy

1. Open Ops → **Scrape runs** — watch all three chips during `--rotate`.
2. On the worker (with `OPS_SERVICE_KEY` set):  
   `python v2/status_check.py`  
   Confirm a **Scraper board** section with `checkers` / `woolworths` lamps and fresh `updated_at`.
3. In worker logs, grep `[checkers]`, `[woolworths]`, or `[pnp]` (scrape lines) and `[status] retailer=` (Ops heartbeats).

---

## What did *not* change

- Scraper fetch/parse logic (Playwright + API for Checkers, etc.) — only **log prefixes** on scrape progress changed for all three retailers.
- Ingest contract (`open run` → `products:batch` → `prices:batch` → `close run`).
- Ops models, routes, or console UI layout.
- App catalogue API usage.