#!/usr/bin/env python3 """ backup-manager.py WebDAV + HF Dataset / : WebDAV + BACKUP_WEBDAV_INTERVAL WEBDAV_MAX_BACKUPS ( 60 ) ( 30 ) HF Dataset BACKUP_HF_INTERVAL_SMALL 1 () ( 30 ) HF Dataset BACKUP_HF_INTERVAL_BIG DATASET_MAX_BACKUPS ( 30 ) ( 30 ) : HF Dataset 1GB, : WebDAV HF Dataset : last_run , """ import hashlib, json, os, tarfile, time, sys from datetime import datetime, timezone, timedelta from pathlib import Path import requests # # # STATE_DIR = os.environ.get("OPENCLAW_STATE_DIR", "/root/.openclaw") # WebDAV WEBDAV_URL = os.environ.get("WEBDAV_URL", "").rstrip("/") WEBDAV_USER = os.environ.get("WEBDAV_USERNAME", "") WEBDAV_PASS = os.environ.get("WEBDAV_PASSWORD", "") WEBDAV_BACKUP_DIR = os.environ.get("WEBDAV_BACKUP_DIR", "openclaw-backup") WEBDAV_INTERVAL = int(os.environ.get("BACKUP_WEBDAV_INTERVAL", "60")) WEBDAV_MAX_BACKUPS = int(os.environ.get("WEBDAV_MAX_BACKUPS", "30")) # HF Dataset HF_REPO = os.environ.get("HF_DATASET", "") HF_TOKEN = os.environ.get("HF_TOKEN", "") HF_SMALL_INTERVAL = int(os.environ.get("BACKUP_HF_INTERVAL_SMALL", "30")) HF_BIG_INTERVAL = int(os.environ.get("BACKUP_HF_INTERVAL_BIG", "30")) DATASET_MAX_BACKUPS = int(os.environ.get("DATASET_MAX_BACKUPS", "30")) # FULL_NAME = "openclaw-full.tar.gz" MANIFEST_NAME = "_incremental_manifest.json" # HF MAX_HF_STORAGE = 1 * 1024 * 1024 * 1024 # 1 GB # HASH_CHUNK_SIZE = 65536 # SHA256 WD_CHECK_TIMEOUT = 3 # WebDAV WD_REQ_TIMEOUT = 30 # WebDAV # RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" RESET = "\033[0m" # # # def _print_banner(): """""" global WD_SKIPPED, HF_SKIPPED sep = "" * 50 print(f"[backup] {sep}") wd_configured = _wd_is_configured() if not wd_configured: print(f"[backup] WebDAV {RED} : WEBDAV_URL / WEBDAV_USERNAME / WEBDAV_PASSWORD / WEBDAV_BACKUP_DIR {RESET}") WD_SKIPPED = True else: print(f"[backup] WebDAV {GREEN} {RESET} ={WEBDAV_INTERVAL} ={WEBDAV_MAX_BACKUPS}") WD_SKIPPED = False if not HF_REPO or not HF_TOKEN: hf_missing = [] if not HF_REPO: hf_missing.append("HF_DATASET") if not HF_TOKEN: hf_missing.append("HF_TOKEN") print(f"[backup] HF {RED} : {', '.join(hf_missing)}{RESET}") print(f"[backup] HF {RED} {RESET}") HF_SKIPPED = True else: print(f"[backup] HF {GREEN} {RESET} ={HF_SMALL_INTERVAL}") print(f"[backup] HF {GREEN} {RESET} ={HF_BIG_INTERVAL} ={DATASET_MAX_BACKUPS}") HF_SKIPPED = False print(f"[backup] {sep}") # banner import _DISPLAYED_BANNER = False def _ensure_banner(): global _DISPLAYED_BANNER if not _DISPLAYED_BANNER: _print_banner() _DISPLAYED_BANNER = True # # # def _create_tar() -> str: """ STATE_DIR tar.gz, .""" tarpath = f"/tmp/openclaw-backup-{int(time.time()*1000)}.tar.gz" try: with tarfile.open(tarpath, "w:gz") as tar: root = Path(STATE_DIR) if root.exists(): for item in root.iterdir(): if item.exists(): name = item.name if name.startswith("_last_") or name.endswith(".tmp") or name == ".lock": continue try: tar.add(str(item), arcname=name) except (PermissionError, FileNotFoundError) as e: print(f"[backup] {name}: {e}") continue else: print(f"[backup] STATE_DIR {STATE_DIR} , ") root.mkdir(parents=True, exist_ok=True) size = os.path.getsize(tarpath) print(f"[backup] Archive created ({size/1024/1024:.1f} MB)") except (OSError, tarfile.TarError) as e: print(f"[backup] : {e}") if os.path.exists(tarpath): os.remove(tarpath) raise # return tarpath def _file_hash(path: str) -> str: """ SHA256 , .""" try: h = hashlib.sha256() with open(path, "rb") as f: while True: chunk = f.read(HASH_CHUNK_SIZE) if not chunk: break h.update(chunk) return h.hexdigest() except (OSError, PermissionError, FileNotFoundError) as e: print(f"[backup] : {path} {e}") return "" def _today_str() -> str: return time.strftime("%Y%m%d") # # WebDAV HTTP # def _wd_auth(): return (WEBDAV_USER, WEBDAV_PASS) if WEBDAV_USER else None def _wd_url(path=""): return f"{WEBDAV_URL}/{WEBDAV_BACKUP_DIR}/{path.lstrip('/')}" def _wd_req(method, path="", **kwargs): url = _wd_url(path) kwargs.setdefault("timeout", WD_REQ_TIMEOUT) # / resp = requests.request(method, url, auth=_wd_auth(), **kwargs) resp.raise_for_status() return resp def _wd_check(): """ WebDAV (3s ), False""" try: url = _wd_url("") resp = requests.request("PROPFIND", url, auth=_wd_auth(), timeout=3) resp.raise_for_status() # 401/403/404 return True except Exception: return False def wd_exists(path): try: _wd_req("PROPFIND", path) return True except Exception: return False def wd_upload(path, data): return _wd_req("PUT", path, data=data) def wd_download(path): return _wd_req("GET", path).content def wd_delete(path): """ WebDAV """ try: _wd_req("DELETE", path) return True except Exception as e: print(f"[backup] WebDAV : {path} {e}") return False def wd_mkdir(parts): """ WebDAV """ try: _wd_req("MKCOL", "") except Exception: pass for i in range(1, len(parts) + 1): p = "/".join(parts[:i]) try: _wd_req("MKCOL", p) except Exception: pass def wd_list(path=""): """ WebDAV """ try: import xml.etree.ElementTree as ET resp = requests.request( "PROPFIND", _wd_url(path), auth=_wd_auth(), timeout=3, headers={"Depth": "1"}, ) resp.raise_for_status() root = ET.fromstring(resp.content) ns = {"d": "DAV:"} items = [] for el in root.findall(".//d:response", ns): href_el = el.find("d:href", ns) if href_el is not None and href_el.text: name = href_el.text.rstrip("/").split("/")[-1] if name: items.append(name) return [x for x in items if x != WEBDAV_BACKUP_DIR.split("/")[-1]] except Exception as e: print(f"[backup] WebDAV : {e}") return [] # # WebDAV (Manifest) # def _load_manifest() -> dict: try: data = wd_download(MANIFEST_NAME) return json.loads(data) except Exception: return {} def _save_manifest(manifest: dict): wd_upload(MANIFEST_NAME, json.dumps(manifest, indent=2).encode()) def _scan_and_upload_changes(root: Path, manifest: dict) -> int: """ WebDAV. .""" changed = 0 for fpath in root.rglob("*"): if not fpath.is_file(): continue rel = str(fpath.relative_to(root)) if rel == MANIFEST_NAME or rel.startswith("_incremental"): continue basename = fpath.name if basename in (".lock", ".pid", ".tmp") or basename.endswith(".log"): continue if "node_modules" in rel: continue if rel.startswith("_last_"): continue cur_h = _file_hash(str(fpath)) prev = manifest.get(rel, {}) if cur_h != prev.get("sha256"): try: parts = ["files", *rel.split("/")] wd_mkdir(parts[:-1]) wd_upload("/".join(parts), fpath.read_bytes()) manifest[rel] = { "sha256": cur_h, "mtime": fpath.stat().st_mtime, "size": fpath.stat().st_size, } changed += 1 except Exception as e: err_str = str(e) # 409 = if "409" in err_str or "Conflict" in err_str: manifest[rel] = { "sha256": cur_h, "mtime": fpath.stat().st_mtime, "size": fpath.stat().st_size, } changed += 1 else: print(f"[backup] (): {rel} {e}") return changed def incremental_backup() -> int: """, WebDAV. .""" root = Path(STATE_DIR) if not root.exists(): return 0 # WebDAV if not _wd_is_configured(): print(f"[backup] WebDAV , ") return 0 # WebDAV if not _wd_check(): print(f"[backup] WebDAV , ") return 0 manifest = _load_manifest() changed = _scan_and_upload_changes(root, manifest) if changed: try: _save_manifest(manifest) except Exception as e: print(f"[backup] Manifest : {e}") return changed # # WebDAV + # def _wd_cleanup(): """ WEBDAV_MAX_BACKUPS WebDAV """ files = wd_list("") now = datetime.now() deleted = 0 for f in files: if not f.startswith("openclaw-full-") or not f.endswith(".tar.gz"): continue if f == FULL_NAME: continue # try: date_str = f.replace("openclaw-full-", "").replace(".tar.gz", "") dt = datetime.strptime(date_str, "%Y%m%d") days_old = (now - dt).days if days_old > WEBDAV_MAX_BACKUPS: if wd_delete(f): print(f"[backup] WebDAV : {f} ({days_old})") deleted += 1 except ValueError: continue if deleted: print(f"[backup] WebDAV {deleted} ") return deleted def _wd_is_configured() -> bool: """ WebDAV 4 """ return bool(WEBDAV_URL and WEBDAV_USER and WEBDAV_PASS and WEBDAV_BACKUP_DIR) def full_backup(): """ tar WebDAV (: openclaw-full-YYYYMMDD.tar.gz)""" tarpath = _create_tar() if _wd_is_configured(): # WebDAV if not _wd_check(): print(f"[backup] WebDAV , ") os.remove(tarpath) return wd_mkdir([]) filename = f"openclaw-full-{_today_str()}.tar.gz" try: with open(tarpath, "rb") as f: wd_upload(filename, f.read()) print(f"[backup] WebDAV {filename}") # with open(tarpath, "rb") as f: wd_upload(FULL_NAME, f.read()) # _wd_cleanup() except Exception as e: print(f"[backup] WebDAV : {e}") os.remove(tarpath) # # HF Dataset # def _hf_list_backup_files() -> dict | None: """ HF Dataset . : {: (bytes)} {} None API """ if not HF_REPO or not HF_TOKEN: return {} try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() # timeout backups = {} for entry in api.list_repo_tree(HF_REPO, repo_type="dataset", token=HF_TOKEN): name = entry.rfilename if name.startswith("openclaw-full") and name.endswith(".tar.gz"): backups[name] = entry.size if hasattr(entry, "size") else 0 return backups except Exception as e: print(f"[backup] HF Dataset list API : {e}") return None # "" ( {}) def hf_small_backup(): """: tar.gz openclaw-full.tar.gz (), """ if not HF_REPO or not HF_TOKEN: return tarpath = _create_tar() tar_size = os.path.getsize(tarpath) # backups = _hf_list_backup_files() if backups is None: # API print(f"[backup] HF: Dataset API ") os.remove(tarpath) return existing_total = sum(backups.values()) if existing_total + tar_size > MAX_HF_STORAGE: print(f"{RED}[backup] HF: 1GB! " f"( {existing_total/1024/1024:.1f}MB + " f" {tar_size/1024/1024:.1f}MB > 1GB){RESET}") os.remove(tarpath) return try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() # timeout with open(tarpath, "rb") as f: api.upload_file( path_or_fileobj=f, path_in_repo=FULL_NAME, repo_id=HF_REPO, repo_type="dataset", token=HF_TOKEN, ) print(f"[backup] HF {FULL_NAME}") except Exception as e: print(f"{RED}[backup] HF: {e}{RESET}") finally: if os.path.exists(tarpath): os.remove(tarpath) # # HF Dataset + # def _hf_cleanup(): """ DATASET_MAX_BACKUPS HF """ backups = _hf_list_backup_files() if not backups: # None(API) {} () return 0 from huggingface_hub import HfApi api = HfApi(timeout=60) now = datetime.now() deleted = 0 for name in backups: if name == FULL_NAME or not name.endswith(".tar.gz"): continue try: date_str = name.replace("openclaw-full-", "").replace(".tar.gz", "") dt = datetime.strptime(date_str, "%Y%m%d") days_old = (now - dt).days if days_old > DATASET_MAX_BACKUPS: api.delete_file( path_in_repo=name, repo_id=HF_REPO, repo_type="dataset", token=HF_TOKEN, ) print(f"[backup] HF : {name} ({days_old})") deleted += 1 except (ValueError, Exception): continue if deleted: print(f"[backup] HF {deleted} ") return deleted def _hf_make_space(backups: dict, tar_size: int, api) -> bool: """, True, False.""" print(f"[backup] HF " f"( {sum(backups.values())/1024/1024:.1f}MB + " f" {tar_size/1024/1024:.1f}MB > 1GB)") print(f"[backup] ...") dated_big = [] for name, sz in backups.items(): if name == FULL_NAME: continue try: ds = name.replace("openclaw-full-", "").replace(".tar.gz", "") _ = datetime.strptime(ds, "%Y%m%d") dated_big.append((ds, name)) except ValueError: continue dated_big.sort() for ds, name in dated_big: try: api.delete_file( path_in_repo=name, repo_id=HF_REPO, repo_type="dataset", token=HF_TOKEN, ) print(f"[backup] {name} ( {ds})") del backups[name] if sum(backups.values()) + tar_size <= MAX_HF_STORAGE: return True except Exception: continue print(f"{RED}[backup] HF: 1GB " f"( {tar_size/1024/1024:.1f}MB > 1GB){RESET}") return False def _hf_upload_backup(tarpath: str, tar_size: int, filename: str, api) -> None: """ HF Dataset.""" # & backups = _hf_list_backup_files() if backups is None: # API print(f"[backup] HF: Dataset API ") os.remove(tarpath) return existing_total = sum(backups.values()) if existing_total + tar_size > MAX_HF_STORAGE: if not _hf_make_space(backups, tar_size, api): os.remove(tarpath) return print(f"[backup] , ") backups = _hf_list_backup_files() # if backups is None: backups = {} # , if filename in backups: try: api.delete_file( path_in_repo=filename, repo_id=HF_REPO, repo_type="dataset", token=HF_TOKEN, ) except Exception: print(f"[backup] (): {filename}") # with open(tarpath, "rb") as f: api.upload_file( path_or_fileobj=f, path_in_repo=filename, repo_id=HF_REPO, repo_type="dataset", token=HF_TOKEN, ) print(f"[backup] HF {filename}") # _hf_cleanup() def hf_large_backup(): """ : tar.gz openclaw-full-YYYYMMDD.tar.gz , """ if not HF_REPO or not HF_TOKEN: return today = _today_str() filename = f"openclaw-full-{today}.tar.gz" # Skip if today's backup already exists (prevents re-upload on restart) existing = _hf_list_backup_files() if existing and filename in existing: print(f"[backup] Today's large backup already exists ({filename}), skipping") return tarpath = _create_tar() tar_size = os.path.getsize(tarpath) try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() # timeout _hf_upload_backup(tarpath, tar_size, filename, api) except Exception as e: print(f"{RED}[backup] HF: {e}{RESET}") finally: if os.path.exists(tarpath): os.remove(tarpath) # # # def _hf_download(filename: str = FULL_NAME) -> str | None: """ HF Dataset .""" if not HF_REPO or not HF_TOKEN: return None try: from huggingface_hub import hf_hub_download try: return hf_hub_download( repo_id=HF_REPO, filename=filename, repo_type="dataset", token=HF_TOKEN, timeout=60, ) except TypeError: return hf_hub_download( repo_id=HF_REPO, filename=filename, repo_type="dataset", token=HF_TOKEN, ) except Exception as e: print(f"[restore] HF download failed: {e}") return None def _stop_openviking(): """Stop OpenViking server via PID file.""" import signal pid_file = "/tmp/openviking.pid" if not os.path.exists(pid_file): print("[backup] OpenViking not running (no PID file)") return True try: with open(pid_file) as f: pid = int(f.read().strip()) print(f"[backup] Stopping OpenViking (PID: {pid})...") os.kill(pid, signal.SIGTERM) for _ in range(20): try: os.kill(pid, 0) time.sleep(0.5) except ProcessLookupError: print("[backup] OpenViking stopped") return True print("[backup] OpenViking did not stop gracefully, sending SIGKILL") os.kill(pid, signal.SIGKILL) time.sleep(0.5) return True except (FileNotFoundError, ValueError, ProcessLookupError, OSError) as e: print(f"[backup] OpenViking stop error: {e}") return True def _start_openviking(): """Start OpenViking server.""" import subprocess config = os.path.expanduser("~/.openviking/ov.conf") if not os.path.exists(config): print("[backup] OpenViking config not found, cannot start") return False try: with open("/tmp/openviking-server.log", "w") as f: proc = subprocess.Popen( ["nohup", "openviking-server", "--config", config], stdout=f, stderr=subprocess.STDOUT, preexec_fn=os.setpgrp, ) with open("/tmp/openviking.pid", "w") as f: f.write(str(proc.pid)) print(f"[backup] OpenViking started (PID: {proc.pid})") return True except Exception as e: print(f"[backup] OpenViking start failed: {e}") return False def _restore_from_webdav(root: Path, target: str, source: str) -> bool: """ WebDAV True""" if not WEBDAV_URL or not wd_exists(target): return False print(f"[restore] Downloading {target} from WebDAV...") data = wd_download(target) tarpath = f"/tmp/{target}" with open(tarpath, "wb") as f: f.write(data) with tarfile.open(tarpath, "r:gz") as tar: tar.extractall(path=STATE_DIR) os.remove(tarpath) # Apply incremental overrides only for default full backup (auto mode) if source == "auto" and target == FULL_NAME: try: manifest = _load_manifest() count = 0 for rel in manifest: p = root / rel p.parent.mkdir(parents=True, exist_ok=True) try: data = wd_download(f"files/{rel}") p.write_bytes(data) count += 1 except Exception as e: print(f"[restore] : {rel} {e}") print(f"[restore] Applied {count} incremental file overrides") except Exception: print("[restore] No incremental manifest found (clean start)") print("[restore] Restore from WebDAV complete") return True def _restore_from_dataset(target: str) -> bool: """ HF Dataset True""" print(f"[restore] Trying HF Dataset: {target}...") path = _hf_download(target) if not path: return False with tarfile.open(path, "r:gz") as tar: tar.extractall(path=STATE_DIR) print("[restore] Restore from HF Dataset complete") return True def restore(source="auto", filename=None): """ : - source="auto": WebDAV ( ) HF Dataset - source="webdav": WebDAV filename - source="dataset": HF Dataset filename filename None openclaw-full.tar.gz """ root = Path(STATE_DIR) root.mkdir(parents=True, exist_ok=True) restored = False target = filename or FULL_NAME # Strategy 1: WebDAV if source in ("webdav", "auto"): restored = _restore_from_webdav(root, target, source) # Strategy 2: HF Dataset fallback if source in ("dataset", "auto") and not restored: restored = _restore_from_dataset(target) if not restored: print("[restore] No backup found fresh start") def list_backups() -> dict: """ {webdav: [], dataset: []}.""" result: dict = {"webdav": [], "dataset": []} # WebDAV banner if _wd_is_configured(): try: files = wd_list("") result["webdav"] = sorted([f for f in files if f.endswith(".tar.gz")]) except Exception as e: print(f"[backup] WebDAV list error: {e}") # HF Dataset if HF_REPO and HF_TOKEN: try: backups = _hf_list_backup_files() if backups: result["dataset"] = sorted(backups.keys()) except Exception as e: print(f"[backup] HF list error: {e}") return result # # (Tick ) OVvectordb # def _get_hf_file_mtime(filename: str) -> float: """ HF Dataset (unix timestamp) 0""" try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() for entry in api.list_repo_tree(HF_REPO, repo_type="dataset", token=HF_TOKEN, expand=True): if getattr(entry, 'rfilename', '') == filename: lc = getattr(entry, 'last_commit', None) if lc and hasattr(lc, 'date'): return lc.date.timestamp() return 0.0 # return 0.0 # file not found except Exception as e: print(f"[backup] HF : {e}") return 0.0 def _get_wd_last_backup_time() -> float: """ WebDAV (unix timestamp) 0""" try: files = wd_list("") latest = 0.0 for f in files: if f.startswith("openclaw-full") and f.endswith(".tar.gz"): url = _wd_url(f) r = requests.head(url, auth=_wd_auth(), timeout=WD_REQ_TIMEOUT) if 'last-modified' in r.headers: from email.utils import parsedate_to_datetime dt = parsedate_to_datetime(r.headers['last-modified']) ts = dt.timestamp() if ts > latest: latest = ts return latest except Exception: return 0.0 def _get_hf_large_backup_mtime() -> float: """ HF Dataset (unix timestamp) 0""" try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() latest_ts = 0.0 for entry in api.list_repo_tree(HF_REPO, repo_type="dataset", token=HF_TOKEN, expand=True): name = getattr(entry, 'rfilename', '') if name.startswith("openclaw-full-") and name.endswith(".tar.gz") and name != FULL_NAME: lc = getattr(entry, 'last_commit', None) if lc and hasattr(lc, 'date'): ts = lc.date.timestamp() if ts > latest_ts: latest_ts = ts return latest_ts except Exception: return 0.0 def _do_one_tick(): """ tick: OV OV """ now = time.time() today = _today_str() need_stop_ov = False # 1. WebDAV if not WD_SKIPPED: last_wd = _get_wd_last_backup_time() if last_wd == 0.0 or (now - last_wd) >= WEBDAV_INTERVAL * 60: need_stop_ov = True # 2. HF if not HF_SKIPPED: last_hf_small = _get_hf_file_mtime(FULL_NAME) if last_hf_small == 0.0 or (now - last_hf_small) >= HF_SMALL_INTERVAL * 60: need_stop_ov = True # 3. HF hf_big_due = False if not HF_SKIPPED: # OV today_file = f"openclaw-full-{_today_str()}.tar.gz" try: from huggingface_hub import HfApi try: api_pre = HfApi(timeout=60) except TypeError: api_pre = HfApi() for entry in api_pre.list_repo_tree(HF_REPO, repo_type="dataset", token=HF_TOKEN, expand=True): if getattr(entry, 'rfilename', '') == today_file and getattr(entry, 'last_commit', None): today_exists = True break else: today_exists = False except Exception: today_exists = False if not today_exists: last_big_mtime = _get_hf_large_backup_mtime() hf_big_due = (last_big_mtime == 0.0) or (now - last_big_mtime >= HF_BIG_INTERVAL * 60) if hf_big_due: need_stop_ov = True if not need_stop_ov: return # # OV OV ov_was_running = False import signal try: with open("/tmp/openviking.pid") as f: ov_pid = int(f.read().strip()) os.kill(ov_pid, 0) ov_was_running = True except (FileNotFoundError, ValueError, ProcessLookupError, OSError): ov_was_running = False if ov_was_running: _stop_openviking() time.sleep(1) # WebDAV if not WD_SKIPPED: last_wd = _get_wd_last_backup_time() if last_wd == 0.0 or (now - last_wd) >= WEBDAV_INTERVAL * 60: try: c = incremental_backup() print(f"[backup] WebDAV : {c} ") full_backup() except Exception as e: print(f"[backup] WebDAV (): {e}") # HF if not HF_SKIPPED: last_hf_small = _get_hf_file_mtime(FULL_NAME) if last_hf_small == 0.0 or (now - last_hf_small) >= HF_SMALL_INTERVAL * 60: try: hf_small_backup() except Exception as e: print(f"[backup] HF (): {e}") # HF if not HF_SKIPPED and hf_big_due: try: hf_large_backup() except Exception as e: print(f"[backup] HF (): {e}") if ov_was_running: _start_openviking() def scheduler_loop(): """ Tick () OV vectordb """ _ensure_banner() print(f"[backup] tick ") print(f"[backup] WebDAV ={WEBDAV_INTERVAL} ={WEBDAV_MAX_BACKUPS}") print(f"[backup] HF ={HF_SMALL_INTERVAL}") print(f"[backup] HF ={HF_BIG_INTERVAL} ={DATASET_MAX_BACKUPS}") print(f"[backup] tick 5") while True: time.sleep(5) _do_one_tick() # # CLI # if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="WebDAV + HF Dataset /") parser.add_argument("cmd", nargs="?", default="restore", choices=["restore", "list", "incremental", "full", "small", "large", "scheduler", "check"], help=" (default: restore)") parser.add_argument("--source", default="auto", help=": auto|webdav|dataset ( restore )") parser.add_argument("--filename", default=None, help=" ( restore )") args = parser.parse_args() # list/check banner stdout JSON if args.cmd not in ("list", "check"): _ensure_banner() if args.cmd == "restore": restore(source=args.source, filename=args.filename) elif args.cmd == "list": import json print(json.dumps(list_backups(), ensure_ascii=False)) elif args.cmd == "incremental": c = incremental_backup() print(f"[backup] WebDAV : {c} ") elif args.cmd == "full": full_backup() elif args.cmd == "small": hf_small_backup() elif args.cmd == "large": hf_large_backup() elif args.cmd == "scheduler": scheduler_loop() elif args.cmd == "check": import json result = {"backup_today": False, "last_backup": None} try: from huggingface_hub import HfApi try: api = HfApi(timeout=60) except TypeError: api = HfApi() for entry in api.list_repo_tree(HF_REPO, repo_type="dataset", token=HF_TOKEN, expand=True): name = getattr(entry, 'rfilename', '') if name == FULL_NAME: # lc = getattr(entry, 'last_commit', None) if lc and hasattr(lc, 'date'): # UTC (UTC+8) bj_time = lc.date.astimezone(timezone(timedelta(hours=8))) result["last_backup"] = bj_time.strftime("%Y-%m-%d %H:%M:%S") if name.startswith("openclaw-full-") and name.endswith(".tar.gz") and name != FULL_NAME: date_str = name.replace("openclaw-full-", "").replace(".tar.gz", "") if date_str == _today_str(): lc = getattr(entry, 'last_commit', None) if lc and hasattr(lc, 'date'): bj_time = lc.date.astimezone(timezone(timedelta(hours=8))) result["backup_today"] = True result["last_backup"] = bj_time.strftime("%Y-%m-%d %H:%M:%S") except Exception as e: print(f"[backup] : {e}") print(json.dumps(result)) else: print(f": {sys.argv[0]} {{restore|incremental|full|small|large|scheduler}}") sys.exit(1)