feat: flatten output into variants/ and intermediates/

Drop per-job subdirs for phone-friendly Syncthing layout with
stem-suffixed filenames; keep web state in meta/.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Frank Schwenk
2026-07-18 14:52:47 +02:00
parent 9a9a3e3afe
commit b925b151f0
11 changed files with 289 additions and 220 deletions
+1 -1
View File
@@ -2,7 +2,7 @@
# --- Host data path (Syncthing share root) ------------------------------ # --- Host data path (Syncthing share root) ------------------------------
# Prod default (compose.yml): /home/frank/sync.schwenk.online/data/livef12 # 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: # Local override when the absolute path does not exist:
# DATA_HOST_DIR=./data # DATA_HOST_DIR=./data
# DATA_HOST_DIR=/home/frank/sync.schwenk.online/data/livef12 # DATA_HOST_DIR=/home/frank/sync.schwenk.online/data/livef12
+2 -1
View File
@@ -11,7 +11,8 @@
- SFTPGo removed; drop path is Syncthing - SFTPGo removed; drop path is Syncthing
- Data share: `/home/frank/sync.schwenk.online/data/livef12` - 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 - Old path `/home/frank/live.f12.rocks/data` no longer used by compose
(migration/cleanup manual) (migration/cleanup manual)
- Host port 12121 can be closed after cutover - Host port 12121 can be closed after cutover
+16 -11
View File
@@ -23,17 +23,22 @@ Syncthing runs on the host — not in this compose.
Prod path: `/home/frank/sync.schwenk.online/data/livef12` Prod path: `/home/frank/sync.schwenk.online/data/livef12`
``` ```
incoming/ # phone drop — worker only ever reads/copies from here incoming/ # phone drop — worker only ever reads/copies from here
jobs/<job_id>/ variants/
original.<ext> # private copy of the uploaded photo {stem}_original.jpg # preprocessed private copy
rembg.png # background removed (computed once, reused) {stem}_rembg.png
intermediates/ # every intermediate step, kept for inspection {stem}_v1.png # final composed images
variants/ # final composed images ({stem}_v1.png, …) intermediates/
manifest.json # filter names/commands/blends per variant {stem}_v1_bg_filtered.png
status.json # pending | processing | done | error {stem}_v1_post_0.png # …
processed.json # worker bookkeeping: which incoming files were handled 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 `assets/` (filter lists + trimmed `filters.json`) lives in the repo and is
bind-mounted read-only into `worker`/`web` at `/data/assets`. 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 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 seconds (poll interval + processing time) it shows up on the web UI; flat
new job; the full `jobs/<id>/` tree syncs back to the phone. files land in `variants/` and `intermediates/` for the phone.
## Configuration (`.env`, see `.env.example`) ## Configuration (`.env`, see `.env.example`)
+10 -5
View File
@@ -60,7 +60,8 @@ bad venue wifi. Nobody signs up, nobody logs in. Whoever has the link
(phone/folder setup is out of band). (phone/folder setup is out of band).
- No beamer/projector product built into this repo. - No beamer/projector product built into this repo.
- Not a DAM (digital asset manager) — no albums, tagging, search, - 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 auto-deletion of incoming uploads. Ever. The worker only reads/copies.
- No Syncthing container in this compose — Syncthing runs on the host. - 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 - Data volume (Syncthing share): `${DATA_HOST_DIR}` — prod default
`/home/frank/sync.schwenk.online/data/livef12`, local override `./data`. `/home/frank/sync.schwenk.online/data/livef12`, local override `./data`.
Bind-mounted into `worker` and `web` at `/data` (see `compose.yml`). Bind-mounted into `worker` and `web` at `/data` (see `compose.yml`).
- Layout under the share: `incoming/` (phone drop), `jobs/` (full tree - Layout under the share:
syncs back), `processed.json`. - `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` - Domain: `live.f12.rocks`, routed via the shared external `traefik`
network, TLS via `myresolver` (see `INFRASTRUCTURE.md`). network, TLS via `myresolver` (see `INFRASTRUCTURE.md`).
- rembg model cache lives in the named Docker volume `rembg_cache`, not - 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 - Never delete or move files under incoming/ from worker code — Syncthing
is the writer there. (reason: uploads are the one copy of the original is the writer there. (reason: uploads are the one copy of the original
that exists outside a phone's camera roll) that exists outside a phone's camera roll)
- Keep every intermediate image in jobs/<id>/intermediates/ — don't clean - Keep every intermediate image under intermediates/ — don't clean them
them up automatically. (reason: useful for debugging bad filter picks, up automatically. (reason: useful for debugging bad filter picks,
disk is cheap compared to re-running gmic) disk is cheap compared to re-running gmic)
- gmic/rembg subprocess calls always run under `nice` (see NICE_LEVEL) — - gmic/rembg subprocess calls always run under `nice` (see NICE_LEVEL) —
the worker box needs to stay responsive for other things during an the worker box needs to stay responsive for other things during an
+5 -1
View File
@@ -27,7 +27,11 @@ def _float_env(name: str, default: float) -> float:
# --- Paths ------------------------------------------------------------- # --- Paths -------------------------------------------------------------
DATA_DIR = Path(os.environ.get("DATA_DIR", "/data")) DATA_DIR = Path(os.environ.get("DATA_DIR", "/data"))
INBOX_DIR = DATA_DIR / "incoming" 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" ASSETS_DIR = DATA_DIR / "assets"
PROCESSED_FILE = DATA_DIR / "processed.json" PROCESSED_FILE = DATA_DIR / "processed.json"
+58 -53
View File
@@ -24,47 +24,45 @@ BASE_DIR = Path(__file__).resolve().parent
templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static") 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: 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") raise HTTPException(status_code=400, detail="Ungueltige Job-ID")
return job_id return stem
def _job_root(job_id: str) -> Path: def _job_exists(job_id: str) -> str:
root = (config.JOBS_DIR / _validate_job_id(job_id)).resolve() stem = _validate_job_id(job_id)
jobs_dir = config.JOBS_DIR.resolve() if pipeline.read_status(stem) is None and not pipeline.job_paths(stem).original.exists():
if root.parent != jobs_dir or not root.is_dir():
raise HTTPException(status_code=404, detail="Job nicht gefunden") raise HTTPException(status_code=404, detail="Job nicht gefunden")
return root return stem
def _safe_job_file(job_id: str, rel_path: str) -> Path: def _safe_job_file(job_id: str, filename: str) -> Path:
root = _job_root(job_id) """Resolve a basename under variants/ or intermediates/ for this stem."""
candidate = (root / rel_path).resolve() stem = _validate_job_id(job_id)
if root not in candidate.parents and candidate != root: 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") raise HTTPException(status_code=400, detail="Ungueltiger Pfad")
if not candidate.is_file():
raise HTTPException(status_code=404, detail="Datei nicht gefunden")
return candidate
for directory in (config.VARIANTS_DIR, config.INTERMEDIATES_DIR):
def _original_suffix(job_root: Path) -> str: candidate = (directory / name).resolve()
for child in job_root.glob("original.*"): if directory.resolve() not in candidate.parents and candidate != directory.resolve():
return child.suffix continue
return ".jpg" if candidate.is_file():
return candidate
raise HTTPException(status_code=404, detail="Datei nicht gefunden")
def list_jobs() -> list[dict[str, Any]]: def list_jobs() -> list[dict[str, Any]]:
if not config.JOBS_DIR.exists():
return []
jobs = [] jobs = []
for job_dir in config.JOBS_DIR.iterdir(): for job_id in pipeline.list_job_ids():
if not job_dir.is_dir():
continue
job_id = job_dir.name
status = pipeline.read_status(job_id) or {} status = pipeline.read_status(job_id) or {}
manifest = pipeline.read_manifest(job_id) or {} manifest = pipeline.read_manifest(job_id) or {}
jobs.append( 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, "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 return jobs
@@ -90,31 +88,33 @@ def index(request: Request) -> HTMLResponse:
@app.get("/jobs/{job_id}", response_class=HTMLResponse) @app.get("/jobs/{job_id}", response_class=HTMLResponse)
def job_detail(request: Request, job_id: str) -> HTMLResponse: def job_detail(request: Request, job_id: str) -> HTMLResponse:
job_root = _job_root(job_id) stem = _job_exists(job_id)
status = pipeline.read_status(job_id) or {} paths = pipeline.job_paths(stem)
manifest = pipeline.read_manifest(job_id) or {"variants": []} status = pipeline.read_status(stem) or {}
original = next(iter(job_root.glob("original.*")), None) manifest = pipeline.read_manifest(stem) or {"variants": []}
rembg_file = job_root / "rembg.png" intermediates = sorted(
intermediates = sorted((job_root / "intermediates").glob("*.png")) if (job_root / "intermediates").exists() else [] p.name for p in paths.intermediates.glob(f"{stem}_*.png") if p.is_file()
) if paths.intermediates.exists() else []
return templates.TemplateResponse( return templates.TemplateResponse(
"job.html", "job.html",
{ {
"request": request, "request": request,
"site_title": config.SITE_TITLE, "site_title": config.SITE_TITLE,
"job_id": job_id, "job_id": stem,
"status": status, "status": status,
"manifest": manifest, "manifest": manifest,
"original_name": original.name if original else None, "original_name": paths.original.name if paths.original.exists() else None,
"has_rembg": rembg_file.exists(), "rembg_name": paths.rembg.name if paths.rembg.exists() else None,
"intermediates": [p.name for p in intermediates], "has_rembg": paths.rembg.exists(),
"intermediates": intermediates,
}, },
) )
@app.get("/jobs/{job_id}/files/{rel_path:path}") @app.get("/jobs/{job_id}/files/{filename:path}")
def job_file(job_id: str, rel_path: str) -> FileResponse: def job_file(job_id: str, filename: str) -> FileResponse:
path = _safe_job_file(job_id, rel_path) path = _safe_job_file(job_id, filename)
return FileResponse(path) return FileResponse(path)
@@ -125,8 +125,9 @@ def remix_form(
error: str | None = None, error: str | None = None,
from_variant: str | None = Query(None, alias="from"), from_variant: str | None = Query(None, alias="from"),
) -> HTMLResponse: ) -> HTMLResponse:
job_root = _job_root(job_id) stem = _job_exists(job_id)
if not (job_root / "rembg.png").exists(): 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.") raise HTTPException(status_code=409, detail="Job hat noch kein Rembg-Ergebnis, Remix noch nicht moeglich.")
assets = pipeline.load_assets() 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%", "opacity": config.BLEND_OPACITY if config.BLEND_OPACITY in config.OPACITY_CHOICES else "30%",
} }
if from_variant: 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) match = next((v for v in manifest.get("variants", []) if v.get("id") == from_variant), None)
if match: if match:
opacity = match.get("blend_opacity") or prefill["opacity"] opacity = match.get("blend_opacity") or prefill["opacity"]
if opacity not in config.OPACITY_CHOICES: if opacity not in config.OPACITY_CHOICES:
# Snap odd random values (e.g. 37%) to nearest offered choice.
try: try:
pct = int(str(opacity).rstrip("%")) pct = int(str(opacity).rstrip("%"))
nearest = min( nearest = min(
@@ -167,7 +167,7 @@ def remix_form(
{ {
"request": request, "request": request,
"site_title": config.SITE_TITLE, "site_title": config.SITE_TITLE,
"job_id": job_id, "job_id": stem,
"options": options, "options": options,
"opacity_choices": config.OPACITY_CHOICES, "opacity_choices": config.OPACITY_CHOICES,
"prefill": prefill, "prefill": prefill,
@@ -186,13 +186,18 @@ def remix_submit(
fg_blend: str = Form(...), fg_blend: str = Form(...),
opacity: str = Form(...), opacity: str = Form(...),
) -> RedirectResponse: ) -> RedirectResponse:
job_root = _job_root(job_id) stem = _job_exists(job_id)
suffix = _original_suffix(job_root) choice = remix.RemixChoice(
choice = remix.RemixChoice(bg_filter=bg_filter, bg_blend=bg_blend, fg_filter=fg_filter, fg_blend=fg_blend, opacity=opacity) bg_filter=bg_filter,
bg_blend=bg_blend,
fg_filter=fg_filter,
fg_blend=fg_blend,
opacity=opacity,
)
try: try:
entry = remix.create_remix_variant(job_id, suffix, choice) entry = remix.create_remix_variant(stem, choice)
logger.info("[%s] remix created variant %s", job_id, entry["id"]) logger.info("[%s] remix created variant %s", stem, entry["id"])
except pipeline.PipelineError as exc: except pipeline.PipelineError as exc:
logger.warning("[%s] remix failed: %s", job_id, exc) logger.warning("[%s] remix failed: %s", stem, exc)
return RedirectResponse(url=f"/jobs/{job_id}/remix?error={quote(str(exc))}", status_code=303) return RedirectResponse(url=f"/jobs/{stem}/remix?error={quote(str(exc))}", status_code=303)
return RedirectResponse(url=f"/jobs/{job_id}", status_code=303) return RedirectResponse(url=f"/jobs/{stem}", status_code=303)
+166 -123
View File
@@ -207,20 +207,20 @@ def run_rembg(input_image: Path, output_image: Path) -> tuple[bool, str]:
def preprocess_original(paths: JobPaths) -> JobPaths: 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 Side-effect free for the incoming copy — only mutates variants/.
incoming copy — only mutates jobs/<id>/.
""" """
src = paths.original src = paths.original
if not src.exists(): if not src.exists():
raise PipelineError(f"Original fehlt: {src}") 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}>" 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( cmd = _nice(
[ [
_magick_bin(), _magick_bin(),
@@ -257,13 +257,12 @@ def preprocess_original(paths: JobPaths) -> JobPaths:
config.MAX_EDGE_PX, config.MAX_EDGE_PX,
) )
return JobPaths( return JobPaths(
root=paths.root, stem=paths.stem,
original=dest, original=dest,
rembg=paths.rembg, rembg=paths.rembg,
intermediates=paths.intermediates, intermediates=paths.intermediates,
variants=paths.variants, variants=paths.variants,
manifest=paths.manifest, meta=paths.meta,
status=paths.status,
) )
@@ -301,81 +300,11 @@ def now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds") 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) _UNSAFE_STEM_RE = re.compile(r"[^\w.\-]+", re.UNICODE)
def sanitize_stem(name: str) -> str: 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 = Path(name).stem if name else ""
stem = stem.replace("/", "_").replace("\\", "_").strip().strip(".") stem = stem.replace("/", "_").replace("\\", "_").strip().strip(".")
stem = _UNSAFE_STEM_RE.sub("_", stem).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" 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: 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: if manifest:
stored = manifest.get("source_stem") stored = manifest.get("source_stem")
if isinstance(stored, str) and stored.strip(): if isinstance(stored, str) and stored.strip():
return sanitize_stem(stored) return sanitize_stem(stored)
return sanitize_stem(paths.original.stem) return paths.stem
def compose_variant( def compose_variant(
@@ -401,7 +451,7 @@ def compose_variant(
*, *,
variant_id: str, variant_id: str,
source: str, source: str,
variant_stem: str, variant_stem: str | None = None,
bg_name: str | None = None, bg_name: str | None = None,
bg_mode: str | None = None, bg_mode: str | None = None,
fg_name: str | None = None, fg_name: str | None = None,
@@ -411,16 +461,15 @@ def compose_variant(
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Compose one variant image from the job's cached original + rembg. """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 Final file is `variants/{stem}_{variant_id}.png`; intermediates share
directly. Otherwise a random working filter is picked with retries, the same stem prefix under `intermediates/`.
exactly like make_random.py's compose_one().
Final file is `variants/{stem}_{variant_id}.png`; manifest id stays `vN`.
""" """
rng = rng or random.Random() rng = rng or random.Random()
tmp_dir = paths.intermediates tmp_dir = paths.intermediates
tmp_dir.mkdir(parents=True, exist_ok=True) 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) bg_mode = bg_mode or rng.choice(assets.blend_modes)
fg_mode = fg_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) fg_name, fg_command = pick_working_filter(assets.foreground_names, assets.commands, paths.rembg, tmp_dir, "fg", rng)
p = { p = {
"bg_filtered": tmp_dir / f"{variant_id}_bg_filtered.png", "bg_filtered": tmp_dir / f"{prefix}_bg_filtered.png",
"step1": tmp_dir / f"{variant_id}_bg_blend.png", "step1": tmp_dir / f"{prefix}_bg_blend.png",
"step2": tmp_dir / f"{variant_id}_rembg_alpha.png", "step2": tmp_dir / f"{prefix}_rembg_alpha.png",
"fg_filtered": tmp_dir / f"{variant_id}_fg_filtered.png", "fg_filtered": tmp_dir / f"{prefix}_fg_filtered.png",
"composed": tmp_dir / f"{variant_id}_composed.png", "composed": tmp_dir / f"{prefix}_composed.png",
"final": paths.variants / variant_filename(stem, variant_id), "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"]) ok, err = apply_filter(paths.original, bg_command, p["bg_filtered"])
_raise_on_gmic_failure(ok, err, "Background-Filter fehlgeschlagen") _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"]) ok, err = blend_layers_opacity(p["step2"], p["fg_filtered"], fg_mode, opacity, p["composed"])
_raise_on_gmic_failure(ok, err, "Foreground-Blend fehlgeschlagen") _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") _raise_on_gmic_failure(ok, err, "Post-Processing fehlgeschlagen")
rel = lambda path: str(path.relative_to(paths.root))
return { return {
"id": variant_id, "id": variant_id,
"source": source, "source": source,
"file": rel(p["final"]), "file": p["final"].name,
"background_filter": bg_name, "background_filter": bg_name,
"background_command": bg_command, "background_command": bg_command,
"background_blend": bg_mode, "background_blend": bg_mode,
@@ -486,11 +533,11 @@ def compose_variant(
"post_filters": list(config.POST_FILTERS), "post_filters": list(config.POST_FILTERS),
"created_at": now_iso(), "created_at": now_iso(),
"intermediates": { "intermediates": {
"bg_filtered": rel(p["bg_filtered"]), "bg_filtered": p["bg_filtered"].name,
"bg_blend": rel(p["step1"]), "bg_blend": p["step1"].name,
"rembg_alpha": rel(p["step2"]), "rembg_alpha": p["step2"].name,
"fg_filtered": rel(p["fg_filtered"]), "fg_filtered": p["fg_filtered"].name,
"composed": rel(p["composed"]), "composed": p["composed"].name,
}, },
} }
@@ -503,21 +550,16 @@ def process_job(
source_stem: str | None = None, source_stem: str | None = None,
) -> None: ) -> None:
"""Full pipeline for a freshly ingested incoming file: preprocess, rembg """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/<id>/original.*) — `job_id` is the sanitized source stem. `source_path` must already be the
callers (worker.py) are responsible for copying out of incoming first, so private copy at variants/{stem}_original{suffix}.
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).
""" """
paths = job_paths(job_id, original_suffix) stem = sanitize_stem(source_stem or job_id)
paths.root.mkdir(parents=True, exist_ok=True) paths = job_paths(stem, original_suffix)
paths.intermediates.mkdir(parents=True, exist_ok=True)
paths.variants.mkdir(parents=True, exist_ok=True) 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) write_status(paths, "processing", source_file=str(source_path.name), source_stem=stem)
@@ -525,13 +567,14 @@ def process_job(
paths = preprocess_original(paths) paths = preprocess_original(paths)
manifest: dict[str, Any] = { manifest: dict[str, Any] = {
"job_id": job_id, "job_id": stem,
"original_file": paths.original.name, "original_file": paths.original.name,
"rembg_file": paths.rembg.name, "rembg_file": paths.rembg.name,
"source_stem": stem, "source_stem": stem,
"created_at": now_iso(), "created_at": now_iso(),
"variants": [], "variants": [],
} }
write_manifest(paths, manifest)
ok, err = run_rembg(paths.original, paths.rembg) ok, err = run_rembg(paths.original, paths.rembg)
if not ok: if not ok:
@@ -556,16 +599,16 @@ def process_job(
manifest["variants"].append(entry) manifest["variants"].append(entry)
write_manifest(paths, manifest) write_manifest(paths, manifest)
except FilterTimeoutError as exc: 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) timeout_retries.append(variant_id)
except PipelineError as exc: 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}") variant_errors.append(f"{variant_id}: {exc}")
if timeout_retries: if timeout_retries:
logger.info( logger.info(
"[%s] long-retry %d variant(s) with timeout=%ds: %s", "[%s] long-retry %d variant(s) with timeout=%ds: %s",
job_id, stem,
len(timeout_retries), len(timeout_retries),
config.FILTER_TIMEOUT_LONG, config.FILTER_TIMEOUT_LONG,
", ".join(timeout_retries), ", ".join(timeout_retries),
@@ -584,7 +627,7 @@ def process_job(
manifest["variants"].append(entry) manifest["variants"].append(entry)
write_manifest(paths, manifest) write_manifest(paths, manifest)
except PipelineError as exc: 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}") variant_errors.append(f"{variant_id}: {exc}")
if not manifest["variants"]: if not manifest["variants"]:
@@ -592,8 +635,8 @@ def process_job(
write_status(paths, "done", variant_errors=variant_errors) write_status(paths, "done", variant_errors=variant_errors)
except PipelineError as exc: 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)) write_status(paths, "error", error=str(exc))
except Exception as exc: # noqa: BLE001 - keep the worker loop alive 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}") write_status(paths, "error", error=f"Unerwarteter Fehler: {exc}")
+7 -5
View File
@@ -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 """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).""" 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(): if not paths.original.exists() or not paths.rembg.exists():
raise PipelineError("Original oder Rembg-Bild fehlt fuer diesen Job — Remix nicht moeglich.") 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: if choice.bg_blend not in assets.blend_modes or choice.fg_blend not in assets.blend_modes:
raise PipelineError("Unbekannter Blend-Modus.") raise PipelineError("Unbekannter Blend-Modus.")
manifest = pipeline.read_manifest(job_id) or { manifest = pipeline.read_manifest(stem) or {
"job_id": job_id, "job_id": stem,
"original_file": paths.original.name, "original_file": paths.original.name,
"rembg_file": paths.rembg.name, "rembg_file": paths.rembg.name,
"source_stem": stem,
"created_at": pipeline.now_iso(), "created_at": pipeline.now_iso(),
"variants": [], "variants": [],
} }
+4 -4
View File
@@ -21,12 +21,12 @@
</figcaption> </figcaption>
</figure> </figure>
{% endif %} {% endif %}
{% if has_rembg %} {% if has_rembg and rembg_name %}
<figure> <figure>
<img class="lightboxable" src="/jobs/{{ job_id }}/files/rembg.png" alt="Freigestellt (rembg)" loading="lazy"> <img class="lightboxable" src="/jobs/{{ job_id }}/files/{{ rembg_name }}" alt="Freigestellt (rembg)" loading="lazy">
<figcaption> <figcaption>
Rembg &middot; Rembg &middot;
<a href="/jobs/{{ job_id }}/files/rembg.png" download>Download</a> <a href="/jobs/{{ job_id }}/files/{{ rembg_name }}" download>Download</a>
</figcaption> </figcaption>
</figure> </figure>
{% endif %} {% endif %}
@@ -73,7 +73,7 @@
<ul class="intermediate-grid"> <ul class="intermediate-grid">
{% for name in intermediates %} {% for name in intermediates %}
<li> <li>
<img class="lightboxable" src="/jobs/{{ job_id }}/files/intermediates/{{ name }}" alt="{{ name }}" loading="lazy"> <img class="lightboxable" src="/jobs/{{ job_id }}/files/{{ name }}" alt="{{ name }}" loading="lazy">
<span>{{ name }}</span> <span>{{ name }}</span>
</li> </li>
{% endfor %} {% endfor %}
+18 -14
View File
@@ -1,10 +1,10 @@
"""Incoming watcher + sequential job runner. """Incoming watcher + sequential job runner.
Polls `DATA_DIR/incoming` for new, size-stable image files, copies each one Polls `DATA_DIR/incoming` for new, size-stable image files, copies each one
into its own `jobs/<job_id>/original.*` and runs the compose pipeline on into `variants/{stem}_original.*` and runs the compose pipeline on it.
it. Files are NEVER deleted or moved from incoming — `processed.json` Files are NEVER deleted or moved from incoming — `processed.json` tracks
tracks what has already been handled (by path + size + mtime) so restarts what has already been handled (by path + size + mtime) so restarts don't
don't reprocess everything. reprocess everything.
Runs one job at a time (no threading) — this is intentional: the compose 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 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: 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) 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]: 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(): if not config.INBOX_DIR.exists():
return [] return []
candidates: list[Path] = [] candidates: list[Path] = []
@@ -113,26 +118,25 @@ def handle_file(path: Path, processed: dict[str, Any]) -> None:
return return
stat = path.stat() stat = path.stat()
job_id = pipeline.new_job_id() stem = pipeline.sanitize_stem(path.stem)
suffix = path.suffix.lower() suffix = path.suffix.lower()
paths = pipeline.job_paths(job_id, suffix) paths = pipeline.job_paths(stem, suffix)
paths.root.mkdir(parents=True, exist_ok=True) paths.variants.mkdir(parents=True, exist_ok=True)
source_stem = pipeline.sanitize_stem(path.stem)
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. # Copy (never move) so incoming stays untouched.
shutil.copy2(path, paths.original) shutil.copy2(path, paths.original)
try: try:
pipeline.process_job(job_id, paths.original, suffix, source_stem=source_stem) pipeline.process_job(stem, paths.original, suffix, source_stem=stem)
finally: finally:
# Mark as processed regardless of pipeline outcome so a permanently # Mark as processed regardless of pipeline outcome so a permanently
# broken image doesn't get retried forever; failures are visible in # broken image doesn't get retried forever; failures are visible in
# jobs/<id>/status.json for manual follow-up. # meta/{stem}.json for manual follow-up.
processed[_file_key(path)] = { processed[_file_key(path)] = {
"size": stat.st_size, "size": stat.st_size,
"mtime": stat.st_mtime, "mtime": stat.st_mtime,
"job_id": job_id, "job_id": stem,
"processed_at": pipeline.now_iso(), "processed_at": pipeline.now_iso(),
} }
save_processed(processed) save_processed(processed)
+2 -2
View File
@@ -2,8 +2,8 @@
# #
# worker + web share one host directory (Syncthing share root): # worker + web share one host directory (Syncthing share root):
# ${DATA_HOST_DIR:-/home/frank/sync.schwenk.online/data/livef12} -> /data # ${DATA_HOST_DIR:-/home/frank/sync.schwenk.online/data/livef12} -> /data
# worker watches /data/incoming, writes /data/jobs # worker watches /data/incoming, writes /data/variants + /data/intermediates
# web reads jobs, writes remix variants # web reads meta/ + image dirs, writes remix variants
# #
# Syncthing itself runs on the host (not in this compose). Copy .env.example # 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. # to .env for local overrides (e.g. DATA_HOST_DIR=./data). Never commit .env.