Update app.py

app.py

@@ -1,161 +1,198 @@
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-import httpx
-import json
-import logging
-
-app = FastAPI()
-logging.basicConfig(level=logging.INFO)
-
-# Enable CORS for all origins
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-BASE_62_MAP = {c: i for i, c in enumerate("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")}
-
-async def get_client() -> httpx.AsyncClient:
-    if not hasattr(app.state, "client"):
-        app.state.client = httpx.AsyncClient(timeout=15.0)
-    return app.state.client
-
-def base62_to_int(token: str) -> int:
-    result = 0
-    for ch in token:
-        result = result * 62 + BASE_62_MAP[ch]
-    return result
-
-async def get_base_url(token: str) -> str:
-    first = token[0]
-    if first == "A":
-        n = base62_to_int(token[1])
-    else:
-        n = base62_to_int(token[1:3])
-    return f"https://p{n:02d}-sharedstreams.icloud.com/{token}/sharedstreams/"
-
-ICLOUD_HEADERS = {
-    "Origin": "https://www.icloud.com",
-    "Content-Type": "text/plain"
-}
-ICLOUD_PAYLOAD = '{"streamCtag":null}'
-
-async def get_redirected_base_url(base_url: str, token: str) -> str:
-    client = await get_client()
-    resp = await client.post(
-        f"{base_url}webstream", headers=ICLOUD_HEADERS, data=ICLOUD_PAYLOAD, follow_redirects=False
-    )
-    if resp.status_code == 330:
-        try:
-            body = resp.json()
-            host = body.get("X-Apple-MMe-Host")
-            if not host:
-                raise ValueError("Missing X-Apple-MMe-Host in 330 response")
-            logging.info(f"Redirected to {host}")
-            return f"https://{host}/{token}/sharedstreams/"
-        except Exception as e:
-            logging.error(f"Redirect parsing failed: {e}")
-            raise
-    elif resp.status_code == 200:
-        return base_url
-    else:
-        resp.raise_for_status()
-
-async def post_json(path: str, base_url: str, payload: str) -> dict:
-    client = await get_client()
-    resp = await client.post(f"{base_url}{path}", headers=ICLOUD_HEADERS, data=payload)
-    resp.raise_for_status()
-    return resp.json()
-
-async def get_metadata(base_url: str) -> list:
-    data = await post_json("webstream", base_url, ICLOUD_PAYLOAD)
-    return data.get("photos", [])
-
-async def get_asset_urls(base_url: str, guids: list) -> dict:
-    payload = json.dumps({"photoGuids": guids})
-    data = await post_json("webasseturls", base_url, payload)
-    return data.get("items", {})
-
-@app.get("/album/{token}")
-async def get_album(token: str):
-    try:
-        base_url = await get_base_url(token)
-        base_url = await get_redirected_base_url(base_url, token)
-
-        metadata = await get_metadata(base_url)
-        guids = [photo["photoGuid"] for photo in metadata]
-        asset_map = await get_asset_urls(base_url, guids)
-
-        videos = []
-        for photo in metadata:
-            if photo.get("mediaAssetType", "").lower() != "video":
+from fastapi import FastAPI, HTTPException, Header
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Any, List
+from uuid import uuid4
+from datetime import datetime, timezone
+import asyncio
+
+app = FastAPI(title="Create3 Robot Command Bridge", version="1.0.0")
+
+# ====== Simple shared secret auth (optional but recommended) ======
+API_TOKEN = "change-me-please"  # set this to your own token
+
+def check_token(auth_header: Optional[str]):
+    if not API_TOKEN:
+        return
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing bearer token")
+    token = auth_header.split(" ", 1)[1].strip()
+    if token != API_TOKEN:
+        raise HTTPException(status_code=403, detail="Invalid token")
+
+# ====== In-memory command store ======
+# For HF Space demo/testing this is OK.
+# If you restart the app, queue clears.
+commands: Dict[str, Dict[str, Any]] = {}
+queue: List[str] = []
+queue_lock = asyncio.Lock()
+
+# ====== Models ======
+class CommandCreate(BaseModel):
+    command: str = Field(..., description="Robot command name")
+    args: Dict[str, Any] = Field(default_factory=dict, description="Command arguments")
+    source: Optional[str] = Field(default="ai", description="Who sent the command")
+    priority: int = Field(default=0, description="Higher = earlier in queue")
+    ttl_seconds: Optional[int] = Field(default=120, description="Optional expiration")
+
+class CommandStatusUpdate(BaseModel):
+    status: str = Field(..., description="queued | running | done | failed | expired")
+    result: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None
+    robot_id: Optional[str] = None
+
+# ====== Helpers ======
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+def command_expired(cmd: Dict[str, Any]) -> bool:
+    ttl = cmd.get("ttl_seconds")
+    if not ttl:
+        return False
+    created = datetime.fromisoformat(cmd["created_at"])
+    age = (datetime.now(timezone.utc) - created).total_seconds()
+    return age > ttl
+
+# ====== API ======
+
+@app.get("/")
+async def root():
+    return {"ok": True, "service": "create3-command-bridge"}
+
+@app.post("/commands")
+async def create_command(cmd: CommandCreate, authorization: Optional[str] = Header(default=None)):
+    check_token(authorization)
+
+    cmd_id = str(uuid4())
+    record = {
+        "id": cmd_id,
+        "command": cmd.command,
+        "args": cmd.args,
+        "source": cmd.source,
+        "priority": cmd.priority,
+        "ttl_seconds": cmd.ttl_seconds,
+        "status": "queued",
+        "result": None,
+        "error": None,
+        "created_at": now_iso(),
+        "updated_at": now_iso(),
+        "claimed_by": None,
+    }
+
+    async with queue_lock:
+        commands[cmd_id] = record
+
+        # Insert by priority (higher first)
+        inserted = False
+        for i, queued_id in enumerate(queue):
+            if record["priority"] > commands[queued_id]["priority"]:
+                queue.insert(i, cmd_id)
+                inserted = True
+                break
+        if not inserted:
+            queue.append(cmd_id)
+
+    return {"ok": True, "command_id": cmd_id, "status": "queued"}
+
+@app.get("/commands")
+async def list_commands(limit: int = 20, authorization: Optional[str] = Header(default=None)):
+    check_token(authorization)
+    # latest updated first
+    items = sorted(commands.values(), key=lambda x: x["updated_at"], reverse=True)[:limit]
+    return {"ok": True, "items": items}
+
+@app.get("/commands/{command_id}")
+async def get_command(command_id: str, authorization: Optional[str] = Header(default=None)):
+    check_token(authorization)
+    if command_id not in commands:
+        raise HTTPException(status_code=404, detail="Command not found")
+    return {"ok": True, "item": commands[command_id]}
+
+@app.post("/commands/next")
+async def claim_next_command(
+    robot_id: str,
+    authorization: Optional[str] = Header(default=None),
+):
+    """
+    Client polls this endpoint to claim the next command.
+    Returns one command and marks it running.
+    """
+    check_token(authorization)
+
+    async with queue_lock:
+        # Clean expired queued commands while scanning
+        while queue:
+            cmd_id = queue.pop(0)
+            cmd = commands.get(cmd_id)
+            if not cmd:
                 continue
