fix: BLS-Datenimport als Hintergrundjob gegen HTTP 504
All checks were successful
Deploy Development / deploy (push) Successful in 1m10s
Build Test / pytest-backend (push) Successful in 5s
Build Test / lint-backend (push) Successful in 1s
Build Test / build-frontend (push) Successful in 24s

Prüfung und Apply laufen serverseitig, die UI pollt nur noch den Status.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Lars 2026-09-12 14:57:13 +02:00
parent 8e964509d7
commit a07a668de2
6 changed files with 281 additions and 35 deletions

171
backend/bls/jobs.py Normal file
View File

@ -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))

View File

@ -9,9 +9,12 @@ from pydantic import BaseModel
from auth import require_admin from auth import require_admin
from bls.import_service import upsert_attributes, upsert_foods 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 bls.parser import parse_components_xlsx, parse_foods_xlsx
from db import get_cursor, get_db, r2d from db import get_cursor, get_db, r2d
MAX_IMPORT_BYTES = 50 * 1024 * 1024
router = APIRouter(prefix="/api/admin/bls", tags=["admin", "bls"]) router = APIRouter(prefix="/api/admin/bls", tags=["admin", "bls"])
@ -111,6 +114,42 @@ async def import_foods(
return {"dry_run": False, **stats} 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") @router.get("/foods")
def admin_list_foods( def admin_list_foods(
q: Optional[str] = None, q: Optional[str] = None,

View File

@ -1,8 +1,10 @@
import time
from io import BytesIO from io import BytesIO
from openpyxl import Workbook from openpyxl import Workbook
from bls.import_service import should_persist_value 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 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": 1.2, "is_trace": False})
assert should_persist_value({"value_num": None, "is_trace": True}) assert should_persist_value({"value_num": None, "is_trace": True})
assert not should_persist_value({"value_num": None, "is_trace": False}) 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

View File

@ -21,7 +21,7 @@ MODULE_VERSIONS = {
"caliper": "1.0.1", "caliper": "1.0.1",
"activity": "1.2.1", # Legacy CSV import: activity_entries feature enforcement "activity": "1.2.1", # Legacy CSV import: activity_entries feature enforcement
"nutrition": "1.1.0", # BLS mapping, items, day marks, import policy "nutrition": "1.1.0", # BLS mapping, items, day marks, import policy
"bls": "1.0.0", "bls": "1.0.1",
"photos": "1.0.0", "photos": "1.0.0",
"insights": "1.3.0", "insights": "1.3.0",
"prompts": "1.1.0", "prompts": "1.1.0",
@ -44,6 +44,7 @@ CHANGELOG = [
"BLS 4.0 Stammdaten (dynamische Attribute, Upsert über bls_code)", "BLS 4.0 Stammdaten (dynamische Attribute, Upsert über bls_code)",
"Lernendes FDDB-Mapping ohne KI, änder- und löschbar", "Lernendes FDDB-Mapping ohne KI, änder- und löschbar",
"Optionale nutrition_items, Import-Policy, Fasten-/Lücken-Marken", "Optionale nutrition_items, Import-Policy, Fasten-/Lücken-Marken",
"BLS-Import als Hintergrundjob (kein Proxy-504)",
], ],
}, },
{ {

View File

@ -1,32 +1,44 @@
import { useEffect, useRef, useState } from 'react' import { useEffect, useRef, useState } from 'react'
import { api } from '../utils/api' import { api } from '../utils/api'
function summarizeCheck(kind, res) { function summarizeCheck(kind, check) {
if (!res) return '' if (!check) return ''
if (kind === 'components') return `${res.attributes ?? 0} Stoffe erkannt` if (kind === 'components') return `${check.attributes ?? 0} Stoffe erkannt`
return `${res.foods ?? 0} Lebensmittel, ${res.attribute_columns ?? 0} Stoffspalten` 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 }) { function FileImportBlock({ label, kind, onImported }) {
const inputRef = useRef(null) const inputRef = useRef(null)
const [file, setFile] = useState(null) const [file, setFile] = useState(null)
const [check, setCheck] = useState(null) const [job, setJob] = useState(null)
const [error, setError] = useState(null) const [error, setError] = useState(null)
const [busy, setBusy] = 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) => { const checkFile = async (nextFile) => {
setError(null) setError(null)
setCheck(null) setJob(null)
setApplyResult(null)
setImported(false)
if (!nextFile) return if (!nextFile) return
setBusy('check') setBusy('check')
try { 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) { } catch (e) {
setError(e.message) setError(e.message)
} finally { } finally {
@ -35,22 +47,24 @@ function FileImportBlock({ label, kind, onImported }) {
} }
const applyImport = async () => { const applyImport = async () => {
if (!file || !check) return if (!job?.id || job.status !== 'checked') return
setBusy('apply') setBusy('apply')
setError(null) setError(null)
try { try {
const res = await apiFn(file, false) await api.adminBlsImportApply(job.id)
setApplyResult(res) const done = await waitForJob(job.id, ['done'], setJob)
setImported(true) setJob(done)
onImported?.() onImported?.()
} catch (e) { } catch (e) {
setError(e.message) setError(e.message)
setImported(false)
} finally { } finally {
setBusy(null) setBusy(null)
} }
} }
const progress = job?.progress
const result = job?.result
return ( return (
<div className="settings-page__field"> <div className="settings-page__field">
<label className="settings-page__field-label">{label}</label> <label className="settings-page__field-label">{label}</label>
@ -74,26 +88,31 @@ function FileImportBlock({ label, kind, onImported }) {
> >
{file ? file.name : 'Datei auswählen'} {file ? file.name : 'Datei auswählen'}
</button> </button>
{busy && <p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}>{busy === 'apply' ? 'Importiere… das kann ein paar Minuten dauern.' : 'Prüfe…'}</p>} {busy === 'check' && <p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}>Prüfe</p>}
{error && <p style={{ color: 'var(--danger)', margin: 0 }}>{error}</p>} {busy === 'apply' && (
{check && !error && (
<p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}> <p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}>
Prüfung: {summarizeCheck(kind, check)} Importiere{progress?.total ? ` ${progress.current}/${progress.total}` : ''}
</p> </p>
)} )}
{applyResult && ( {error && <p style={{ color: 'var(--danger)', margin: 0 }}>{error}</p>}
{job?.status === 'checked' && !error && (
<p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}> <p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}>
{(applyResult.inserted ?? applyResult.foods_inserted) ?? 0} neu · {(applyResult.updated ?? applyResult.foods_updated) ?? 0} aktualisiert Prüfung: {summarizeCheck(kind, job.check)}
{applyResult.values_written != null ? ` · ${applyResult.values_written} Werte` : ''} </p>
)}
{job?.status === 'done' && result && (
<p style={{ fontSize: 13, color: 'var(--text2)', margin: 0 }}>
{(result.inserted ?? result.foods_inserted) ?? 0} neu · {(result.updated ?? result.foods_updated) ?? 0} aktualisiert
{result.values_written != null ? ` · ${result.values_written} Werte` : ''}
</p> </p>
)} )}
<button <button
type="button" type="button"
className="btn btn-primary btn-full" className="btn btn-primary btn-full"
disabled={!check || !!error || !!busy || imported} disabled={job?.status !== 'checked' || !!error || !!busy}
onClick={applyImport} onClick={applyImport}
> >
{imported ? 'Importiert' : 'Importieren'} {job?.status === 'done' ? 'Importiert' : 'Importieren'}
</button> </button>
</div> </div>
) )

View File

@ -251,16 +251,13 @@ export const api = {
upsertMyFoodMapping: (d) => req('/bls/mappings', json(d)), upsertMyFoodMapping: (d) => req('/bls/mappings', json(d)),
deleteMyFoodMapping: (id) => req(`/bls/mappings/${id}`, {method:'DELETE'}), deleteMyFoodMapping: (id) => req(`/bls/mappings/${id}`, {method:'DELETE'}),
adminBlsStatus: () => req('/admin/bls/status'), adminBlsStatus: () => req('/admin/bls/status'),
adminBlsImportComponents: async (file, dryRun=true) => { adminBlsImportStart: async (kind, file) => {
const fd=new FormData();fd.append('file',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()}) const r=await fetch(`${BASE}/admin/bls/import/jobs?kind=${encodeURIComponent(kind)}`,{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()})
return readJsonResponse(r) 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):''}`), adminBlsFoods: (q, kind) => req(`/admin/bls/foods?${q?('q='+encodeURIComponent(q)+'&'):''}${kind?('kind='+kind):''}`),
adminBlsFoodDetail: (id) => req(`/admin/bls/foods/${id}`), adminBlsFoodDetail: (id) => req(`/admin/bls/foods/${id}`),
adminCreateManualFood: (d) => req('/admin/bls/foods/manual', json(d)), adminCreateManualFood: (d) => req('/admin/bls/foods/manual', json(d)),