#!/usr/bin/env python3 """ Crystal worker — polls MySQL for PENDING runs, claims one atomically, executes the estimation engine, streams progress into the run row, writes outputs. No web framework. MySQL is the only interface between the PHP dashboard and this process. Run under systemd (see crystal-worker.service). Engine integration point: run_estimation() below. Today it calls the demo engine (engine_stub) so the full loop works end to end; Steps 0-7 of the rebuild replace that one import with the real engine package. """ from __future__ import annotations import json import socket import time import traceback from pathlib import Path import pymysql # ---------------------------------------------------------------- config BASE = Path(__file__).resolve().parent.parent CONFIG = json.loads((BASE / "worker" / "config.json").read_text()) UPLOADS = Path(CONFIG["uploads_dir"]) EXPORTS = Path(CONFIG["exports_dir"]) POLL_SECONDS = float(CONFIG.get("poll_seconds", 2.0)) WORKER_ID = f"{socket.gethostname()}:{CONFIG.get('worker_name', 'w1')}" def db(): return pymysql.connect( host=CONFIG["db_host"], user=CONFIG["db_user"], password=CONFIG["db_pass"], database=CONFIG["db_name"], charset="utf8mb4", autocommit=True, cursorclass=pymysql.cursors.DictCursor, ) # ------------------------------------------------------------- claim/update def claim_next(conn) -> dict | None: """Atomically claim the oldest PENDING run (safe with multiple workers).""" with conn.cursor() as c: c.execute( "UPDATE estimate_runs SET status='RUNNING', claimed_by=%s, " "started_at=NOW(), stage_label='Getting started', progress_pct=1 " "WHERE status='PENDING' ORDER BY created_at LIMIT 1", (WORKER_ID,), ) if c.rowcount == 0: return None c.execute( "SELECT * FROM estimate_runs WHERE claimed_by=%s AND status='RUNNING' " "ORDER BY started_at DESC LIMIT 1", (WORKER_ID,), ) return c.fetchone() def progress(conn, run_id: str, pct: int, stage: str, detail: str = "") -> None: with conn.cursor() as c: c.execute( "UPDATE estimate_runs SET progress_pct=%s, stage_label=%s, detail_line=%s " "WHERE run_id=%s", (max(1, min(99, pct)), stage[:120], detail[:255], run_id), ) def finish(conn, run_id: str, *, rows_total: int, rows_estimated: int, rows_flagged: int, summary: Path, detailed: Path) -> None: status = "NEEDS_REVIEW" if rows_flagged > 0 else "COMPLETED" with conn.cursor() as c: c.execute( "UPDATE estimate_runs SET status=%s, progress_pct=100, " "stage_label='All done', detail_line='', rows_total=%s, rows_estimated=%s, " "rows_flagged=%s, summary_path=%s, detailed_path=%s, finished_at=NOW() " "WHERE run_id=%s", (status, rows_total, rows_estimated, rows_flagged, str(summary), str(detailed), run_id), ) def fail(conn, run_id: str, message: str) -> None: with conn.cursor() as c: c.execute( "UPDATE estimate_runs SET status='FAILED', error_message=%s, " "finished_at=NOW() WHERE run_id=%s", (message[:500], run_id), ) # ------------------------------------------------------------ engine hook def run_estimation(conn, run: dict) -> None: """THE single integration point. Replace `engine_stub` with the real engine package as the rebuild lands: from crystal_engine import run_pipeline result = run_pipeline(upload, exports_dir=EXPORTS, on_progress=lambda p, s, d: progress(...)) The contract stays identical: read the upload, call progress() as stages advance, write summary/detailed workbooks + _flags.json to EXPORTS, then call finish(). """ import sys sys.path.insert(0, str(BASE)) _vendor = BASE / "vendor" if _vendor.is_dir() and str(_vendor) not in sys.path: sys.path.insert(0, str(_vendor)) # Drop the engine from the module cache before every run so a long-lived # worker always executes the code currently on disk. Without this the worker # keeps serving whatever it imported at boot, and an engine fix appears to # have had no effect until someone remembers to restart the service -- # which looks exactly like the fix not working. for _name in [n for n in list(sys.modules) if n == "engine" or n.startswith("engine.")]: del sys.modules[_name] cfg_path = BASE / "worker" / "config.json" reference_ready = cfg_path.exists() and ( json.loads(cfg_path.read_text()).get("reference_mode") in ("fixtures", "mysql")) if reference_ready: from engine.pipeline import run_pipeline as engine_run # the real engine else: from engine_stub import run_pipeline as engine_run # demo until configured run_id = run["run_id"] upload = Path(run["upload_path"]) kwargs = {"config_path": cfg_path} if reference_ready else {} result = engine_run( upload_path=upload, exports_dir=EXPORTS, run_id=run_id, on_progress=lambda pct, stage, detail="": progress(conn, run_id, pct, stage, detail), **kwargs, ) finish(conn, run_id, rows_total=result["rows_total"], rows_estimated=result["rows_estimated"], rows_flagged=result["rows_flagged"], summary=result["summary_path"], detailed=result["detailed_path"]) # ------------------------------------------------------------------- loop def main() -> None: print(f"[crystal-worker] {WORKER_ID} started; polling every {POLL_SECONDS}s") EXPORTS.mkdir(parents=True, exist_ok=True) while True: try: conn = db() run = claim_next(conn) if run is None: conn.close() time.sleep(POLL_SECONDS) continue print(f"[crystal-worker] claimed {run['run_id']} ({run['original_filename']})") try: run_estimation(conn, run) print(f"[crystal-worker] finished {run['run_id']}") except Exception as exc: # noqa: BLE001 — a run failure must not kill the worker traceback.print_exc() fail(conn, run["run_id"], "The estimator hit an unexpected problem. Details were logged for your administrator.") conn.close() except Exception: # DB outage etc. — back off and keep living traceback.print_exc() time.sleep(10) if __name__ == "__main__": main()