diff --git a/.env.example b/.env.example index e5b694c..729f517 100644 --- a/.env.example +++ b/.env.example @@ -2,7 +2,7 @@ # --- Host data path (Syncthing share root) ------------------------------ # Prod default (compose.yml): /home/frank/sync.schwenk.online/data/livef12 -# Layout under that path: incoming/ jobs/ processed.json +# Layout under that path: incoming/ variants/ intermediates/ meta/ # Local override when the absolute path does not exist: # DATA_HOST_DIR=./data # DATA_HOST_DIR=/home/frank/sync.schwenk.online/data/livef12 diff --git a/NOTES.md b/NOTES.md index b97c9a6..6a42a06 100644 --- a/NOTES.md +++ b/NOTES.md @@ -11,7 +11,8 @@ - SFTPGo removed; drop path is Syncthing - Data share: `/home/frank/sync.schwenk.online/data/livef12` - (`incoming/` + `jobs/` + `processed.json`) + (`incoming/` + `variants/` + `intermediates/` + `meta/` + `processed.json`) +- Flat phone-friendly names: `{stem}_original.jpg`, `{stem}_v1.png`, … - Old path `/home/frank/live.f12.rocks/data` no longer used by compose (migration/cleanup manual) - Host port 12121 can be closed after cutover diff --git a/README.md b/README.md index 747bb7c..c028f53 100644 --- a/README.md +++ b/README.md @@ -23,17 +23,22 @@ Syncthing runs on the host — not in this compose. Prod path: `/home/frank/sync.schwenk.online/data/livef12` ``` -incoming/ # phone drop — worker only ever reads/copies from here -jobs// - original. # private copy of the uploaded photo - rembg.png # background removed (computed once, reused) - intermediates/ # every intermediate step, kept for inspection - variants/ # final composed images ({stem}_v1.png, …) - manifest.json # filter names/commands/blends per variant - status.json # pending | processing | done | error -processed.json # worker bookkeeping: which incoming files were handled +incoming/ # phone drop — worker only ever reads/copies from here +variants/ + {stem}_original.jpg # preprocessed private copy + {stem}_rembg.png + {stem}_v1.png # final composed images +intermediates/ + {stem}_v1_bg_filtered.png + {stem}_v1_post_0.png # … +meta/ + {stem}.json # status + manifest for the web UI +processed.json # worker bookkeeping: which incoming files were handled ``` +Phone tip: `.stignore` can drop `meta/` and `processed.json` if you only +want images on the device. + `assets/` (filter lists + trimmed `filters.json`) lives in the repo and is bind-mounted read-only into `worker`/`web` at `/data/assets`. @@ -47,8 +52,8 @@ docker compose up -d --build ``` Drop a photo into `incoming/` (via Syncthing or locally). After a few -seconds (poll interval + processing time) it shows up on the web UI as a -new job; the full `jobs//` tree syncs back to the phone. +seconds (poll interval + processing time) it shows up on the web UI; flat +files land in `variants/` and `intermediates/` for the phone. ## Configuration (`.env`, see `.env.example`) diff --git a/SOUL.md b/SOUL.md index e4ab434..8addb9b 100644 --- a/SOUL.md +++ b/SOUL.md @@ -60,7 +60,8 @@ bad venue wifi. Nobody signs up, nobody logs in. Whoever has the link (phone/folder setup is out of band). - No beamer/projector product built into this repo. - Not a DAM (digital asset manager) — no albums, tagging, search, - retention policies. `incoming/` and `jobs/` are the whole data model. + retention policies. `incoming/`, `variants/`, and `intermediates/` + are the whole data model. - No auto-deletion of incoming uploads. Ever. The worker only reads/copies. - No Syncthing container in this compose — Syncthing runs on the host. @@ -70,8 +71,12 @@ bad venue wifi. Nobody signs up, nobody logs in. Whoever has the link - Data volume (Syncthing share): `${DATA_HOST_DIR}` — prod default `/home/frank/sync.schwenk.online/data/livef12`, local override `./data`. Bind-mounted into `worker` and `web` at `/data` (see `compose.yml`). -- Layout under the share: `incoming/` (phone drop), `jobs/` (full tree - syncs back), `processed.json`. +- Layout under the share: + - `incoming/` — phone drop + - `variants/` — `{stem}_original.jpg`, `{stem}_rembg.png`, `{stem}_vN.png` + - `intermediates/` — `{stem}_vN_*.png` (incl. `_post_0` …) + - `meta/` — web bookkeeping JSON (optional `.stignore` on phone) + - `processed.json` - Domain: `live.f12.rocks`, routed via the shared external `traefik` network, TLS via `myresolver` (see `INFRASTRUCTURE.md`). - rembg model cache lives in the named Docker volume `rembg_cache`, not @@ -83,8 +88,8 @@ bad venue wifi. Nobody signs up, nobody logs in. Whoever has the link - Never delete or move files under incoming/ from worker code — Syncthing is the writer there. (reason: uploads are the one copy of the original that exists outside a phone's camera roll) -- Keep every intermediate image in jobs//intermediates/ — don't clean - them up automatically. (reason: useful for debugging bad filter picks, +- Keep every intermediate image under intermediates/ — don't clean them + up automatically. (reason: useful for debugging bad filter picks, disk is cheap compared to re-running gmic) - gmic/rembg subprocess calls always run under `nice` (see NICE_LEVEL) — the worker box needs to stay responsive for other things during an diff --git a/app/config.py b/app/config.py index 6ccd81b..60596f3 100644 --- a/app/config.py +++ b/app/config.py @@ -27,7 +27,11 @@ def _float_env(name: str, default: float) -> float: # --- Paths ------------------------------------------------------------- DATA_DIR = Path(os.environ.get("DATA_DIR", "/data")) INBOX_DIR = DATA_DIR / "incoming" -JOBS_DIR = DATA_DIR / "jobs" +VARIANTS_DIR = DATA_DIR / "variants" +INTERMEDIATES_DIR = DATA_DIR / "intermediates" +# Web bookkeeping (status + manifest). Not needed on the phone — ignore via +# Syncthing .stignore if desired. +META_DIR = DATA_DIR / "meta" ASSETS_DIR = DATA_DIR / "assets" PROCESSED_FILE = DATA_DIR / "processed.json" diff --git a/app/main.py b/app/main.py index e3b62fb..1d661d7 100644 --- a/app/main.py +++ b/app/main.py @@ -24,47 +24,45 @@ BASE_DIR = Path(__file__).resolve().parent templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") -_JOB_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") +# Job id = sanitized source stem (may include dots). +_JOB_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") def _validate_job_id(job_id: str) -> str: - if not _JOB_ID_RE.match(job_id): + stem = pipeline.sanitize_stem(job_id) + if not _JOB_ID_RE.match(stem): raise HTTPException(status_code=400, detail="Ungueltige Job-ID") - return job_id + return stem -def _job_root(job_id: str) -> Path: - root = (config.JOBS_DIR / _validate_job_id(job_id)).resolve() - jobs_dir = config.JOBS_DIR.resolve() - if root.parent != jobs_dir or not root.is_dir(): +def _job_exists(job_id: str) -> str: + stem = _validate_job_id(job_id) + if pipeline.read_status(stem) is None and not pipeline.job_paths(stem).original.exists(): raise HTTPException(status_code=404, detail="Job nicht gefunden") - return root + return stem -def _safe_job_file(job_id: str, rel_path: str) -> Path: - root = _job_root(job_id) - candidate = (root / rel_path).resolve() - if root not in candidate.parents and candidate != root: +def _safe_job_file(job_id: str, filename: str) -> Path: + """Resolve a basename under variants/ or intermediates/ for this stem.""" + stem = _validate_job_id(job_id) + name = Path(filename).name + if name != filename or "/" in filename or "\\" in filename or name.startswith("."): + raise HTTPException(status_code=400, detail="Ungueltiger Pfad") + if not name.startswith(f"{stem}_"): raise HTTPException(status_code=400, detail="Ungueltiger Pfad") - if not candidate.is_file(): - raise HTTPException(status_code=404, detail="Datei nicht gefunden") - return candidate - -def _original_suffix(job_root: Path) -> str: - for child in job_root.glob("original.*"): - return child.suffix - return ".jpg" + for directory in (config.VARIANTS_DIR, config.INTERMEDIATES_DIR): + candidate = (directory / name).resolve() + if directory.resolve() not in candidate.parents and candidate != directory.resolve(): + continue + if candidate.is_file(): + return candidate + raise HTTPException(status_code=404, detail="Datei nicht gefunden") def list_jobs() -> list[dict[str, Any]]: - if not config.JOBS_DIR.exists(): - return [] jobs = [] - for job_dir in config.JOBS_DIR.iterdir(): - if not job_dir.is_dir(): - continue - job_id = job_dir.name + for job_id in pipeline.list_job_ids(): status = pipeline.read_status(job_id) or {} manifest = pipeline.read_manifest(job_id) or {} jobs.append( @@ -76,7 +74,7 @@ def list_jobs() -> list[dict[str, Any]]: "thumbnail": (manifest.get("variants") or [{}])[-1].get("file") if manifest.get("variants") else None, } ) - jobs.sort(key=lambda j: j["job_id"], reverse=True) + jobs.sort(key=lambda j: j["created_at"] or j["job_id"], reverse=True) return jobs @@ -90,31 +88,33 @@ def index(request: Request) -> HTMLResponse: @app.get("/jobs/{job_id}", response_class=HTMLResponse) def job_detail(request: Request, job_id: str) -> HTMLResponse: - job_root = _job_root(job_id) - status = pipeline.read_status(job_id) or {} - manifest = pipeline.read_manifest(job_id) or {"variants": []} - original = next(iter(job_root.glob("original.*")), None) - rembg_file = job_root / "rembg.png" - intermediates = sorted((job_root / "intermediates").glob("*.png")) if (job_root / "intermediates").exists() else [] + stem = _job_exists(job_id) + paths = pipeline.job_paths(stem) + status = pipeline.read_status(stem) or {} + manifest = pipeline.read_manifest(stem) or {"variants": []} + intermediates = sorted( + p.name for p in paths.intermediates.glob(f"{stem}_*.png") if p.is_file() + ) if paths.intermediates.exists() else [] return templates.TemplateResponse( "job.html", { "request": request, "site_title": config.SITE_TITLE, - "job_id": job_id, + "job_id": stem, "status": status, "manifest": manifest, - "original_name": original.name if original else None, - "has_rembg": rembg_file.exists(), - "intermediates": [p.name for p in intermediates], + "original_name": paths.original.name if paths.original.exists() else None, + "rembg_name": paths.rembg.name if paths.rembg.exists() else None, + "has_rembg": paths.rembg.exists(), + "intermediates": intermediates, }, ) -@app.get("/jobs/{job_id}/files/{rel_path:path}") -def job_file(job_id: str, rel_path: str) -> FileResponse: - path = _safe_job_file(job_id, rel_path) +@app.get("/jobs/{job_id}/files/{filename:path}") +def job_file(job_id: str, filename: str) -> FileResponse: + path = _safe_job_file(job_id, filename) return FileResponse(path) @@ -125,8 +125,9 @@ def remix_form( error: str | None = None, from_variant: str | None = Query(None, alias="from"), ) -> HTMLResponse: - job_root = _job_root(job_id) - if not (job_root / "rembg.png").exists(): + stem = _job_exists(job_id) + paths = pipeline.job_paths(stem) + if not paths.rembg.exists(): raise HTTPException(status_code=409, detail="Job hat noch kein Rembg-Ergebnis, Remix noch nicht moeglich.") assets = pipeline.load_assets() @@ -139,12 +140,11 @@ def remix_form( "opacity": config.BLEND_OPACITY if config.BLEND_OPACITY in config.OPACITY_CHOICES else "30%", } if from_variant: - manifest = pipeline.read_manifest(job_id) or {} + manifest = pipeline.read_manifest(stem) or {} match = next((v for v in manifest.get("variants", []) if v.get("id") == from_variant), None) if match: opacity = match.get("blend_opacity") or prefill["opacity"] if opacity not in config.OPACITY_CHOICES: - # Snap odd random values (e.g. 37%) to nearest offered choice. try: pct = int(str(opacity).rstrip("%")) nearest = min( @@ -167,7 +167,7 @@ def remix_form( { "request": request, "site_title": config.SITE_TITLE, - "job_id": job_id, + "job_id": stem, "options": options, "opacity_choices": config.OPACITY_CHOICES, "prefill": prefill, @@ -186,13 +186,18 @@ def remix_submit( fg_blend: str = Form(...), opacity: str = Form(...), ) -> RedirectResponse: - job_root = _job_root(job_id) - suffix = _original_suffix(job_root) - choice = remix.RemixChoice(bg_filter=bg_filter, bg_blend=bg_blend, fg_filter=fg_filter, fg_blend=fg_blend, opacity=opacity) + stem = _job_exists(job_id) + choice = remix.RemixChoice( + bg_filter=bg_filter, + bg_blend=bg_blend, + fg_filter=fg_filter, + fg_blend=fg_blend, + opacity=opacity, + ) try: - entry = remix.create_remix_variant(job_id, suffix, choice) - logger.info("[%s] remix created variant %s", job_id, entry["id"]) + entry = remix.create_remix_variant(stem, choice) + logger.info("[%s] remix created variant %s", stem, entry["id"]) except pipeline.PipelineError as exc: - logger.warning("[%s] remix failed: %s", job_id, exc) - return RedirectResponse(url=f"/jobs/{job_id}/remix?error={quote(str(exc))}", status_code=303) - return RedirectResponse(url=f"/jobs/{job_id}", status_code=303) + logger.warning("[%s] remix failed: %s", stem, exc) + return RedirectResponse(url=f"/jobs/{stem}/remix?error={quote(str(exc))}", status_code=303) + return RedirectResponse(url=f"/jobs/{stem}", status_code=303) diff --git a/app/pipeline.py b/app/pipeline.py index b0f2c83..d63c2a5 100644 --- a/app/pipeline.py +++ b/app/pipeline.py @@ -207,20 +207,20 @@ def run_rembg(input_image: Path, output_image: Path) -> tuple[bool, str]: def preprocess_original(paths: JobPaths) -> JobPaths: - """Downscale to MAX_EDGE_PX and normalize to sRGB JPEG as `original.jpg`. + """Downscale to MAX_EDGE_PX and normalize to sRGB JPEG as `{stem}_original.jpg`. - Replaces any non-jpg original in the job dir. Side-effect free for the - incoming copy — only mutates jobs//. + Side-effect free for the incoming copy — only mutates variants/. """ src = paths.original if not src.exists(): raise PipelineError(f"Original fehlt: {src}") - dest = paths.root / "original.jpg" + dest = paths.variants / f"{paths.stem}_original.jpg" + paths.variants.mkdir(parents=True, exist_ok=True) + # Write to a temp name first so we can replace an existing file in-place + # without reading/writing the same path. + tmp = paths.variants / f".{paths.stem}_pre_{uuid.uuid4().hex[:8]}.jpg" resize = f"{config.MAX_EDGE_PX}x{config.MAX_EDGE_PX}>" - # Write to a temp name first so we can replace an existing original.jpg - # in-place without reading/writing the same path. - tmp = paths.root / f".original_pre_{uuid.uuid4().hex[:8]}.jpg" cmd = _nice( [ _magick_bin(), @@ -257,13 +257,12 @@ def preprocess_original(paths: JobPaths) -> JobPaths: config.MAX_EDGE_PX, ) return JobPaths( - root=paths.root, + stem=paths.stem, original=dest, rembg=paths.rembg, intermediates=paths.intermediates, variants=paths.variants, - manifest=paths.manifest, - status=paths.status, + meta=paths.meta, ) @@ -301,81 +300,11 @@ def now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") -def new_job_id() -> str: - stamp = datetime.now().strftime("%Y%m%d-%H%M%S") - return f"{stamp}-{uuid.uuid4().hex[:6]}" - - -@dataclass -class JobPaths: - root: Path - original: Path - rembg: Path - intermediates: Path - variants: Path - manifest: Path - status: Path - - -def job_paths(job_id: str, original_suffix: str = ".jpg") -> JobPaths: - root = config.JOBS_DIR / job_id - return JobPaths( - root=root, - original=root / f"original{original_suffix}", - rembg=root / "rembg.png", - intermediates=root / "intermediates", - variants=root / "variants", - manifest=root / "manifest.json", - status=root / "status.json", - ) - - -def read_status(job_id: str) -> dict[str, Any] | None: - paths = job_paths(job_id) - if not paths.status.exists(): - return None - return json.loads(paths.status.read_text(encoding="utf-8")) - - -def write_status(paths: JobPaths, status: str, **extra: Any) -> None: - data = {} - if paths.status.exists(): - try: - data = json.loads(paths.status.read_text(encoding="utf-8")) - except (OSError, ValueError): - data = {} - data["status"] = status - data["updated_at"] = now_iso() - data.setdefault("created_at", data["updated_at"]) - data.update(extra) - paths.status.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") - - -def read_manifest(job_id: str) -> dict[str, Any] | None: - paths = job_paths(job_id) - if not paths.manifest.exists(): - return None - return json.loads(paths.manifest.read_text(encoding="utf-8")) - - -def write_manifest(paths: JobPaths, manifest: dict[str, Any]) -> None: - manifest["updated_at"] = now_iso() - paths.manifest.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") - - -def next_variant_id(manifest: dict[str, Any]) -> str: - existing = {v["id"] for v in manifest.get("variants", [])} - i = len(manifest.get("variants", [])) + 1 - while f"v{i}" in existing: - i += 1 - return f"v{i}" - - _UNSAFE_STEM_RE = re.compile(r"[^\w.\-]+", re.UNICODE) def sanitize_stem(name: str) -> str: - """Safe basename stem for variant files (no path separators / junk).""" + """Safe basename stem for flat files (no path separators / junk).""" stem = Path(name).stem if name else "" stem = stem.replace("/", "_").replace("\\", "_").strip().strip(".") stem = _UNSAFE_STEM_RE.sub("_", stem).strip("._") @@ -386,13 +315,134 @@ def variant_filename(stem: str, variant_id: str) -> str: return f"{sanitize_stem(stem)}_{variant_id}.png" +@dataclass +class JobPaths: + """Flat layout under DATA_DIR — one stem, shared variants/ + intermediates/. + + variants/{stem}_original.jpg + variants/{stem}_rembg.png + variants/{stem}_v1.png + intermediates/{stem}_v1_*.png + meta/{stem}.json + """ + + stem: str + original: Path + rembg: Path + intermediates: Path + variants: Path + meta: Path + + +def job_paths(stem: str, original_suffix: str = ".jpg") -> JobPaths: + stem = sanitize_stem(stem) + return JobPaths( + stem=stem, + original=config.VARIANTS_DIR / f"{stem}_original{original_suffix}", + rembg=config.VARIANTS_DIR / f"{stem}_rembg.png", + intermediates=config.INTERMEDIATES_DIR, + variants=config.VARIANTS_DIR, + meta=config.META_DIR / f"{stem}.json", + ) + + +def _read_meta(stem: str) -> dict[str, Any] | None: + 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 + + +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) + + +def read_status(job_id: str) -> dict[str, Any] | None: + data = _read_meta(job_id) + if not data: + return None + return { + "status": data.get("status", "unknown"), + "created_at": data.get("created_at"), + "updated_at": data.get("updated_at"), + "error": data.get("error"), + "variant_errors": data.get("variant_errors"), + "source_file": data.get("source_file"), + "source_stem": data.get("source_stem"), + } + + +def write_status(paths: JobPaths, status: str, **extra: Any) -> None: + data = _read_meta(paths.stem) or {} + data["job_id"] = paths.stem + data["source_stem"] = paths.stem + data["status"] = status + data.setdefault("created_at", now_iso()) + data.update(extra) + _write_meta(paths, data) + + +def read_manifest(job_id: str) -> dict[str, Any] | None: + data = _read_meta(job_id) + if not data: + return None + return { + "job_id": data.get("job_id", job_id), + "original_file": data.get("original_file"), + "rembg_file": data.get("rembg_file"), + "source_stem": data.get("source_stem", job_id), + "created_at": data.get("created_at"), + "updated_at": data.get("updated_at"), + "variants": data.get("variants") or [], + } + + +def write_manifest(paths: JobPaths, manifest: dict[str, Any]) -> None: + 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") + data["rembg_file"] = manifest.get("rembg_file") + data["variants"] = manifest.get("variants") or [] + if "created_at" in manifest: + data.setdefault("created_at", manifest["created_at"]) + data.setdefault("status", data.get("status", "processing")) + _write_meta(paths, data) + + +def list_job_ids() -> list[str]: + if not config.META_DIR.exists(): + return [] + ids = [] + for path in config.META_DIR.glob("*.json"): + if path.name.endswith(".json.tmp"): + continue + ids.append(path.stem) + return ids + + +def next_variant_id(manifest: dict[str, Any]) -> str: + existing = {v["id"] for v in manifest.get("variants", [])} + i = len(manifest.get("variants", [])) + 1 + while f"v{i}" in existing: + i += 1 + return f"v{i}" + + def resolve_source_stem(manifest: dict[str, Any] | None, paths: JobPaths) -> str: - """Prefer manifest source_stem; fall back to original path stem.""" + """Prefer manifest source_stem; fall back to JobPaths.stem.""" if manifest: stored = manifest.get("source_stem") if isinstance(stored, str) and stored.strip(): return sanitize_stem(stored) - return sanitize_stem(paths.original.stem) + return paths.stem def compose_variant( @@ -401,7 +451,7 @@ def compose_variant( *, variant_id: str, source: str, - variant_stem: str, + variant_stem: str | None = None, bg_name: str | None = None, bg_mode: str | None = None, fg_name: str | None = None, @@ -411,16 +461,15 @@ def compose_variant( ) -> dict[str, Any]: """Compose one variant image from the job's cached original + rembg. - If bg_name/fg_name/bg_mode/fg_mode are given (remix path) they are used - directly. Otherwise a random working filter is picked with retries, - exactly like make_random.py's compose_one(). - - Final file is `variants/{stem}_{variant_id}.png`; manifest id stays `vN`. + Final file is `variants/{stem}_{variant_id}.png`; intermediates share + the same stem prefix under `intermediates/`. """ rng = rng or random.Random() tmp_dir = paths.intermediates tmp_dir.mkdir(parents=True, exist_ok=True) - stem = sanitize_stem(variant_stem) + paths.variants.mkdir(parents=True, exist_ok=True) + stem = sanitize_stem(variant_stem or paths.stem) + prefix = f"{stem}_{variant_id}" bg_mode = bg_mode or rng.choice(assets.blend_modes) fg_mode = fg_mode or rng.choice(assets.blend_modes) @@ -444,14 +493,13 @@ def compose_variant( fg_name, fg_command = pick_working_filter(assets.foreground_names, assets.commands, paths.rembg, tmp_dir, "fg", rng) p = { - "bg_filtered": tmp_dir / f"{variant_id}_bg_filtered.png", - "step1": tmp_dir / f"{variant_id}_bg_blend.png", - "step2": tmp_dir / f"{variant_id}_rembg_alpha.png", - "fg_filtered": tmp_dir / f"{variant_id}_fg_filtered.png", - "composed": tmp_dir / f"{variant_id}_composed.png", + "bg_filtered": tmp_dir / f"{prefix}_bg_filtered.png", + "step1": tmp_dir / f"{prefix}_bg_blend.png", + "step2": tmp_dir / f"{prefix}_rembg_alpha.png", + "fg_filtered": tmp_dir / f"{prefix}_fg_filtered.png", + "composed": tmp_dir / f"{prefix}_composed.png", "final": paths.variants / variant_filename(stem, variant_id), } - paths.variants.mkdir(parents=True, exist_ok=True) ok, err = apply_filter(paths.original, bg_command, p["bg_filtered"]) _raise_on_gmic_failure(ok, err, "Background-Filter fehlgeschlagen") @@ -468,14 +516,13 @@ def compose_variant( ok, err = blend_layers_opacity(p["step2"], p["fg_filtered"], fg_mode, opacity, p["composed"]) _raise_on_gmic_failure(ok, err, "Foreground-Blend fehlgeschlagen") - ok, err = apply_filter_chain(p["composed"], config.POST_FILTERS, p["final"], tmp_dir, variant_id) + ok, err = apply_filter_chain(p["composed"], config.POST_FILTERS, p["final"], tmp_dir, prefix) _raise_on_gmic_failure(ok, err, "Post-Processing fehlgeschlagen") - rel = lambda path: str(path.relative_to(paths.root)) return { "id": variant_id, "source": source, - "file": rel(p["final"]), + "file": p["final"].name, "background_filter": bg_name, "background_command": bg_command, "background_blend": bg_mode, @@ -486,11 +533,11 @@ def compose_variant( "post_filters": list(config.POST_FILTERS), "created_at": now_iso(), "intermediates": { - "bg_filtered": rel(p["bg_filtered"]), - "bg_blend": rel(p["step1"]), - "rembg_alpha": rel(p["step2"]), - "fg_filtered": rel(p["fg_filtered"]), - "composed": rel(p["composed"]), + "bg_filtered": p["bg_filtered"].name, + "bg_blend": p["step1"].name, + "rembg_alpha": p["step2"].name, + "fg_filtered": p["fg_filtered"].name, + "composed": p["composed"].name, }, } @@ -503,21 +550,16 @@ def process_job( source_stem: str | None = None, ) -> None: """Full pipeline for a freshly ingested incoming file: preprocess, rembg - once, generate OUTPUT_COUNT variants, write manifest + status. + once, generate OUTPUT_COUNT variants, write meta (status + manifest). - `source_path` must already be a private copy (jobs//original.*) — - callers (worker.py) are responsible for copying out of incoming first, so - the incoming file itself is never touched here. - - Variants that hit FILTER_TIMEOUT during the normal pass are retried - once at the end with FILTER_TIMEOUT_LONG. Filter probes in - pick_working_filter stay on the short timeout (skip, don't escalate). + `job_id` is the sanitized source stem. `source_path` must already be the + private copy at variants/{stem}_original{suffix}. """ - paths = job_paths(job_id, original_suffix) - paths.root.mkdir(parents=True, exist_ok=True) - paths.intermediates.mkdir(parents=True, exist_ok=True) + stem = sanitize_stem(source_stem or job_id) + paths = job_paths(stem, original_suffix) paths.variants.mkdir(parents=True, exist_ok=True) - stem = sanitize_stem(source_stem or source_path.stem) + paths.intermediates.mkdir(parents=True, exist_ok=True) + paths.meta.parent.mkdir(parents=True, exist_ok=True) write_status(paths, "processing", source_file=str(source_path.name), source_stem=stem) @@ -525,13 +567,14 @@ def process_job( paths = preprocess_original(paths) manifest: dict[str, Any] = { - "job_id": job_id, + "job_id": stem, "original_file": paths.original.name, "rembg_file": paths.rembg.name, "source_stem": stem, "created_at": now_iso(), "variants": [], } + write_manifest(paths, manifest) ok, err = run_rembg(paths.original, paths.rembg) if not ok: @@ -556,16 +599,16 @@ def process_job( manifest["variants"].append(entry) write_manifest(paths, manifest) except FilterTimeoutError as exc: - logger.warning("[%s] variant %s timed out (will retry long): %s", job_id, variant_id, exc) + logger.warning("[%s] variant %s timed out (will retry long): %s", stem, variant_id, exc) timeout_retries.append(variant_id) except PipelineError as exc: - logger.error("[%s] variant %s failed: %s", job_id, variant_id, exc) + logger.error("[%s] variant %s failed: %s", stem, variant_id, exc) variant_errors.append(f"{variant_id}: {exc}") if timeout_retries: logger.info( "[%s] long-retry %d variant(s) with timeout=%ds: %s", - job_id, + stem, len(timeout_retries), config.FILTER_TIMEOUT_LONG, ", ".join(timeout_retries), @@ -584,7 +627,7 @@ def process_job( manifest["variants"].append(entry) write_manifest(paths, manifest) except PipelineError as exc: - logger.error("[%s] variant %s long-retry failed: %s", job_id, variant_id, exc) + logger.error("[%s] variant %s long-retry failed: %s", stem, variant_id, exc) variant_errors.append(f"{variant_id}: {exc}") if not manifest["variants"]: @@ -592,8 +635,8 @@ def process_job( write_status(paths, "done", variant_errors=variant_errors) except PipelineError as exc: - logger.error("[%s] job failed: %s", job_id, exc) + logger.error("[%s] job failed: %s", stem, exc) write_status(paths, "error", error=str(exc)) except Exception as exc: # noqa: BLE001 - keep the worker loop alive - logger.exception("[%s] unexpected error", job_id) + logger.exception("[%s] unexpected error", stem) write_status(paths, "error", error=f"Unerwarteter Fehler: {exc}") diff --git a/app/remix.py b/app/remix.py index 7516688..a55e177 100644 --- a/app/remix.py +++ b/app/remix.py @@ -26,11 +26,12 @@ def build_remix_options(assets: FilterAssets) -> dict[str, list[str]]: } -def create_remix_variant(job_id: str, original_suffix: str, choice: RemixChoice) -> dict: +def create_remix_variant(job_id: str, choice: RemixChoice) -> dict: """Compose exactly one variant from explicit choices and append it to - the job's manifest. Raises PipelineError on failure (caller should show + the job's meta. Raises PipelineError on failure (caller should show it to the user, no half-written manifest entries are ever created).""" - paths: JobPaths = pipeline.job_paths(job_id, original_suffix) + stem = pipeline.sanitize_stem(job_id) + paths: JobPaths = pipeline.job_paths(stem, ".jpg") if not paths.original.exists() or not paths.rembg.exists(): raise PipelineError("Original oder Rembg-Bild fehlt fuer diesen Job — Remix nicht moeglich.") @@ -42,10 +43,11 @@ def create_remix_variant(job_id: str, original_suffix: str, choice: RemixChoice) if choice.bg_blend not in assets.blend_modes or choice.fg_blend not in assets.blend_modes: raise PipelineError("Unbekannter Blend-Modus.") - manifest = pipeline.read_manifest(job_id) or { - "job_id": job_id, + manifest = pipeline.read_manifest(stem) or { + "job_id": stem, "original_file": paths.original.name, "rembg_file": paths.rembg.name, + "source_stem": stem, "created_at": pipeline.now_iso(), "variants": [], } diff --git a/app/templates/job.html b/app/templates/job.html index a1e45b4..5bcb780 100644 --- a/app/templates/job.html +++ b/app/templates/job.html @@ -21,12 +21,12 @@ {% endif %} - {% if has_rembg %} + {% if has_rembg and rembg_name %}
- Freigestellt (rembg) + Freigestellt (rembg)
Rembg · - Download + Download
{% endif %} @@ -73,7 +73,7 @@
    {% for name in intermediates %}
  • - {{ name }} + {{ name }} {{ name }}
    • {% endfor %} diff --git a/app/worker.py b/app/worker.py index 546fb81..38ec362 100644 --- a/app/worker.py +++ b/app/worker.py @@ -1,10 +1,10 @@ """Incoming watcher + sequential job runner. Polls `DATA_DIR/incoming` for new, size-stable image files, copies each one -into its own `jobs//original.*` and runs the compose pipeline on -it. Files are NEVER deleted or moved from incoming — `processed.json` -tracks what has already been handled (by path + size + mtime) so restarts -don't reprocess everything. +into `variants/{stem}_original.*` and runs the compose pipeline on it. +Files are NEVER deleted or moved from incoming — `processed.json` tracks +what has already been handled (by path + size + mtime) so restarts don't +reprocess everything. Runs one job at a time (no threading) — this is intentional: the compose pipeline is CPU heavy (gmic/rembg) and the worker container is capped at @@ -30,7 +30,12 @@ logger = logging.getLogger("livef12.worker") def ensure_dirs() -> None: - for path in (config.INBOX_DIR, config.JOBS_DIR): + for path in ( + config.INBOX_DIR, + config.VARIANTS_DIR, + config.INTERMEDIATES_DIR, + config.META_DIR, + ): path.mkdir(parents=True, exist_ok=True) @@ -86,7 +91,7 @@ def _is_stable(path: Path) -> bool: def find_new_files(processed: dict[str, Any]) -> list[Path]: - """Top-level images under incoming/ only — never recurse into jobs/.""" + """Top-level images under incoming/ only — never recurse.""" if not config.INBOX_DIR.exists(): return [] candidates: list[Path] = [] @@ -113,26 +118,25 @@ def handle_file(path: Path, processed: dict[str, Any]) -> None: return stat = path.stat() - job_id = pipeline.new_job_id() + stem = pipeline.sanitize_stem(path.stem) suffix = path.suffix.lower() - paths = pipeline.job_paths(job_id, suffix) - paths.root.mkdir(parents=True, exist_ok=True) - source_stem = pipeline.sanitize_stem(path.stem) + paths = pipeline.job_paths(stem, suffix) + paths.variants.mkdir(parents=True, exist_ok=True) - logger.info("new incoming file %s -> job %s (stem=%s)", path.name, job_id, source_stem) + logger.info("new incoming file %s -> stem %s", path.name, stem) # Copy (never move) so incoming stays untouched. shutil.copy2(path, paths.original) try: - pipeline.process_job(job_id, paths.original, suffix, source_stem=source_stem) + pipeline.process_job(stem, paths.original, suffix, source_stem=stem) finally: # Mark as processed regardless of pipeline outcome so a permanently # broken image doesn't get retried forever; failures are visible in - # jobs//status.json for manual follow-up. + # meta/{stem}.json for manual follow-up. processed[_file_key(path)] = { "size": stat.st_size, "mtime": stat.st_mtime, - "job_id": job_id, + "job_id": stem, "processed_at": pipeline.now_iso(), } save_processed(processed) diff --git a/compose.yml b/compose.yml index 512064a..34ec8db 100644 --- a/compose.yml +++ b/compose.yml @@ -2,8 +2,8 @@ # # worker + web share one host directory (Syncthing share root): # ${DATA_HOST_DIR:-/home/frank/sync.schwenk.online/data/livef12} -> /data -# worker watches /data/incoming, writes /data/jobs -# web reads jobs, writes remix variants +# worker watches /data/incoming, writes /data/variants + /data/intermediates +# web reads meta/ + image dirs, writes remix variants # # Syncthing itself runs on the host (not in this compose). Copy .env.example # to .env for local overrides (e.g. DATA_HOST_DIR=./data). Never commit .env.