{title}
订单已经记录。你可以关闭这个页面，或返回 Feishu 继续查看通知。
当前库存快照已更新。
返回控制台
"""
return Response(content=html, media_type="text/html; charset=utf-8", headers={"Cache-Control": "no-store"})
@app.get("/api/inventory/order")
def place_inventory_order_get(
sku: str,
quantity: int = 1,
source: str = "feishu",
token: str | None = None,
) -> Response:
order = INVENTORY.record_order(
sku=sku,
quantity=quantity,
source=source,
token=token,
)
snapshot = INVENTORY.snapshot()
notify_robot_backend(
{
"kind": "inventory_order",
"order": order,
"inventory": snapshot,
}
)
return _inventory_order_html(order, snapshot)
@app.post("/api/inventory/order")
def place_inventory_order_post(payload: dict[str, Any]) -> dict[str, Any]:
sku = str(payload.get("sku") or "").strip()
if not sku:
raise HTTPException(status_code=400, detail="sku is required")
quantity = int(payload.get("quantity") or 1)
source = str(payload.get("source") or "manual").strip() or "manual"
token = str(payload.get("token") or payload.get("order_token") or "").strip() or None
order = INVENTORY.record_order(
sku=sku,
quantity=quantity,
source=source,
token=token,
)
snapshot = INVENTORY.snapshot()
notify_robot_backend(
{
"kind": "inventory_order",
"order": order,
"inventory": snapshot,
}
)
return {
"order": order,
"inventory": snapshot,
}
@app.post("/api/inventory/replenish")
def replenish_inventory(payload: dict[str, Any]) -> dict[str, Any]:
sku = str(payload.get("sku") or "").strip()
if not sku:
raise HTTPException(status_code=400, detail="sku is required")
quantity = int(payload.get("quantity") or 0)
event = INVENTORY.replenish(sku=sku, quantity=quantity)
snapshot = INVENTORY.snapshot()
notify_robot_backend(
{
"kind": "inventory_replenish",
"event": event,
"inventory": snapshot,
}
)
return {
"event": event,
"inventory": snapshot,
}
@app.post("/api/command")
def submit_command(payload: dict[str, Any]) -> dict[str, Any]:
command = _normalize_text(str(payload.get("command", "")))
if not command:
raise HTTPException(status_code=400, detail="command is required")
session_id = str(payload.get("session_id") or payload.get("sessionId") or "").strip()
with sessions_lock:
session = sessions.get(session_id) if session_id else None
if session is None:
session = SessionRecord()
sessions[session.session_id] = session
task = _parse_task(command)
blocked_reason = _extract_block_reason(command)
trace = _build_trace(task, blocked_reason=blocked_reason or None)
with session.condition:
session.command = command
session.task = task
session.trace = trace
session.preview = _fake_preview(task)
session.inventory = get_inventory_store().snapshot()
session.inventory_event = {}
session.logs = [
f"[{_now_iso()}] INFO: command accepted: {command}",
f"[{_now_iso()}] INFO: parsed task: {task['type']}",
]
session.updated_at = _now_iso()
session.revision += 1
if blocked_reason:
session.status = "failure"
session.current_step = "策略检查"
session.result = f"已被策略关键字拦截：{blocked_reason}"
session.logs.append(f"[{_now_iso()}] WARN: 策略拦截关键字：{blocked_reason}")
session.trace = _progress_trace(session.trace, 1, terminal=True)
session.revision += 1
session.condition.notify_all()
elif task["type"] == "interrupted":
session.status = "interrupted"
session.current_step = "会话结束"
session.result = "会话已关闭。"
session.logs.append(f"[{_now_iso()}] OK: 会话已由用户中断")
session.trace = _progress_trace(session.trace, None, terminal=True)
session.revision += 1
session.condition.notify_all()
else:
session.status = "running"
session.current_step = trace[0]["name"] if trace else "Running"
session.result = "指令正在执行中..."
session.trace = _progress_trace(session.trace, 0)
session.revision += 1
session.condition.notify_all()
threading.Thread(target=_run_session, args=(session.session_id,), daemon=True).start()
with sessions_lock:
sessions[session.session_id] = session
return session.payload()
@app.get("/api/session/{session_id}")
def get_session(session_id: str) -> dict[str, Any]:
with sessions_lock:
session = sessions.get(session_id)
if session is None:
raise HTTPException(status_code=404, detail="session not found")
return session.payload()
@app.get("/api/session/{session_id}/events")
def get_session_events(session_id: str) -> StreamingResponse:
with sessions_lock:
session = sessions.get(session_id)
if session is None:
raise HTTPException(status_code=404, detail="session not found")
def stream():
last_revision = -1
yield f"data: {json.dumps(session.payload(), ensure_ascii=False)}\n\n"
last_revision = session.revision
while True:
with session.condition:
session.condition.wait(timeout=15)
current_revision = session.revision
current_status = session.status
payload = session.payload()
if current_revision != last_revision:
last_revision = current_revision
yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
if current_status != "running":
break
return StreamingResponse(stream(), media_type="text/event-stream")
@app.get("/api/session/{session_id}/frame")
def get_session_frame(session_id: str) -> Response:
with sessions_lock:
session = sessions.get(session_id)
if session is None:
raise HTTPException(status_code=404, detail="session not found")
if not session.frame_jpeg:
raise HTTPException(status_code=404, detail="frame not ready")
return Response(
content=session.frame_jpeg,
media_type="image/png",
headers={"Cache-Control": "no-store, max-age=0"},
)
@app.get("/")
def root() -> FileResponse:
return FileResponse(INDEX_FILE)
app.mount("/", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend")
def _open_browser_later(url: str) -> None:
time.sleep(1.0)
try:
webbrowser.open(url)
except Exception:
pass
def main() -> None:
_assert_frontend_present()
os.environ.setdefault("OPENCLAW_RENDER_BACKEND", "glfw")
host = os.getenv("OPENCLAW_WEB_HOST", "127.0.0.1")
port = int(os.getenv("OPENCLAW_WEB_PORT", "8000"))
auto_open = os.getenv("OPENCLAW_WEB_AUTO_OPEN", "1").strip().lower() not in {"0", "false", "no"}
url = f"http://{host}:{port}/"
if auto_open:
threading.Thread(target=_open_browser_later, args=(url,), daemon=True).start()
import uvicorn
uvicorn.run(app, host=host, port=port, reload=False, log_level="info")
if __name__ == "__main__":
main()