CoolFace
Apppublic

ckriti/HuggingClaw

sourceHugging Facemitupdated 7mo agoView on Hugging Face
0likes
restore_from_dataset.py80 linesDownload Raw Back to scripts
1import os2import tarfile3import sys4 5from huggingface_hub import hf_hub_download, HfApi6 7 8def main() -> None:9    """10    从 Hugging Face Dataset 恢复 ~/.openclaw 目录到本地。11 12    依赖环境变量:13    - HF_TOKEN: 具有写入/读取权限的 HF Access Token14    - OPENCLAW_DATASET_REPO: 数据集 repo_id,例如 "username/dataset-name"15    """16    repo_id = os.environ.get("OPENCLAW_DATASET_REPO")17    token = os.environ.get("HF_TOKEN")18 19    if not repo_id or not token:20        # 未配置就直接跳过,不报错以免阻塞网关启动21        return22 23    state_dir = os.path.expanduser("~/.openclaw")24    os.makedirs(state_dir, exist_ok=True)25 26    try:27        # List all files and find the latest backup28        api = HfApi(token=token)29        files = api.list_repo_files(repo_id=repo_id, repo_type="dataset")30        31        # Filter for our backup pattern (support both .tar and .tar.gz)32        backups = sorted([f for f in files if f.startswith("state/backup-") and (f.endswith(".tar") or f.endswith(".tar.gz"))], reverse=True)33        34        if not backups:35            # Fallback to legacy filename if no rolling backups exist36            if "state/openclaw.tar" in files:37                backups = ["state/openclaw.tar"]38            else:39                print("[restore_from_dataset] No backups found.", file=sys.stderr)40                return41 42        # Try to restore from the latest backup, falling back to older ones if needed43        success = False44        for backup_file in backups:45            print(f"[restore_from_dataset] Attempting to restore from: {backup_file}")46            try:47                tar_path = hf_hub_download(48                    repo_id=repo_id,49                    repo_type="dataset",50                    filename=backup_file,51                    token=token,52                )53                54                # Auto-detect compression based on file extension or header (r:*)55                with tarfile.open(tar_path, "r:*") as tf:56                    tf.extractall(state_dir)57                58                print(f"[restore_from_dataset] Successfully restored from {backup_file}")59                success = True60                break61            except Exception as e:62                print(f"[restore_from_dataset] Failed to restore {backup_file}: {e}", file=sys.stderr)63                # Continue to next backup64        65        if not success:66             print("[restore_from_dataset] All backup restore attempts failed.", file=sys.stderr)67             return68 69    except Exception as e:70        # General failure (network, auth, etc)71        print(f"[restore_from_dataset] Restore process failed: {e}", file=sys.stderr)72        return73 74    # 重要:不要删除 credentials/whatsapp。恢复的凭证用于自动连接;75    # 若在此处删除会导致每次启动都需重新扫码,且 dataset 中的好状态无法被使用。76 77 78if __name__ == "__main__":79    main()80