refactor: extract atomic JSON helpers to io_utils

Centralize read_json/write_json_atomic used by pipeline meta I/O and
worker processed.json bookkeeping. Preserves corrupt-file warning in
load_processed().

Co-authored-by: Cursor <cursoragent@cursor.com>

app/io_utils.py

@@ -0,0 +1,25 @@
+"""Small shared I/O helpers."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+def read_json(path: Path, *, default: Any = None) -> Any:
+    """Read JSON from *path*, returning *default* when missing or corrupt."""
+    if not path.exists():
+        return default
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return default
+
+
+def write_json_atomic(path: Path, data: Any) -> None:
+    """Write JSON via a temp file and atomic replace."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+    tmp.replace(path)

app/pipeline.py

@@ -25,6 +25,7 @@ from pathlib import Path
 from typing import Any
 
 from . import config
+from .io_utils import read_json, write_json_atomic
 
 logger = logging.getLogger("livef12.pipeline")
 
@@ -236,10 +237,10 @@ def preprocess_original(paths: JobPaths) -> JobPaths:
         ]
     )
     try:
-        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
+        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=config.PREPROCESS_TIMEOUT)
     except subprocess.TimeoutExpired as exc:
         tmp.unlink(missing_ok=True)
-        raise PipelineError("Preprocess Timeout nach 120s") from exc
+        raise PipelineError(f"Preprocess Timeout nach {config.PREPROCESS_TIMEOUT}s") from exc
     if proc.returncode != 0 or not tmp.exists() or tmp.stat().st_size == 0:
         tmp.unlink(missing_ok=True)
         err = strip_ansi((proc.stderr or proc.stdout or "").strip())
@@ -346,26 +347,20 @@ def job_paths(stem: str, original_suffix: str = ".jpg") -> JobPaths:
     )
 
 
-def _read_meta(stem: str) -> dict[str, Any] | None:
+def read_meta(stem: str) -> dict[str, Any] | None:
+    """Return the full meta record for a job, or None if missing/corrupt."""
     paths = job_paths(stem)
-    if not paths.meta.exists():
+    data = read_json(paths.meta)
-        return None
+    return data if isinstance(data, dict) else None
-    try:
-        return json.loads(paths.meta.read_text(encoding="utf-8"))
-    except (OSError, ValueError):
-        return None
 
 
 def _write_meta(paths: JobPaths, data: dict[str, Any]) -> None:
-    paths.meta.parent.mkdir(parents=True, exist_ok=True)
     data["updated_at"] = now_iso()
-    tmp = paths.meta.with_suffix(".json.tmp")
+    write_json_atomic(paths.meta, data)
-    tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
-    tmp.replace(paths.meta)
 
 
 def read_status(job_id: str) -> dict[str, Any] | None:
-    data = _read_meta(job_id)
+    data = read_meta(job_id)
     if not data:
         return None
     return {
@@ -380,7 +375,7 @@ def read_status(job_id: str) -> dict[str, Any] | None:
 
 
 def write_status(paths: JobPaths, status: str, **extra: Any) -> None:
-    data = _read_meta(paths.stem) or {}
+    data = read_meta(paths.stem) or {}
     data["job_id"] = paths.stem
     data["source_stem"] = paths.stem
     data["status"] = status
@@ -390,7 +385,7 @@ def write_status(paths: JobPaths, status: str, **extra: Any) -> None:
 
 
 def read_manifest(job_id: str) -> dict[str, Any] | None:
-    data = _read_meta(job_id)
+    data = read_meta(job_id)
     if not data:
         return None
     return {
@@ -405,7 +400,7 @@ def read_manifest(job_id: str) -> dict[str, Any] | None:
 
 
 def write_manifest(paths: JobPaths, manifest: dict[str, Any]) -> None:
-    data = _read_meta(paths.stem) or {}
+    data = read_meta(paths.stem) or {}
     data["job_id"] = paths.stem
     data["source_stem"] = manifest.get("source_stem") or paths.stem
     data["original_file"] = manifest.get("original_file")

app/worker.py

@@ -13,7 +13,6 @@ pipeline is CPU heavy (gmic/rembg) and the worker container is capped at
 
 from __future__ import annotations
 
-import json
 import logging
 import shutil
 import time
@@ -21,11 +20,10 @@ from pathlib import Path
 from typing import Any
 
 from . import config, pipeline
+from .io_utils import read_json, write_json_atomic
+from .logging_config import configure_logging
 
-logging.basicConfig(
+configure_logging()
-    level=logging.INFO,
-    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
-)
 logger = logging.getLogger("livef12.worker")
 
 
@@ -42,17 +40,15 @@ def ensure_dirs() -> None:
 def load_processed() -> dict[str, Any]:
     if not config.PROCESSED_FILE.exists():
         return {}
-    try:
+    data = read_json(config.PROCESSED_FILE)
-        return json.loads(config.PROCESSED_FILE.read_text(encoding="utf-8"))
+    if isinstance(data, dict):
-    except (OSError, ValueError):
+        return data
     logger.warning("processed.json unreadable, starting fresh")
     return {}
 
 
 def save_processed(processed: dict[str, Any]) -> None:
-    tmp = config.PROCESSED_FILE.with_suffix(".json.tmp")
+    write_json_atomic(config.PROCESSED_FILE, processed)
-    tmp.write_text(json.dumps(processed, indent=2, ensure_ascii=False), encoding="utf-8")
-    tmp.replace(config.PROCESSED_FILE)
 
 
 def _file_key(path: Path) -> str: