mitai-jinkendo/backend/bls/jobs.py

"""In-memory BLS import jobs so HTTP requests stay short (avoid proxy 504)."""
from __future__ import annotations

import threading
import time
import uuid
from typing import Any

from bls.import_service import upsert_attributes, upsert_foods
from bls.parser import parse_components_xlsx, parse_foods_xlsx
from db import get_cursor, get_db

FOOD_CHUNK = 300
JOB_TTL_S = 2 * 60 * 60
MAX_JOBS = 12

_lock = threading.Lock()
_jobs: dict[str, dict[str, Any]] = {}


def _public(job: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": job["id"],
        "kind": job["kind"],
        "status": job["status"],
        "check": job.get("check"),
        "progress": job.get("progress"),
        "result": job.get("result"),
        "error": job.get("error"),
    }


def _purge_locked(now: float) -> None:
    stale = [jid for jid, job in _jobs.items() if now - job["created_at"] > JOB_TTL_S]
    for jid in stale:
        _jobs.pop(jid, None)
    if len(_jobs) <= MAX_JOBS:
        return
    oldest = sorted(_jobs.values(), key=lambda j: j["created_at"])
    for job in oldest[: max(0, len(_jobs) - MAX_JOBS)]:
        if job["status"] in ("checking", "applying"):
            continue
        _jobs.pop(job["id"], None)


def get_job(job_id: str) -> dict[str, Any] | None:
    with _lock:
        job = _jobs.get(job_id)
        return _public(job) if job else None


def create_and_check(kind: str, raw: bytes) -> str:
    if kind not in ("components", "foods"):
        raise ValueError("kind muss components oder foods sein")
    job_id = str(uuid.uuid4())
    now = time.time()
    with _lock:
        _purge_locked(now)
        _jobs[job_id] = {
            "id": job_id,
            "kind": kind,
            "status": "checking",
            "created_at": now,
            "raw": raw,
            "parsed": None,
            "check": None,
            "progress": None,
            "result": None,
            "error": None,
        }
    threading.Thread(target=_run_check, args=(job_id,), daemon=True).start()
    return job_id


def start_apply(job_id: str) -> None:
    with _lock:
        job = _jobs.get(job_id)
        if job is None:
            raise KeyError(job_id)
        if job["status"] != "checked":
            raise ValueError("Zuerst die Prüfung abwarten")
        if job.get("parsed") is None:
            raise ValueError("Geparste Datei nicht mehr vorhanden — Datei neu wählen")
        job["status"] = "applying"
        job["error"] = None
        job["progress"] = {"current": 0, "total": 0}
    threading.Thread(target=_run_apply, args=(job_id,), daemon=True).start()


def _update(job_id: str, **fields: Any) -> None:
    with _lock:
        job = _jobs.get(job_id)
        if not job:
            return
        job.update(fields)


def _run_check(job_id: str) -> None:
    with _lock:
        job = _jobs.get(job_id)
        if not job:
            return
        kind = job["kind"]
        raw = job["raw"]
    try:
        if kind == "components":
            attrs = parse_components_xlsx(raw)
            _update(
                job_id,
                parsed=attrs,
                raw=None,
                check={"attributes": len(attrs)},
                status="checked",
            )
            return
        parsed = parse_foods_xlsx(raw)
        _update(
            job_id,
            parsed=parsed,
            raw=None,
            check={
                "foods": len(parsed["foods"]),
                "attribute_columns": len(parsed["attribute_headers"]),
            },
            status="checked",
        )
    except Exception as e:
        _update(job_id, status="error", error=str(e), raw=None, parsed=None)


def _run_apply(job_id: str) -> None:
    with _lock:
        job = _jobs.get(job_id)
        if not job:
            return
        kind = job["kind"]
        parsed = job["parsed"]
    try:
        if kind == "components":
            with get_db() as conn:
                stats = upsert_attributes(get_cursor(conn), parsed)
            _update(job_id, status="done", result=stats, parsed=None, progress={"current": stats.get("total", 0), "total": stats.get("total", 0)})
            return

        foods = parsed["foods"]
        total = len(foods)
        inserted = updated = values_written = 0
        _update(job_id, progress={"current": 0, "total": total})
        for i in range(0, total, FOOD_CHUNK):
            chunk = foods[i : i + FOOD_CHUNK]
            with get_db() as conn:
                stats = upsert_foods(get_cursor(conn), chunk)
            inserted += stats.get("inserted", 0)
            updated += stats.get("updated", 0)
            values_written += stats.get("values_written", 0)
            _update(job_id, progress={"current": min(i + FOOD_CHUNK, total), "total": total})
        _update(
            job_id,
            status="done",
            parsed=None,
            result={
                "inserted": inserted,
                "updated": updated,
                "foods_inserted": inserted,
                "foods_updated": updated,
                "values_written": values_written,
                "foods_total": total,
            },
        )
    except Exception as e:
        _update(job_id, status="error", error=str(e))