backup-manager.py

#!/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)