diff --git a/backend/bls/jobs.py b/backend/bls/jobs.py new file mode 100644 index 0000000..286a760 --- /dev/null +++ b/backend/bls/jobs.py @@ -0,0 +1,171 @@ +"""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)) diff --git a/backend/routers/admin_bls.py b/backend/routers/admin_bls.py index d7479bf..3266e3e 100644 --- a/backend/routers/admin_bls.py +++ b/backend/routers/admin_bls.py @@ -9,9 +9,12 @@ from pydantic import BaseModel from auth import require_admin from bls.import_service import upsert_attributes, upsert_foods +from bls.jobs import create_and_check, get_job, start_apply from bls.parser import parse_components_xlsx, parse_foods_xlsx from db import get_cursor, get_db, r2d +MAX_IMPORT_BYTES = 50 * 1024 * 1024 + router = APIRouter(prefix="/api/admin/bls", tags=["admin", "bls"]) @@ -111,6 +114,42 @@ async def import_foods( return {"dry_run": False, **stats} +@router.post("/import/jobs") +async def start_import_job( + kind: str, + file: UploadFile = File(...), + session: dict = Depends(require_admin), +): + if kind not in ("components", "foods"): + raise HTTPException(400, "kind muss components oder foods sein") + raw = await file.read() + if not raw: + raise HTTPException(400, "Leere Datei") + if len(raw) > MAX_IMPORT_BYTES: + raise HTTPException(400, "Datei größer als 50 MB") + job_id = create_and_check(kind, raw) + return {"id": job_id, "status": "checking"} + + +@router.get("/import/jobs/{job_id}") +def import_job_status(job_id: str, session: dict = Depends(require_admin)): + job = get_job(job_id) + if not job: + raise HTTPException(404, "Import-Job nicht gefunden") + return job + + +@router.post("/import/jobs/{job_id}/apply") +def apply_import_job(job_id: str, session: dict = Depends(require_admin)): + try: + start_apply(job_id) + except KeyError: + raise HTTPException(404, "Import-Job nicht gefunden") from None + except ValueError as e: + raise HTTPException(409, str(e)) from e + return get_job(job_id) + + @router.get("/foods") def admin_list_foods( q: Optional[str] = None, diff --git a/backend/tests/test_bls_parser.py b/backend/tests/test_bls_parser.py index c7a2e27..d399c33 100644 --- a/backend/tests/test_bls_parser.py +++ b/backend/tests/test_bls_parser.py @@ -1,8 +1,10 @@ +import time from io import BytesIO from openpyxl import Workbook from bls.import_service import should_persist_value +from bls.jobs import create_and_check, get_job from bls.parser import parse_components_xlsx, parse_foods_xlsx @@ -44,3 +46,20 @@ def test_persist_only_numeric_or_trace(): assert should_persist_value({"value_num": 1.2, "is_trace": False}) assert should_persist_value({"value_num": None, "is_trace": True}) assert not should_persist_value({"value_num": None, "is_trace": False}) + + +def test_import_job_check_does_not_need_http(): + data = _xlsx([ + ["BLS Code", "Name", "Food name", "ENERCC Energie [kcal/100g]", "ENERCC Herkunft", "ENERCC Referenz"], + ["C131000", "Hafer roh", "Oats raw", 350, "Analyse", "MRI"], + ]) + job_id = create_and_check("foods", data) + job = None + for _ in range(80): + job = get_job(job_id) + if job and job["status"] in ("checked", "error"): + break + time.sleep(0.05) + assert job is not None + assert job["status"] == "checked", job.get("error") + assert job["check"]["foods"] == 1 diff --git a/backend/version.py b/backend/version.py index 05b2add..487ab0d 100644 --- a/backend/version.py +++ b/backend/version.py @@ -21,7 +21,7 @@ MODULE_VERSIONS = { "caliper": "1.0.1", "activity": "1.2.1", # Legacy CSV import: activity_entries feature enforcement "nutrition": "1.1.0", # BLS mapping, items, day marks, import policy - "bls": "1.0.0", + "bls": "1.0.1", "photos": "1.0.0", "insights": "1.3.0", "prompts": "1.1.0", @@ -44,6 +44,7 @@ CHANGELOG = [ "BLS 4.0 Stammdaten (dynamische Attribute, Upsert über bls_code)", "Lernendes FDDB-Mapping ohne KI, änder- und löschbar", "Optionale nutrition_items, Import-Policy, Fasten-/Lücken-Marken", + "BLS-Import als Hintergrundjob (kein Proxy-504)", ], }, { diff --git a/frontend/src/pages/AdminBlsImportPage.jsx b/frontend/src/pages/AdminBlsImportPage.jsx index 0a2e06e..851174e 100644 --- a/frontend/src/pages/AdminBlsImportPage.jsx +++ b/frontend/src/pages/AdminBlsImportPage.jsx @@ -1,32 +1,44 @@ import { useEffect, useRef, useState } from 'react' import { api } from '../utils/api' -function summarizeCheck(kind, res) { - if (!res) return '' - if (kind === 'components') return `${res.attributes ?? 0} Stoffe erkannt` - return `${res.foods ?? 0} Lebensmittel, ${res.attribute_columns ?? 0} Stoffspalten` +function summarizeCheck(kind, check) { + if (!check) return '' + if (kind === 'components') return `${check.attributes ?? 0} Stoffe erkannt` + return `${check.foods ?? 0} Lebensmittel, ${check.attribute_columns ?? 0} Stoffspalten` +} + +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +async function waitForJob(jobId, doneStatuses, onTick) { + for (;;) { + const job = await api.adminBlsImportJob(jobId) + onTick?.(job) + if (job.status === 'error') { + throw new Error(job.error || 'Import fehlgeschlagen') + } + if (doneStatuses.includes(job.status)) return job + await sleep(1200) + } } function FileImportBlock({ label, kind, onImported }) { const inputRef = useRef(null) const [file, setFile] = useState(null) - const [check, setCheck] = useState(null) + const [job, setJob] = useState(null) const [error, setError] = useState(null) const [busy, setBusy] = useState(null) - const [imported, setImported] = useState(false) - const [applyResult, setApplyResult] = useState(null) - - const apiFn = kind === 'components' ? api.adminBlsImportComponents : api.adminBlsImportFoods const checkFile = async (nextFile) => { setError(null) - setCheck(null) - setApplyResult(null) - setImported(false) + setJob(null) if (!nextFile) return setBusy('check') try { - setCheck(await apiFn(nextFile, true)) + const started = await api.adminBlsImportStart(kind, nextFile) + const done = await waitForJob(started.id, ['checked'], setJob) + setJob(done) } catch (e) { setError(e.message) } finally { @@ -35,22 +47,24 @@ function FileImportBlock({ label, kind, onImported }) { } const applyImport = async () => { - if (!file || !check) return + if (!job?.id || job.status !== 'checked') return setBusy('apply') setError(null) try { - const res = await apiFn(file, false) - setApplyResult(res) - setImported(true) + await api.adminBlsImportApply(job.id) + const done = await waitForJob(job.id, ['done'], setJob) + setJob(done) onImported?.() } catch (e) { setError(e.message) - setImported(false) } finally { setBusy(null) } } + const progress = job?.progress + const result = job?.result + return (
@@ -74,26 +88,31 @@ function FileImportBlock({ label, kind, onImported }) { > {file ? file.name : 'Datei auswählen'} - {busy &&

{busy === 'apply' ? 'Importiere… das kann ein paar Minuten dauern.' : 'Prüfe…'}

} - {error &&

{error}

} - {check && !error && ( + {busy === 'check' &&

Prüfe…

} + {busy === 'apply' && (

- Prüfung: {summarizeCheck(kind, check)} + Importiere…{progress?.total ? ` ${progress.current}/${progress.total}` : ''}

)} - {applyResult && ( + {error &&

{error}

} + {job?.status === 'checked' && !error && (

- {(applyResult.inserted ?? applyResult.foods_inserted) ?? 0} neu · {(applyResult.updated ?? applyResult.foods_updated) ?? 0} aktualisiert - {applyResult.values_written != null ? ` · ${applyResult.values_written} Werte` : ''} + Prüfung: {summarizeCheck(kind, job.check)} +

+ )} + {job?.status === 'done' && result && ( +

+ {(result.inserted ?? result.foods_inserted) ?? 0} neu · {(result.updated ?? result.foods_updated) ?? 0} aktualisiert + {result.values_written != null ? ` · ${result.values_written} Werte` : ''}

)}
) diff --git a/frontend/src/utils/api.js b/frontend/src/utils/api.js index 9a3276f..a6d1d32 100644 --- a/frontend/src/utils/api.js +++ b/frontend/src/utils/api.js @@ -251,16 +251,13 @@ export const api = { upsertMyFoodMapping: (d) => req('/bls/mappings', json(d)), deleteMyFoodMapping: (id) => req(`/bls/mappings/${id}`, {method:'DELETE'}), adminBlsStatus: () => req('/admin/bls/status'), - adminBlsImportComponents: async (file, dryRun=true) => { + adminBlsImportStart: async (kind, file) => { const fd=new FormData();fd.append('file',file) - const r=await fetch(`${BASE}/admin/bls/import/components?dry_run=${dryRun}`,{method:'POST',body:fd,headers:hdrs()}) - return readJsonResponse(r) - }, - adminBlsImportFoods: async (file, dryRun=true) => { - const fd=new FormData();fd.append('file',file) - const r=await fetch(`${BASE}/admin/bls/import/foods?dry_run=${dryRun}`,{method:'POST',body:fd,headers:hdrs()}) + const r=await fetch(`${BASE}/admin/bls/import/jobs?kind=${encodeURIComponent(kind)}`,{method:'POST',body:fd,headers:hdrs()}) return readJsonResponse(r) }, + adminBlsImportJob: (id) => req(`/admin/bls/import/jobs/${id}`), + adminBlsImportApply: (id) => req(`/admin/bls/import/jobs/${id}/apply`, {method:'POST'}), adminBlsFoods: (q, kind) => req(`/admin/bls/foods?${q?('q='+encodeURIComponent(q)+'&'):''}${kind?('kind='+kind):''}`), adminBlsFoodDetail: (id) => req(`/admin/bls/foods/${id}`), adminCreateManualFood: (d) => req('/admin/bls/foods/manual', json(d)),