# Baskit pipeline v2 — Ops ingestion sink

Same scraping as the original pipeline, but results are **POSTed to the Baskit Ops
ingestion API** (`/ingest/v1/*` on `baskitops.appnotify.co.za`) instead of being
loaded straight into MySQL by `load_mysql.py`.

The scrapers (`pnp_scraper.py`, `checkers_scraper.py`, `woolworths_scraper.py`) and
the parent `config.py` are **unchanged**. v2 only swaps the sink: it reads each
retailer's existing SQLite `products` table and pushes it to the API.

## Layout

```
pipeline/v2/
  ops_config.py       - OPS_* env + reuse of parent config (DB paths, retailers)
  size_extract.py     - derive the required `size` field from product names
  ingest_client.py    - X-Service-Key HTTP client (runs/products/prices/close)
  push_to_ops.py      - read SQLite -> open/products/prices/close per retailer
  run_pipeline_v2.py  - orchestrator: scrape+push per retailer (--rotate loop)
  alerts.py           - error reporting (local log + Ops + optional webhook)
  status_check.py     - quick health check (runs + alerts)
  .env.example        - copy to .env and set OPS_SERVICE_KEY
```

## Ingestion contract (how the sink behaves)

Per retailer, one run: `open run -> products:batch -> prices:batch -> close run`.

- **products:batch** registers each retailer SKU with required `sku`, `name`/`raw_name`,
  `size`, plus optional scrape metadata:
  `barcode`, `brand`, `image_url`, `product_url`, `category_raw`, `unit_of_measure`,
  `manufacturer`, `average_rating`, `number_of_reviews`, `scraped_at`.
  New SKUs land **unmatched** until mapped (auto-match tries barcode first, then
  name+size). `size` is **required** — rows with no derivable size are skipped locally.
- **prices:batch** sends `sku`, `price`, `in_stock`, and optional `was_price`,
  `on_promotion`, `currency`. Observations are only stored for **matched** SKUs.
- **close run** returns the reconciliation accounting
  (`fetched = created + updated + unchanged + skipped + rejected`).

Retailer IDs (from the Ops `RetailerSeeder`): pnp=1, checkers/sixty60=2,
woolworths=3, spar=4. Override via `OPS_RETAILER_ID_*` if the server differs.

## Setup

```powershell
cd C:\wamp64\www\baskit_worker\pipeline\v2
copy .env.example .env
# edit .env -> set OPS_SERVICE_KEY
```

Uses the parent pipeline's virtualenv / `requirements.txt` (requests,
python-dotenv, schedule are already there).

## Run

```powershell
# Smoke test: build payloads only, no HTTP
python push_to_ops.py --dry-run --limit 20

# Push a small live batch for one retailer
python push_to_ops.py --retailers woolworths --limit 20

# Push all configured retailers (BASKIT_RETAILERS)
python push_to_ops.py

# Full cycle: scrape+push all retailers once
python run_pipeline_v2.py run

# Push only (skip scraping)
python run_pipeline_v2.py push

# Production: continuous rotate loop (Checkers -> Woolworths -> PnP -> ...)
# After 45 min, starts the next retailer in parallel (max 2 concurrent).
python run_pipeline_v2.py --rotate

# Abort open ingest runs stuck longer than 6 hours
python run_pipeline_v2.py abort-stale
python run_pipeline_v2.py abort-stale --older-than 2
```

## Parallel rotate

| Env | Default | Meaning |
|-----|---------|---------|
| `BASKIT_ROTATE_MAX_PARALLEL` | `2` | Cap concurrent scrape+push jobs |
| `BASKIT_ROTATE_START_AFTER_SEC` | `2700` (45 min) | Age of the oldest in-flight job before starting another |
| `BASKIT_ROTATE_COOLDOWN_SEC` | `60` | Pause when the queue is empty |

Long PnP scrapes no longer hold Checkers/Woolworths. Heartbeats pulse every 5 minutes during scrape/push so Ops does not mark the worker stale mid-job.

## Verify

```powershell
# Quick health: latest runs + alerts (local + Ops)
python status_check.py

# List recent runs with their accounting (needs the key)
curl -k -H "X-Service-Key: <key>" https://baskitops.appnotify.co.za/ingest/v1/runs

# List pipeline alerts stored in Ops
curl -k -H "X-Service-Key: <key>" https://baskitops.appnotify.co.za/ingest/v1/alerts
```

## Error alerts

Scrape/push failures are reported automatically:

| Where | What |
|-------|------|
| `data/pipeline_alerts.jsonl` | Local audit log on the worker |
| Ops console | **Scrape runs → Alerts** (`/admin/ingest/alerts`) |
| Email | Ops sends to `PIPELINE_ALERT_EMAIL` or `ADMIN_EMAIL` when SMTP is set |
| Webhook | `PIPELINE_ALERT_WEBHOOK_URL` on worker (default: cursor.testworks.co.za) |

Set `PIPELINE_ALERT_WEBHOOK_URL` in Dokploy for the worker. On Ops, set
`PIPELINE_ALERT_EMAIL` / `PIPELINE_ALERT_WEBHOOK_URL` and SMTP vars for email.

## Notes

- `OPS_VERIFY_TLS=false` disables cert verification — only needed if the prod TLS
  chain is flagged locally (e.g. Windows schannel). Prefer leaving it `true`.
- If the API returns `500`, the Ops backend (separate `baskit_ops` app) needs its
  ingestion migrations applied / a redeploy — the scraper side is unaffected.