Spaces:
Sleeping
Sleeping
| """ | |
| result.py — GET /result/{job_id} | |
| Returns the full step history for a job plus backward-compatible top-level | |
| status/image URL fields derived from the current step. | |
| Response design | |
| --------------- | |
| The response includes two layers: | |
| 1. Top-level fields (backward-compatible with the current frontend polling): | |
| status — current step's status | |
| result_image_url — current step's result image URL | |
| original_image_url — step 0's original image URL (the uploaded file) | |
| error_message — current step's error (if failed) | |
| 2. steps[] array (for the future multi-step navigation UI): | |
| Each entry has step_number, status, original_image_url, result_image_url, | |
| error_message. The frontend (next step) uses this to render a step | |
| navigator. | |
| URL derivation | |
| -------------- | |
| Files live under STORAGE_DIR and are served at /storage/... by StaticFiles. | |
| Path.relative_to(STORAGE_DIR).as_posix() extracts the relative path and | |
| normalises Windows backslashes to forward slashes. | |
| """ | |
| from pathlib import Path | |
| from fastapi import APIRouter, HTTPException | |
| from app.config import STORAGE_DIR | |
| from app.jobs_store import get_job | |
| router = APIRouter() | |
| def _to_url(abs_path: str | None) -> str | None: | |
| """Convert an absolute path under STORAGE_DIR to a /storage/… URL.""" | |
| if abs_path is None: | |
| return None | |
| try: | |
| rel = Path(abs_path).relative_to(STORAGE_DIR) | |
| return "/storage/" + rel.as_posix() | |
| except ValueError: | |
| return None | |
| def get_result(job_id: str) -> dict: | |
| """Return current job state including full step history. | |
| The top-level status/result_image_url/original_image_url fields reflect | |
| the current step, keeping existing frontend polling code working until | |
| the multi-step UI is implemented. | |
| """ | |
| job = get_job(job_id) | |
| if job is None: | |
| raise HTTPException(status_code=404, detail=f"Job '{job_id}' not found.") | |
| # Find current step record. | |
| current = next( | |
| (s for s in job.steps if s.step_number == job.current_step), None | |
| ) | |
| # Step 0's original is the canonical "uploaded image" for before/after. | |
| step0 = next((s for s in job.steps if s.step_number == 0), None) | |
| # Build per-step summary for the history navigator. | |
| steps_summary = [ | |
| { | |
| "step_number": s.step_number, | |
| "status": s.status, | |
| "original_image_url": _to_url(s.original_image), | |
| "result_image_url": _to_url(s.result_image), | |
| "error_message": s.error_message, | |
| } | |
| for s in sorted(job.steps, key=lambda s: s.step_number) | |
| ] | |
| return { | |
| # Backward-compatible top-level fields (polled by existing frontend). | |
| "job_id": job.id, | |
| "current_step": job.current_step, | |
| "status": current.status if current else "unknown", | |
| "error_message": current.error_message if current else None, | |
| "original_image_url": _to_url(step0.original_image) if step0 else None, | |
| "result_image_url": _to_url(current.result_image) if current else None, | |
| # Full history for the multi-step UI (used in the next frontend step). | |
| "steps": steps_summary, | |
| } | |