-
-            derivatives = photo.get("derivatives", {})
-            best = max(
-                (d for k, d in derivatives.items() if k.lower() != "posterframe"),
-                key=lambda d: int(d.get("fileSize") or 0),
-                default=None
-            )
-            if not best:
+            if cmd["status"] != "queued":
                 continue
 
-            checksum = best.get("checksum")
-            info = asset_map.get(checksum)
-            if not info:
+            if command_expired(cmd):
+                cmd["status"] = "expired"
+                cmd["error"] = "TTL expired before execution"
+                cmd["updated_at"] = now_iso()
                 continue
-            video_url = f"https://{info['url_location']}{info['url_path']}"
-
-            poster = None
-            pf = derivatives.get("PosterFrame")
-            if pf:
-                pf_info = asset_map.get(pf.get("checksum"))
-                if pf_info:
-                    poster = f"https://{pf_info['url_location']}{pf_info['url_path']}"
-
-            videos.append({
-                "caption": photo.get("caption", ""),
-                "url": video_url,
-                "poster": poster or ""
-            })
-
-        return {"videos": videos}
-
-    except Exception as e:
-        logging.exception("Error in get_album")
-        return {"error": str(e)}
-
-@app.get("/album/{token}/raw")
-async def get_album_raw(token: str):
-    try:
-        base_url = await get_base_url(token)
-        base_url = await get_redirected_base_url(base_url, token)
-        metadata = await get_metadata(base_url)
-        guids = [photo["photoGuid"] for photo in metadata]
-        asset_map = await get_asset_urls(base_url, guids)
-        return {"metadata": metadata, "asset_urls": asset_map}
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
 
-    except Exception as e:
-        logging.exception("Error in get_album_raw")
-        return {"error": str(e)}
+            cmd["status"] = "running"
+            cmd["claimed_by"] = robot_id
+            cmd["updated_at"] = now_iso()
+            return {"ok": True, "item": cmd}
+
+    return {"ok": True, "item": None}
+
+@app.post("/commands/{command_id}/status")
+async def update_command_status(
+    command_id: str,
+    update: CommandStatusUpdate,
+    authorization: Optional[str] = Header(default=None),
+):
+    check_token(authorization)
+
+    cmd = commands.get(command_id)
+    if not cmd:
+        raise HTTPException(status_code=404, detail="Command not found")
+
+    # Basic state update
+    cmd["status"] = update.status
+    cmd["result"] = update.result
+    cmd["error"] = update.error
+    if update.robot_id:
+        cmd["claimed_by"] = update.robot_id
+    cmd["updated_at"] = now_iso()
+
+    return {"ok": True}
+
+@app.post("/commands/{command_id}/cancel")
+async def cancel_command(command_id: str, authorization: Optional[str] = Header(default=None)):
+    check_token(authorization)
+
+    cmd = commands.get(command_id)
+    if not cmd:
+        raise HTTPException(status_code=404, detail="Command not found")
+
+    if cmd["status"] in ("done", "failed", "expired"):
+        return {"ok": True, "status": cmd["status"], "message": "Already finished"}
+
+    cmd["status"] = "failed"
+    cmd["error"] = "Cancelled by operator"
+    cmd["updated_at"] = now_iso()
+
+    # Remove from queue if still queued
+    async with queue_lock:
+        if command_id in queue:
+            queue.remove(command_id)
+
+    return {"ok": True, "status": "failed"}
+
+@app.get("/health")
+async def health():
+    queued = sum(1 for c in commands.values() if c["status"] == "queued")
+    running = sum(1 for c in commands.values() if c["status"] == "running")
+    return {
+        "ok": True,
+        "queued": queued,
+        "running": running,
+        "total": len(commands),
+        "time": now_iso(),
+    }