Upload sync_hf.py

scripts/sync_hf.py

@@ -85,6 +85,7 @@ HF_TOKEN      = os.environ.get("HF_TOKEN")
 OPENCLAW_HOME = Path.home() / ".openclaw"
 APP_DIR       = Path(os.environ.get("OPENCLAW_APP_DIR", "/app/openclaw"))
 DATASET_PATH  = ".openclaw"
+SQLITE_SNAPSHOT_PATH = "sqlite-snapshots"
 
 # Z.AI API key (optional)
 ZAI_API_KEY   = os.environ.get("ZAI_API_KEY", "")
@@ -584,6 +585,7 @@ class OpenClawFullSync:
                             dest.parent.mkdir(parents=True, exist_ok=True)
                             shutil.copy2(str(item), str(dest))
                     print("[SYNC] ✓ Restore completed.")
+                    self._restore_sqlite_snapshots()
                 else:
                     print("[SYNC] Downloaded snapshot but dir not found. Starting fresh.")
 
@@ -715,12 +717,128 @@ class OpenClawFullSync:
             except Exception:
                 pass
 
+            self._upload_sqlite_snapshots()
+
         except Exception as e:
             print(f"[SYNC] ✗ Upload failed: {str(e)[:500]}")
             if "429" in str(e) or "rate limit" in str(e).lower():
                 self._cooldown_until = time.time() + 65 * 60
                 print("[SYNC] Hub rate limit hit — pausing uploads for 65 min")
 
+    # ── SQLite snapshots (live .sqlite files are never uploaded directly) ──
+
+    def _iter_persistable_sqlites(self):
+        for root, dirs, files in os.walk(OPENCLAW_HOME):
+            for fn in files:
+                if not fn.endswith(".sqlite"):
+                    continue
+                fp = Path(os.path.join(root, fn))
+                try:
+                    rel = fp.relative_to(OPENCLAW_HOME)
+                except ValueError:
+                    continue
+                if rel.as_posix().startswith("state/"):
+                    continue
+                yield rel, fp
+
+    def _stage_sqlite_snapshots(self, staging_dir):
+        staged = []
+        for rel, fp in self._iter_persistable_sqlites():
+            try:
+                if fp.stat().st_size == 0:
+                    continue
+                dest = Path(staging_dir) / rel
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                src = sqlite3.connect(str(fp))
+                dst = sqlite3.connect(str(dest))
+                try:
+                    src.backup(dst)
+                finally:
+                    dst.close()
+                    src.close()
+                staged.append(rel)
+            except Exception as e:
+                print(f"[SYNC] sqlite snapshot skipped for {rel}: {str(e)[:120]}")
+        return staged
+
+    def _upload_sqlite_snapshots(self):
+        staging = tempfile.mkdtemp(prefix="oc-sqlite-snap-")
+        try:
+            staged = self._stage_sqlite_snapshots(staging)
+            if not staged:
+                print("[SYNC] No sqlite snapshots to upload")
+                return
+            print(f"[SYNC] ▶ Uploading {len(staged)} sqlite snapshot(s) → "
+                  f"{SQLITE_SNAPSHOT_PATH}/ ...")
+            self.api.upload_folder(
+                folder_path=staging,
+                path_in_repo=SQLITE_SNAPSHOT_PATH,
+                repo_id=HF_REPO_ID,
+                repo_type="dataset",
+                token=HF_TOKEN,
+                commit_message=f"Sync sqlite snapshots — {datetime.now().isoformat()}",
+            )
+            print("[SYNC] ✓ Sqlite snapshots uploaded")
+        except Exception as e:
+            print(f"[SYNC] ✗ Sqlite snapshot upload failed: {str(e)[:300]}")
+            if "429" in str(e) or "rate limit" in str(e).lower():
+                self._cooldown_until = time.time() + 65 * 60
+                print("[SYNC] Hub rate limit hit — pausing uploads for 65 min")
+        finally:
+            shutil.rmtree(staging, ignore_errors=True)
+
+    def _restore_sqlite_snapshots(self):
+        try:
+            files = self.api.list_repo_files(repo_id=HF_REPO_ID, repo_type="dataset")
+            snap_files = [f for f in files if f.startswith(f"{SQLITE_SNAPSHOT_PATH}/")]
+            if not snap_files:
+                return
+            print(f"[SYNC] Found {len(snap_files)} sqlite snapshot(s) in dataset")
+            with tempfile.TemporaryDirectory() as tmpdir:
+                snapshot_download(
+                    repo_id=HF_REPO_ID,
+                    repo_type="dataset",
+                    allow_patterns=f"{SQLITE_SNAPSHOT_PATH}/**",
+                    local_dir=tmpdir,
+                    token=HF_TOKEN,
+                )
+                snap_root = Path(tmpdir) / SQLITE_SNAPSHOT_PATH
+                if not snap_root.exists():
+                    return
+                restored = 0
+                for item in snap_root.rglob("*"):
+                    if not item.is_file():
+                        continue
+                    rel = item.relative_to(snap_root)
+                    dest = OPENCLAW_HOME / rel
+                    try:
+                        conn = sqlite3.connect(str(item))
+                        try:
+                            ok = conn.execute("PRAGMA quick_check").fetchone()
+                        finally:
+                            conn.close()
+                    except Exception as e:
+                        print(f"[SYNC] Snapshot integrity check failed, skipped: "
+                              f"{rel} ({str(e)[:80]})")
+                        continue
+                    if not ok or ok[0] != "ok":
+                        print(f"[SYNC] Snapshot corrupt, skipped: {rel}")
+                        continue
+                    dest.parent.mkdir(parents=True, exist_ok=True)
+                    for suf in ("", "-wal", "-shm", "-journal"):
+                        old = Path(str(dest) + suf)
+                        if old.exists():
+                            try:
+                                old.unlink()
+                            except OSError:
+                                pass
+                    shutil.copy2(str(item), str(dest))
+                    restored += 1
+                if restored:
+                    print(f"[SYNC] ✓ Restored {restored} sqlite snapshot(s)")
+        except Exception as e:
+            print(f"[SYNC] sqlite snapshot restore failed: {str(e)[:300]}")
+
     # ── Pre-gateway legacy cleanup (CRITICAL) ─────────────────────────
 
     def _retire_workspace_markers(self):