diff --git a/app/io_utils.py b/app/io_utils.py new file mode 100644 index 0000000..79e0c23 --- /dev/null +++ b/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) diff --git a/app/pipeline.py b/app/pipeline.py index aa78f98..8bcc736 100644 --- a/app/pipeline.py +++ b/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(): - return None - try: - return json.loads(paths.meta.read_text(encoding="utf-8")) - except (OSError, ValueError): - return None + data = read_json(paths.meta) + return data if isinstance(data, dict) else 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") - tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") - tmp.replace(paths.meta) + write_json_atomic(paths.meta, data) 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") diff --git a/app/worker.py b/app/worker.py index d4176b5..09e7b9a 100644 --- a/app/worker.py +++ b/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( - level=logging.INFO, - format="%(asctime)s %(levelname)s [%(name)s] %(message)s", -) +configure_logging() 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: - return json.loads(config.PROCESSED_FILE.read_text(encoding="utf-8")) - except (OSError, ValueError): - logger.warning("processed.json unreadable, starting fresh") - return {} + data = read_json(config.PROCESSED_FILE) + if isinstance(data, dict): + 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") - tmp.write_text(json.dumps(processed, indent=2, ensure_ascii=False), encoding="utf-8") - tmp.replace(config.PROCESSED_FILE) + write_json_atomic(config.PROCESSED_FILE, processed) def _file_key(path: Path) -> str: