Upload 20 files

routers/backup.py

@@ -1,9 +1,9 @@
 """
-Backup & Restore API - proxies through the Admin Worker (Cloudflare Worker).
+Backup & Restore API — talks directly to HuggingFace Dataset API.
 
-The Admin Worker handles HuggingFace Dataset storage with per-user folders.
-This router creates local tar.gz archives and sends them to the Worker,
-or downloads archives from the Worker and extracts them locally.
+Flow:
+  1. Space asks Worker for HF credentials (token, repo, user path prefix)
+  2. Space uploads/downloads/lists archives directly via HuggingFace API
 
 Requires env var:
   ADMIN_API_URL - URL of the Cloudflare Worker admin API
@@ -23,15 +23,7 @@ from storage import load_meta, save_meta, validate_zone_name
 
 router = APIRouter(prefix="/api/backup", tags=["backup"])
 
-
-def _worker_headers(token: str) -> dict:
-    """Build headers for Worker API calls."""
-    return {"Authorization": f"Bearer {token}"}
-
-
-def _url(path: str) -> str:
-    """Build the full Worker URL for a given path."""
-    return f"{ADMIN_API_URL}{path}"
+HF_API = "https://huggingface.co/api"
 
 
 def _get_token(request: Request) -> str:
@@ -42,6 +34,32 @@ def _get_token(request: Request) -> str:
     return ""
 
 
+def _get_credentials(token: str) -> dict:
+    """Get HF credentials from Worker. Returns {hf_token, repo, path_prefix}."""
+    with httpx.Client(timeout=15) as client:
+        resp = client.get(
+            f"{ADMIN_API_URL}/backup/credentials",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        if resp.status_code != 200:
+            data = resp.json() if "application/json" in resp.headers.get("content-type", "") else {"error": resp.text}
+            raise ValueError(data.get("error", f"Worker error: {resp.status_code}"))
+        return resp.json()
+
+
+def _log_action(token: str, zone_name: str, action: str, status: str, file_path: str = ""):
+    """Log backup/restore action to Worker (best-effort)."""
+    try:
+        with httpx.Client(timeout=10) as client:
+            client.post(
+                f"{ADMIN_API_URL}/backup/log",
+                headers={"Authorization": f"Bearer {token}"},
+                json={"zone_name": zone_name, "action": action, "status": status, "file_path": file_path},
+            )
+    except Exception:
+        pass
+
+
 def _create_zone_archive(zone_name: str) -> Path:
     """Create a tar.gz archive of a zone directory."""
     zone_path = DATA_DIR / zone_name
@@ -77,14 +95,47 @@ async def list_backups(request: Request):
     if not token:
         raise HTTPException(401, "Chua dang nhap")
     try:
+        creds = _get_credentials(token)
         async with httpx.AsyncClient(timeout=30) as client:
-            resp = await client.get(_url("/backup/list"), headers=_worker_headers(token))
+            resp = await client.get(
+                f"{HF_API}/datasets/{creds['repo']}/tree/main/{creds['path_prefix']}",
+                headers={"Authorization": f"Bearer {creds['hf_token']}"},
+            )
+            if resp.status_code == 404:
+                return []
             if resp.status_code != 200:
-                data = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {"error": resp.text}
-                raise HTTPException(resp.status_code, data.get("error", "Worker error"))
-            return resp.json()
+                raise HTTPException(502, f"HF API error: {resp.status_code} {resp.text}")
+            tree = resp.json()
+            return [
+                {
+                    "zone_name": f["path"].split("/")[-1].replace(".tar.gz", ""),
+                    "file": f["path"],
+                    "size": (f.get("lfs") or {}).get("size") or f.get("size", 0),
+                }
+                for f in tree
+                if f.get("type") == "file" and f["path"].endswith(".tar.gz")
+            ]
+    except ValueError as e:
+        raise HTTPException(502, str(e))
     except httpx.HTTPError as e:
-        raise HTTPException(502, f"Khong the ket noi Worker: {e}")
+        raise HTTPException(502, f"Khong the ket noi: {e}")
+
+
+def _upload_to_hf(creds: dict, zone_name: str, archive_path: Path):
+    """Upload archive directly to HuggingFace Dataset API."""
+    file_path = f"{creds['path_prefix']}/{zone_name}.tar.gz"
+    hf_token = creds["hf_token"]
+    repo = creds["repo"]
+
+    with httpx.Client(timeout=300) as client:
+        with open(archive_path, "rb") as f:
+            resp = client.post(
+                f"{HF_API}/datasets/{repo}/upload/main/{file_path}",
+                headers={"Authorization": f"Bearer {hf_token}"},
+                content=f.read(),
+            )
+        if resp.status_code not in (200, 201):
+            raise ValueError(f"HF upload error: {resp.status_code} {resp.text}")
 
 
 @router.post("/zone/{zone_name}")
@@ -103,6 +154,12 @@ async def backup_zone(zone_name: str, request: Request, background_tasks: Backgr
     if _backup_status["running"]:
         raise HTTPException(409, "Dang co backup khac dang chay")
 
+    # Fetch credentials before background task (validates JWT now)
+    try:
+        creds = _get_credentials(token)
+    except ValueError as e:
+        raise HTTPException(502, str(e))
+
     def _run():
         _backup_status["running"] = True
         _backup_status["error"] = None
@@ -110,22 +167,17 @@ async def backup_zone(zone_name: str, request: Request, background_tasks: Backgr
         try:
             archive_path = _create_zone_archive(zone_name)
             try:
-                with httpx.Client(timeout=300) as client:
-                    with open(archive_path, "rb") as f:
-                        resp = client.post(
-                            _url(f"/backup/upload/{zone_name}"),
-                            headers={**_worker_headers(token), "Content-Type": "application/octet-stream"},
-                            content=f.read(),
-                        )
-                    if resp.status_code != 200:
-                        raise ValueError(f"Worker error: {resp.text}")
+                _upload_to_hf(creds, zone_name, archive_path)
             finally:
                 archive_path.unlink(missing_ok=True)
+            _log_action(token, zone_name, "backup", "success",
+                        f"{creds['path_prefix']}/{zone_name}.tar.gz")
             _backup_status["last"] = datetime.now().isoformat()
             _backup_status["progress"] = f"Backup zone {zone_name} thanh cong"
         except Exception as e:
             _backup_status["error"] = str(e)
             _backup_status["progress"] = f"Loi backup: {e}"
+            _log_action(token, zone_name, "backup", "error")
         finally:
             _backup_status["running"] = False
 
@@ -143,6 +195,11 @@ async def backup_all(request: Request, background_tasks: BackgroundTasks):
     if _backup_status["running"]:
         raise HTTPException(409, "Dang co backup khac dang chay")
 
+    try:
+        creds = _get_credentials(token)
+    except ValueError as e:
+        raise HTTPException(502, str(e))
+
     def _run():
         _backup_status["running"] = True
         _backup_status["error"] = None
@@ -158,17 +215,11 @@ async def backup_all(request: Request, background_tasks: BackgroundTasks):
                 _backup_status["progress"] = f"Dang backup zone {zone_name} ({done + 1}/{total})..."
                 archive_path = _create_zone_archive(zone_name)
                 try:
-                    with httpx.Client(timeout=300) as client:
-                        with open(archive_path, "rb") as f:
-                            resp = client.post(
-                                _url(f"/backup/upload/{zone_name}"),
-                                headers={**_worker_headers(token), "Content-Type": "application/octet-stream"},
-                                content=f.read(),
-                            )
-                        if resp.status_code != 200:
-                            raise ValueError(f"Worker error for {zone_name}: {resp.text}")
+                    _upload_to_hf(creds, zone_name, archive_path)
                 finally:
                     archive_path.unlink(missing_ok=True)
+                _log_action(token, zone_name, "backup", "success",
+                            f"{creds['path_prefix']}/{zone_name}.tar.gz")
                 done += 1
             _backup_status["last"] = datetime.now().isoformat()
             _backup_status["progress"] = "Backup tat ca zones thanh cong"
@@ -182,6 +233,22 @@ async def backup_all(request: Request, background_tasks: BackgroundTasks):
     return {"ok": True, "message": "Dang backup tat ca zones trong nen..."}
 
 
+def _download_from_hf(creds: dict, zone_name: str) -> bytes:
+    """Download archive directly from HuggingFace Dataset."""
+    file_path = f"{creds['path_prefix']}/{zone_name}.tar.gz"
+    with httpx.Client(timeout=300) as client:
+        resp = client.get(
+            f"https://huggingface.co/datasets/{creds['repo']}/resolve/main/{file_path}",
+            headers={"Authorization": f"Bearer {creds['hf_token']}"},
+            follow_redirects=True,
+        )
+        if resp.status_code == 404:
+            raise FileNotFoundError(f"Backup zone '{zone_name}' khong ton tai")
+        if resp.status_code != 200:
+            raise ValueError(f"HF download error: {resp.status_code}")
+        return resp.content
+
+
 @router.post("/restore/{zone_name}")
 async def restore_zone(zone_name: str, request: Request, background_tasks: BackgroundTasks):
     if not ADMIN_API_URL:
@@ -196,19 +263,19 @@ async def restore_zone(zone_name: str, request: Request, background_tasks: Backg
     if _backup_status["running"]:
         raise HTTPException(409, "Dang co backup/restore khac dang chay")
 
+    try:
+        creds = _get_credentials(token)
+    except ValueError as e:
+        raise HTTPException(502, str(e))
+
     def _run():
         _backup_status["running"] = True
         _backup_status["error"] = None
         _backup_status["progress"] = f"Dang restore zone: {zone_name}..."
         try:
-            with httpx.Client(timeout=300) as client:
-                resp = client.get(_url(f"/backup/download/{zone_name}"), headers=_worker_headers(token))
-                if resp.status_code == 404:
-                    raise ValueError(f"Backup zone '{zone_name}' khong ton tai")
-                if resp.status_code != 200:
-                    raise ValueError(f"Worker error: {resp.text}")
-                archive_path = BACKUP_DIR / f"{zone_name}.tar.gz"
-                archive_path.write_bytes(resp.content)
+            data = _download_from_hf(creds, zone_name)
+            archive_path = BACKUP_DIR / f"{zone_name}.tar.gz"
+            archive_path.write_bytes(data)
 
             try:
                 zone_path = DATA_DIR / zone_name
@@ -229,11 +296,14 @@ async def restore_zone(zone_name: str, request: Request, background_tasks: Backg
             finally:
                 archive_path.unlink(missing_ok=True)
 
+            _log_action(token, zone_name, "restore", "success",
+                        f"{creds['path_prefix']}/{zone_name}.tar.gz")
             _backup_status["last"] = datetime.now().isoformat()
             _backup_status["progress"] = f"Restore zone {zone_name} thanh cong"
         except Exception as e:
             _backup_status["error"] = str(e)
             _backup_status["progress"] = f"Loi restore: {e}"
+            _log_action(token, zone_name, "restore", "error")
         finally:
             _backup_status["running"] = False
 
@@ -251,31 +321,47 @@ async def restore_all(request: Request, background_tasks: BackgroundTasks):
     if _backup_status["running"]:
         raise HTTPException(409, "Dang co backup/restore khac dang chay")
 
+    try:
+        creds = _get_credentials(token)
+    except ValueError as e:
+        raise HTTPException(502, str(e))
+
     def _run():
         _backup_status["running"] = True
         _backup_status["error"] = None
         _backup_status["progress"] = "Dang restore tat ca zones..."
         try:
+            # List backups from HF directly
             with httpx.Client(timeout=30) as client:
-                resp = client.get(_url("/backup/list"), headers=_worker_headers(token))
-                if resp.status_code != 200:
-                    raise ValueError(f"Khong the lay danh sach backup: {resp.text}")
-                backup_list = resp.json()
-
-            total = len(backup_list)
+                resp = client.get(
+                    f"{HF_API}/datasets/{creds['repo']}/tree/main/{creds['path_prefix']}",
+                    headers={"Authorization": f"Bearer {creds['hf_token']}"},
+                )
+            if resp.status_code == 404:
+                _backup_status["progress"] = "Khong co backup nao"
+                return
+            if resp.status_code != 200:
+                raise ValueError(f"HF API error: {resp.status_code}")
+
+            tree = resp.json()
+            backup_files = [
+                f for f in tree
+                if f.get("type") == "file" and f["path"].endswith(".tar.gz")
+            ]
+            total = len(backup_files)
             done = 0
-            for b in backup_list:
-                zone_name = b["zone_name"]
-                _backup_status["progress"] = f"Dang restore zone {zone_name} ({done + 1}/{total})..."
-                with httpx.Client(timeout=300) as client:
-                    resp = client.get(_url(f"/backup/download/{zone_name}"), headers=_worker_headers(token))
-                    if resp.status_code != 200:
-                        continue
-                    archive_path = BACKUP_DIR / f"{zone_name}.tar.gz"
-                    archive_path.write_bytes(resp.content)
+            for bf in backup_files:
+                zn = bf["path"].split("/")[-1].replace(".tar.gz", "")
+                _backup_status["progress"] = f"Dang restore zone {zn} ({done + 1}/{total})..."
+                try:
+                    data = _download_from_hf(creds, zn)
+                except FileNotFoundError:
+                    continue
+                archive_path = BACKUP_DIR / f"{zn}.tar.gz"
+                archive_path.write_bytes(data)
 
                 try:
-                    zone_path = DATA_DIR / zone_name
+                    zone_path = DATA_DIR / zn
                     if zone_path.exists():
                         shutil.rmtree(zone_path)
                     with tarfile.open(archive_path, "r:gz") as tar:
@@ -283,12 +369,12 @@ async def restore_all(request: Request, background_tasks: BackgroundTasks):
                             member_path = os.path.normpath(member.name)
                             if member_path.startswith("..") or os.path.isabs(member_path):
                                 raise ValueError(f"Archive chua path khong an toan: {member.name}")
-                            if not member_path.startswith(zone_name):
+                            if not member_path.startswith(zn):
                                 raise ValueError(f"Archive chua path ngoai zone: {member.name}")
                         tar.extractall(path=str(DATA_DIR), filter="data")
                     meta = load_meta()
-                    if zone_name not in meta:
-                        meta[zone_name] = {"description": "", "created": datetime.now().isoformat()}
+                    if zn not in meta:
+                        meta[zn] = {"description": "", "created": datetime.now().isoformat()}
                         save_meta(meta)
                 finally:
                     archive_path.unlink(missing_ok=True)