"""
@app.get("/", response_class=HTMLResponse)
async def root():
"""Home page."""
return HOME_HTML
@app.post("/runs", response_model=RunStatus)
async def create_run(request: RunRequest):
"""Start a new rendering run."""
run_id = str(uuid.uuid4())
# Generate output name if not provided
output_name = request.output_name or f"{request.recipe}-{run_id[:8]}"
# Create run record
run = RunStatus(
run_id=run_id,
status="pending",
recipe=request.recipe,
inputs=request.inputs,
output_name=output_name,
created_at=datetime.now(timezone.utc).isoformat()
)
# Submit to Celery
# For now, we only support single-input recipes
if len(request.inputs) != 1:
raise HTTPException(400, "Currently only single-input recipes supported")
input_hash = request.inputs[0]
task = render_effect.delay(input_hash, request.recipe, output_name)
run.celery_task_id = task.id
run.status = "running"
save_run(run)
return run
@app.get("/runs/{run_id}", response_model=RunStatus)
async def get_run(run_id: str):
"""Get status of a run."""
run = load_run(run_id)
if not run:
raise HTTPException(404, f"Run {run_id} not found")
# Check Celery task status if running
if run.status == "running" and run.celery_task_id:
task = celery_app.AsyncResult(run.celery_task_id)
if task.ready():
if task.successful():
result = task.result
run.status = "completed"
run.completed_at = datetime.now(timezone.utc).isoformat()
run.output_hash = result.get("output", {}).get("content_hash")
# Cache the output
output_path = Path(result.get("output", {}).get("local_path", ""))
if output_path.exists():
cache_file(output_path)
else:
run.status = "failed"
run.error = str(task.result)
# Save updated status
save_run(run)
return run
@app.get("/runs")
async def list_runs():
"""List all runs."""
return list_all_runs()
@app.get("/cache/{content_hash}")
async def get_cached(content_hash: str):
"""Get cached content by hash."""
cache_path = CACHE_DIR / content_hash
if not cache_path.exists():
raise HTTPException(404, f"Content {content_hash} not in cache")
return FileResponse(cache_path)
@app.get("/ui/cache/{content_hash}", response_class=HTMLResponse)
async def ui_cache_view(content_hash: str):
"""View cached content with appropriate display."""
cache_path = CACHE_DIR / content_hash
if not cache_path.exists():
return HTMLResponse(f"""
Not Found | Art DAG L1
"""
@app.get("/ui", response_class=HTMLResponse)
async def ui_index():
"""Web UI for viewing runs."""
return UI_HTML
@app.get("/ui/runs", response_class=HTMLResponse)
async def ui_runs():
"""HTMX partial: list of runs."""
runs = list_all_runs()
if not runs:
return '
No runs yet.
'
html_parts = ['
']
for run in runs[:20]: # Limit to 20 most recent
status_class = run.status
effect_url = f"https://git.rose-ash.com/art-dag/effects/src/branch/main/{run.recipe}"
html_parts.append(f'''
')
return '\n'.join(html_parts)
@app.get("/ui/detail/{run_id}", response_class=HTMLResponse)
async def ui_detail_page(run_id: str):
"""Full detail page for a run."""
run = load_run(run_id)
if not run:
return HTMLResponse('
Run not found
', status_code=404)
# Check Celery task status if running
if run.status == "running" and run.celery_task_id:
task = celery_app.AsyncResult(run.celery_task_id)
if task.ready():
if task.successful():
result = task.result
run.status = "completed"
run.completed_at = datetime.now(timezone.utc).isoformat()
run.output_hash = result.get("output", {}).get("content_hash")
output_path = Path(result.get("output", {}).get("local_path", ""))
if output_path.exists():
cache_file(output_path)
else:
run.status = "failed"
run.error = str(task.result)
save_run(run)
effect_url = f"https://git.rose-ash.com/art-dag/effects/src/branch/main/{run.recipe}"
status_class = run.status
html = f"""
{run.recipe} - {run.run_id[:8]} | Art DAG L1
"""
# Media row
has_input = run.inputs and (CACHE_DIR / run.inputs[0]).exists()
has_output = run.status == "completed" and run.output_hash and (CACHE_DIR / run.output_hash).exists()
if has_input or has_output:
html += '
'
if has_input:
input_hash = run.inputs[0]
input_media_type = detect_media_type(CACHE_DIR / input_hash)
html += f'''
'''
if input_media_type == "video":
html += f''
elif input_media_type == "image":
html += f''
html += '
'
if has_output:
output_hash = run.output_hash
output_media_type = detect_media_type(CACHE_DIR / output_hash)
html += f'''
'''
if output_media_type == "video":
html += f''
elif output_media_type == "image":
html += f''
html += '
'''
return html
@app.get("/ui/run/{run_id}", response_class=HTMLResponse)
async def ui_run_partial(run_id: str):
"""HTMX partial: single run (for polling updates)."""
run = load_run(run_id)
if not run:
return '
Run not found
'
# Check Celery task status if running
if run.status == "running" and run.celery_task_id:
task = celery_app.AsyncResult(run.celery_task_id)
if task.ready():
if task.successful():
result = task.result
run.status = "completed"
run.completed_at = datetime.now(timezone.utc).isoformat()
run.output_hash = result.get("output", {}).get("content_hash")
output_path = Path(result.get("output", {}).get("local_path", ""))
if output_path.exists():
cache_file(output_path)
else:
run.status = "failed"
run.error = str(task.result)
save_run(run)
status_class = run.status
poll_attr = 'hx-get="/ui/run/{}" hx-trigger="every 2s" hx-swap="outerHTML"'.format(run_id) if run.status == "running" else ""
html = f'''
{run.recipe}{run.run_id}
{run.status}
Created: {run.created_at[:19].replace('T', ' ')}
'''
# Show input and output side by side
has_input = run.inputs and (CACHE_DIR / run.inputs[0]).exists()
has_output = run.status == "completed" and run.output_hash and (CACHE_DIR / run.output_hash).exists()
if has_input or has_output:
html += '
'
if has_input:
input_hash = run.inputs[0]
input_media_type = detect_media_type(CACHE_DIR / input_hash)
html += f'
'
if input_media_type == "video":
html += f''
elif input_media_type == "image":
html += f''
html += '
'
if has_output:
output_hash = run.output_hash
output_media_type = detect_media_type(CACHE_DIR / output_hash)
html += f'
'
if output_media_type == "video":
html += f''
elif output_media_type == "image":
html += f''
html += '
'
html += '
'
if run.status == "failed" and run.error:
html += f'
Error: {run.error}
'
html += '
'
return html
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8100)
'
else:
html += f'