HuggingFace Space 部署 OpenClaw 指南

import os
import sys
import tarfile
import shutil
import hashlib
import subprocess
from huggingface_hub import HfApi, hf_hub_download

api = HfApi()
repo_id = os.getenv("HF_DATASET")
token = os.getenv("HF_TOKEN")
FILENAME = "latest_backup.tar.gz"
CHECKSUM_FILE = "latest_backup.sha256"
BACKUP_TIMEOUT = 1200  # 20 分钟


def calc_sha256(filepath):
    h = hashlib.sha256()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def restore():
    print(f"[sync] restore() 开始")
    print(f"  HF_DATASET: {repo_id}")
    print(f"  HF_TOKEN:   {'已设置' if token else '未设置'}")
    if not repo_id or not token:
        print("[sync] Skip Restore: HF_DATASET or HF_TOKEN not set")
        return

    # ── 阶段 1：下载 & 校验 ──
    print(f"[sync] 阶段 1/3：下载 & 校验")
    try:
        print(f"  下载 {FILENAME} ...")
        path = hf_hub_download(
            repo_id=repo_id,
            filename=FILENAME,
            repo_type="dataset",
            token=token,
        )
        print(f"  下载完成: {os.path.getsize(path) / 1024 / 1024:.1f} MB")
    except Exception as e:
        print(f"  远程无备份或下载失败 ({type(e).__name__})，跳过恢复")
        return

    # tar 包完整性校验
    print(f"  校验 tar 包完整性 ...")
    try:
        with tarfile.open(path, "r:gz") as tar:
            members = tar.getmembers()
        print(f"  ✓ tar 包完整，{len(members)} 个条目")
    except tarfile.TarError as e:
        print(f"  ✗ tar 包损坏: {e}")
        return

    # SHA256 二次校验
    print(f"  校验 SHA256 ...")
    try:
        cs_path = hf_hub_download(repo_id=repo_id, filename=CHECKSUM_FILE,
                                 repo_type="dataset", token=token)
        expected = open(cs_path).read().strip()
        actual = calc_sha256(path)
        if expected != actual:
            print(f"  ✗ SHA256 不匹配！expected={expected[:16]}..., actual={actual[:16]}...")
            return
        print(f"  ✓ SHA256 匹配")
    except Exception as e:
        print(f"  ! SHA256 校验跳过: {e}")

    # ── 阶段 2：清空 & 解压 ──
    print(f"[sync] 阶段 2/3：清空 & 解压")
    target = "/root/.openclaw"
    if os.path.exists(target):
        before = set(os.listdir(target))
        shutil.rmtree(target)
        print(f"  已清空 {target}（{len(before)} 项）")
    os.makedirs(target)

    print(f"  解压到 /root/ ...")
    with tarfile.open(path, "r:gz") as tar:
        tar.extractall(path="/root")
    print(f"  解压完成")

    # 恢复后文件列表
    restored = sorted(os.listdir("/root/.openclaw"))
    print(f"[sync] ✓ 恢复成功，共 {len(restored)} 项：")
    for item in restored:
        m = os.stat(os.path.join("/root/.openclaw", item)).st_mode & 0o777
        print(f"    {oct(m)[-3:]}  {item}")


def backup():
    print(f"[sync] backup() 开始")
    print(f"  HF_DATASET: {repo_id}")
    print(f"  HF_TOKEN:   {'已设置' if token else '未设置'}")
    if not repo_id or not token:
        print("[sync] Skip Backup: HF_DATASET or HF_TOKEN not set")
        return

    src = "/root/.openclaw"
    tmp_tar = "/tmp/latest_backup.tar.gz"
    tmp_sha = "/tmp/latest_backup.sha256"

    print(f"[sync] 打包 {src} ...")
    tar_cmd = (
        f"cd /root && tar --numeric-owner -czf {tmp_tar} "
        f".openclaw > /dev/null 2>&1 && echo OK"
    )
    print(f"  tar 命令执行中 (timeout={BACKUP_TIMEOUT}s) ...")
    proc = subprocess.run(
        tar_cmd,
        shell=True,
        capture_output=True,
        text=True,
        timeout=BACKUP_TIMEOUT
    )
    if proc.returncode != 0:
        print(f"[sync] ✗ tar 失败: {proc.stderr[:200]}")
        return

    size_mb = os.path.getsize(tmp_tar) / 1024 / 1024
    print(f"  打包完成: {size_mb:.1f} MB")

    # SHA256
    checksum = calc_sha256(tmp_tar)
    with open(tmp_sha, "w") as f:
        f.write(checksum)
    print(f"  SHA256: {checksum[:16]}...")

    # 上传
    print(f"[sync] 上传 {FILENAME} ({size_mb:.1f} MB) ...")
    try:
        api.upload_file(
            path_or_fileobj=tmp_tar,
            path_in_repo=FILENAME,
            repo_id=repo_id,
            repo_type="dataset",
            token=token,
        )
        print(f"  ✓ {FILENAME} 上传成功")
    except Exception as e:
        print(f"[sync] ✗ {FILENAME} 上传失败: {e}")
        for f in [tmp_tar, tmp_sha]:
            if os.path.exists(f):
                os.remove(f)
        print("[sync] Backup aborted")
        return

    try:
        api.upload_file(
            path_or_fileobj=tmp_sha,
            path_in_repo=CHECKSUM_FILE,
            repo_id=repo_id,
            repo_type="dataset",
            token=token,
        )
        print(f"  ✓ {CHECKSUM_FILE} 上传成功")
    except Exception as e:
        print(f"[sync] ✗ {CHECKSUM_FILE} 上传失败: {e}")

    # 清理
    for f in [tmp_tar, tmp_sha]:
        if os.path.exists(f):
            os.remove(f)
    print(f"[sync] ✓ 备份完成: {FILENAME} ({size_mb:.1f} MB)")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "backup":
        backup()
    else:
        restore()