"""
Password-protected catalogue viewer — same MySQL target as load_mysql.py.
python view_server.py
Login via HTTP Basic Auth (BASKIT_VIEW_USER / BASKIT_VIEW_PASSWORD).
"""
from __future__ import annotations
import html
import math
from decimal import Decimal
from functools import wraps
from urllib.parse import urlencode
import pymysql
from flask import Flask, Response, jsonify, request
import config
app = Flask(__name__)
RETAILERS = [("pnp", "Pick n Pay"), ("checkers", "Checkers"), ("woolworths", "Woolworths")]
RETAILER_LABELS = dict(RETAILERS)
PER_PAGE = 60
def _auth_ok(user: str, password: str) -> bool:
return user == config.VIEW_USER and password == config.VIEW_PASSWORD
def requires_auth(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
auth = request.authorization
if not auth or not _auth_ok(auth.username, auth.password):
return Response(
"Login required",
401,
{"WWW-Authenticate": 'Basic realm="Baskit Pipeline Viewer"'},
)
return fn(*args, **kwargs)
return wrapper
def _conn():
return pymysql.connect(**config.MYSQL, cursorclass=pymysql.cursors.DictCursor)
def _h(s) -> str:
return html.escape("" if s is None else str(s))
def _money(v) -> str:
return "—" if v is None else f"R{float(v):,.2f}"
def _json_val(v):
"""Make MySQL row values JSON-serialisable."""
if v is None:
return None
if isinstance(v, Decimal):
return float(v)
return v
@app.get("/api/product")
@requires_auth
def product_detail():
"""Lazy-load per-retailer catalogue rows for one barcode (admin drill-down)."""
barcode = (request.args.get("barcode") or "").strip()
if not barcode:
return jsonify({"error": "barcode required"}), 400
with _conn() as conn, conn.cursor() as cur:
cur.execute("SHOW COLUMNS FROM catalogue LIKE 'scraped_at'")
has_scraped = cur.fetchone() is not None
cols = (
"retailer, product_key, barcode_raw, barcode_norm, is_instore_bc, "
"name, brand, price, was_price, image_url, url, bucket"
+ (", scraped_at" if has_scraped else "")
)
cur.execute(
f"SELECT {cols} FROM catalogue WHERE barcode_norm = %s OR barcode_raw = %s "
"ORDER BY retailer",
(barcode, barcode),
)
rows = cur.fetchall()
cur.execute("SELECT * FROM mvp_catalogue WHERE barcode = %s", (barcode,))
mvp = cur.fetchone()
by_retailer = {key: None for key, _ in RETAILERS}
for row in rows:
by_retailer[row["retailer"]] = {k: _json_val(v) for k, v in row.items()}
return jsonify({
"barcode": barcode,
"mvp": {k: _json_val(v) for k, v in mvp.items()} if mvp else None,
"retailers": by_retailer,
"labels": RETAILER_LABELS,
})
def _qs(overrides: dict | None = None) -> str:
base = {
"mode": request.args.get("mode", "mvp"),
"q": request.args.get("q", ""),
"bucket": request.args.get("bucket", ""),
"page": request.args.get("page", "1"),
}
if overrides:
base.update(overrides)
return urlencode(base)
@app.get("/health")
def health():
"""Unauthenticated liveness check."""
return {"ok": True}
@app.get("/")
@requires_auth
def index():
mode = "all" if request.args.get("mode") == "all" else "mvp"
q = (request.args.get("q") or "").strip()
bucket = (request.args.get("bucket") or "").strip()
page = max(1, int(request.args.get("page") or 1))
offset = (page - 1) * PER_PAGE
with _conn() as conn, conn.cursor() as cur:
cur.execute(
"SELECT (SELECT COUNT(*) FROM catalogue) AS catalogue_rows,"
" (SELECT COUNT(*) FROM mvp_catalogue) AS mvp_rows,"
" (SELECT COUNT(*) FROM mvp_catalogue WHERE n_retailers>=3) AS mvp_all3"
)
stats = cur.fetchone()
cur.execute(
"SELECT retailer, COUNT(*) c FROM catalogue GROUP BY retailer ORDER BY retailer"
)
by_retailer = {r["retailer"]: r["c"] for r in cur.fetchall()}
cur.execute("SELECT bucket, COUNT(*) c FROM mvp_catalogue GROUP BY bucket ORDER BY bucket")
buckets = cur.fetchall()
rows: list[dict] = []
total_rows = 0
if mode == "mvp":
where, args = [], {}
if q:
where.append(
"(CONVERT(m.name USING utf8mb4) LIKE %(q)s OR CONVERT(m.brand USING utf8mb4) LIKE %(q)s "
"OR CONVERT(m.barcode USING utf8mb4) LIKE %(q)s)"
)
args["q"] = f"%{q}%"
if bucket:
where.append("CONVERT(m.bucket USING utf8mb4) = %(bucket)s")
args["bucket"] = bucket
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
cur.execute(f"SELECT COUNT(*) c FROM mvp_catalogue m {where_sql}", args)
total_rows = int(cur.fetchone()["c"])
cur.execute(
f"SELECT m.*, img.image_url FROM mvp_catalogue m LEFT JOIN ("
f" SELECT barcode_norm, MIN(image_url) AS image_url FROM catalogue"
f" WHERE image_url IS NOT NULL AND image_url <> '' GROUP BY barcode_norm"
f") img ON CONVERT(img.barcode_norm USING utf8mb4) = CONVERT(m.barcode USING utf8mb4)"
f" {where_sql} ORDER BY m.bucket, m.n_retailers DESC, m.name"
f" LIMIT {PER_PAGE} OFFSET {offset}",
args,
)
rows = cur.fetchall()
else:
where = ["barcode_norm IS NOT NULL", "is_instore_bc = 0", "price IS NOT NULL"]
args: dict = {}
if q:
where.append(
"(CONVERT(name USING utf8mb4) LIKE %(q)s OR CONVERT(brand USING utf8mb4) LIKE %(q)s "
"OR CONVERT(barcode_norm USING utf8mb4) LIKE %(q)s)"
)
args["q"] = f"%{q}%"
where_sql = "WHERE " + " AND ".join(where)
having = "HAVING COUNT(DISTINCT retailer) >= 2"
cur.execute(
f"SELECT COUNT(*) c FROM (SELECT barcode_norm FROM catalogue {where_sql}"
f" GROUP BY barcode_norm {having}) t",
args,
)
total_rows = int(cur.fetchone()["c"])
cur.execute(
f"SELECT barcode_norm AS barcode, MAX(name) AS name, MAX(brand) AS brand,"
f" COUNT(DISTINCT retailer) AS n_retailers,"
f" MAX(CASE WHEN retailer='pnp' THEN price END) AS pnp_price,"
f" MAX(CASE WHEN retailer='checkers' THEN price END) AS checkers_price,"
f" MAX(CASE WHEN retailer='woolworths' THEN price END) AS woolworths_price,"
f" MIN(price) AS min_price, MAX(price) AS max_price,"
f" (SELECT image_url FROM catalogue c2 WHERE c2.barcode_norm = catalogue.barcode_norm"
f" AND c2.image_url IS NOT NULL LIMIT 1) AS image_url"
f" FROM catalogue {where_sql} GROUP BY barcode_norm {having}"
f" ORDER BY (MAX(price)-MIN(price)) DESC LIMIT {PER_PAGE} OFFSET {offset}",
args,
)
rows = cur.fetchall()
total_pages = max(1, math.ceil(total_rows / PER_PAGE))
mysql_label = f"{config.MYSQL['user']}@{config.MYSQL['host']}:{config.MYSQL['port']}/{config.MYSQL['database']}"
return _render(mode, q, bucket, page, total_pages, total_rows, stats, buckets, by_retailer, rows, mysql_label)
def _render(mode, q, bucket, page, total_pages, total_rows, stats, buckets, by_retailer, rows, mysql_label):
pills = ""
if mode == "mvp":
pills += f'All buckets'
for b in buckets:
on = "on" if bucket == b["bucket"] else ""
label = _h(b["bucket"].replace("_", " ").title())
pills += f'{label} {b["c"]}'
body_rows = ""
if not rows:
body_rows = '| No products found. |
'
for r in rows:
prices = {"pnp": r.get("pnp_price"), "checkers": r.get("checkers_price"), "woolworths": r.get("woolworths_price")}
valid = {k: v for k, v in prices.items() if v is not None}
mn = min(valid.values()) if valid else None
mx = max(valid.values()) if valid else None
save = (mx - mn) if mn is not None and mx is not None else 0
save_pct = round(save / mx * 100) if mx and save > 0 else 0
cheapest = next((k for k, v in prices.items() if v is not None and float(v) == float(mn)), None) if mn else None
img = f'
' if r.get("image_url") else ""
bk = f'{_h(str(r.get("bucket","")).replace("_"," "))} ' if r.get("bucket") else ""
brand = f'{_h(r.get("brand"))} · ' if r.get("brand") else ""
price_cells = ""
for key, label in RETAILERS:
v = prices[key]
cheap = v is not None and cheapest == key and len(valid) > 1
cls = "price cheap" if cheap else ("price none" if v is None else "price")
price_cells += f'{_money(v)} | '
save_cell = f'{_money(save)}{save_pct}%' if save > 0 else "—"
cheap_label = dict(RETAILERS).get(cheapest, "—") if cheapest else "—"
body_rows += (
f""
f"| {img} | {_h(r.get('name'))} "
f"{bk}{brand}barcode {_h(r.get('barcode'))} | {price_cells}"
f"{_h(cheap_label)} | {save_cell} |
"
)
retailer_stats = " · ".join(f"{_h(l)}: {by_retailer.get(k, 0):,}" for k, l in RETAILERS)
return f"""
Baskit Pipeline Viewer
🧺 Baskit
pipeline viewer · {_h(mysql_label)}
{stats['catalogue_rows']:,} catalogue rows
{stats['mvp_rows']:,} MVP SKUs
{stats['mvp_all3']:,} priced in all 3
{retailer_stats}
{pills}
{total_rows:,} products · page {page} of {total_pages}
| Product |
{"".join(f'{_h(l)} | ' for _, l in RETAILERS)}
Cheapest | You save |
|---|
{body_rows}
MVP summary · '+esc(m.bucket)+' · '
+m.n_retailers+' retailers · PnP '+money(m.pnp_price)+' · Checkers '+money(m.checkers_price)
+' · Woolworths '+money(m.woolworths_price)+'
';
}
const labels=data.labels||{};
for(const key of ['pnp','checkers','woolworths']){
const r=data.retailers[key];
const name=labels[key]||key;
if(!r){
h+=''+esc(name)+'
Not in catalogue
+')
':'';
h+=''+esc(name)+'
'
+img+'
'
+field('Store product ID',r.product_key)
+field('Barcode (raw)',r.barcode_raw)
+field('Barcode (norm)',r.barcode_norm)
+field('Name',r.name)
+field('Brand',r.brand)
+field('Price',money(r.price))
+field('Was price',money(r.was_price))
+field('Bucket',r.bucket)
+field('In-store barcode',r.is_instore_bc?'yes':'no')
+field('Scraped at',fmtDate(r.scraped_at))
+'
';
if(r.url) h+='
View on store ↗
';
h+='
Loading…
';
modal.hidden=false; modal.setAttribute('aria-hidden','false');
document.body.style.overflow='hidden';
if(cache[barcode]){body.innerHTML=render(cache[barcode]);return;}
fetch('/api/product?barcode='+encodeURIComponent(barcode),{credentials:'same-origin'})
.then(r=>{if(!r.ok)throw new Error('HTTP '+r.status);return r.json();})
.then(d=>{cache[barcode]=d;body.innerHTML=render(d);})
.catch(e=>{body.innerHTML='Failed: '+esc(e.message)+'
';});
}
function close(){modal.hidden=true;modal.setAttribute('aria-hidden','true');document.body.style.overflow='';}
document.querySelector('tbody').addEventListener('click',e=>{
const row=e.target.closest('tr.prow'); if(row) open(row.dataset.barcode);
});
document.querySelector('tbody').addEventListener('keydown',e=>{
if(e.key!=='Enter'&&e.key!==' ')return;
const row=e.target.closest('tr.prow'); if(row){e.preventDefault();open(row.dataset.barcode);}
});
modal.addEventListener('click',e=>{if(e.target.dataset.close!==undefined)close();});
document.addEventListener('keydown',e=>{if(e.key==='Escape'&&!modal.hidden)close();});
})();
"""
if __name__ == "__main__":
app.run(host="0.0.0.0", port=config.VIEW_PORT, debug=False)