File size: 4,418 Bytes
5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d 5bd484d 6825f4d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | import os
import sys
import tarfile
import tempfile
from pathlib import Path
from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.utils import EntryNotFoundError, RepositoryNotFoundError, HfHubHTTPError
api = HfApi()
REPO_ID = os.getenv("HF_DATASET")
TOKEN = os.getenv("HF_TOKEN")
FILENAME = "latest_backup.tar.gz"
BASE_DIR = Path("/root/.openclaw")
PATHS_TO_BACKUP = [
BASE_DIR / "sessions",
BASE_DIR / "agents" / "main" / "sessions",
BASE_DIR / "openclaw.json",
]
def log(msg: str) -> None:
print(msg, flush=True)
def log_err(msg: str) -> None:
print(msg, file=sys.stderr, flush=True)
def is_subpath(child: Path, parent: Path) -> bool:
try:
child.resolve().relative_to(parent.resolve())
return True
except ValueError:
return False
def safe_extract(tar: tarfile.TarFile, target_dir: Path) -> None:
target_dir = target_dir.resolve()
for member in tar.getmembers():
member_path = target_dir / member.name
if not is_subpath(member_path, target_dir):
raise RuntimeError(f"Unsafe path detected in archive: {member.name}")
tar.extractall(path=target_dir)
def restore() -> bool:
if not REPO_ID or not TOKEN:
log("[RESTORE] Skip: HF_DATASET or HF_TOKEN not set")
return False
try:
log(f"[RESTORE] Downloading {FILENAME} from dataset repo: {REPO_ID}")
path = hf_hub_download(
repo_id=REPO_ID,
filename=FILENAME,
repo_type="dataset",
token=TOKEN,
)
BASE_DIR.mkdir(parents=True, exist_ok=True)
with tarfile.open(path, "r:gz") as tar:
safe_extract(tar, BASE_DIR)
log(f"[RESTORE] Success: restored from {FILENAME}")
return True
except EntryNotFoundError:
log(f"[RESTORE] Note: {FILENAME} not found in repo, probably first run")
return False
except RepositoryNotFoundError:
log_err(f"[RESTORE] Error: dataset repo not found: {REPO_ID}")
return False
except HfHubHTTPError as e:
log_err(f"[RESTORE] Hub HTTP error: {e}")
return False
except tarfile.TarError as e:
log_err(f"[RESTORE] Invalid tar archive: {e}")
return False
except Exception as e:
log_err(f"[RESTORE] Unexpected error: {e}")
return False
def backup() -> bool:
if not REPO_ID or not TOKEN:
log("[BACKUP] Skip: HF_DATASET or HF_TOKEN not set")
return False
existing_paths = [p for p in PATHS_TO_BACKUP if p.exists()]
if not existing_paths:
log("[BACKUP] Skip: no paths to backup")
return False
temp_path = None
try:
with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp:
temp_path = Path(tmp.name)
with tarfile.open(temp_path, "w:gz") as tar:
for p in existing_paths:
if p.is_relative_to(BASE_DIR):
arcname = p.relative_to(BASE_DIR)
else:
arcname = p.name
tar.add(str(p), arcname=str(arcname))
log(f"[BACKUP] Added: {p} -> {arcname}")
api.upload_file(
path_or_fileobj=str(temp_path),
path_in_repo=FILENAME,
repo_id=REPO_ID,
repo_type="dataset",
token=TOKEN,
commit_message=f"Update {FILENAME}",
)
log(f"[BACKUP] Success: uploaded {FILENAME}")
return True
except RepositoryNotFoundError:
log_err(f"[BACKUP] Error: dataset repo not found: {REPO_ID}")
return False
except HfHubHTTPError as e:
log_err(f"[BACKUP] Hub HTTP error: {e}")
return False
except tarfile.TarError as e:
log_err(f"[BACKUP] Tar error: {e}")
return False
except Exception as e:
log_err(f"[BACKUP] Unexpected error: {e}")
return False
finally:
if temp_path and temp_path.exists():
try:
temp_path.unlink()
except Exception as e:
log_err(f"[BACKUP] Warning: failed to delete temp file {temp_path}: {e}")
if __name__ == "__main__":
action = sys.argv[1].strip().lower() if len(sys.argv) > 1 else "restore"
if action == "backup":
ok = backup()
sys.exit(0 if ok else 1)
else:
ok = restore()
sys.exit(0 if ok else 1) |