ckriti/HuggingClaw

1import os2import tarfile3import tempfile4import sys5import time6from datetime import datetime7 8from huggingface_hub import HfApi9 10def main() -> None:11    """12    Backs up ~/.openclaw to Hugging Face Dataset with rolling history.13    Keeps the last 5 backups to prevent data loss from corruption.14    15    Env vars:16    - HF_TOKEN17    - OPENCLAW_DATASET_REPO18    """19    repo_id = os.environ.get("OPENCLAW_DATASET_REPO")20    token = os.environ.get("HF_TOKEN")21 22    state_dir = os.path.expanduser("~/.openclaw")23 24    if not repo_id or not token:25        print("[save_to_dataset] Missing configuration.", file=sys.stderr)26        return27 28    if not os.path.isdir(state_dir):29        print("[save_to_dataset] No state to save.", file=sys.stderr)30        return31 32    # 1. Validation: Ensure we have valid credentials before backing up33    wa_creds_dir = os.path.join(state_dir, "credentials", "whatsapp", "default")34    if os.path.isdir(wa_creds_dir):35        file_count = len([f for f in os.listdir(wa_creds_dir) if os.path.isfile(os.path.join(wa_creds_dir, f))])36        if file_count < 2:37             # Basic sanity check: needs at least creds.json + keys. 38             # Lowered from 10 to 2 to be less aggressive but still catch empty/broken states.39            print(f"[save_to_dataset] Skip: WhatsApp credentials incomplete ({file_count} files).", file=sys.stderr)40            return41 42    api = HfApi(token=token)43    44    # Sync system logs to state dir for persistence45    try:46        sys_log_path = "/home/node/logs"47        backup_log_path = os.path.join(state_dir, "logs/sys_logs")48        if os.path.exists(sys_log_path):49            if os.path.exists(backup_log_path):50                import shutil51                shutil.rmtree(backup_log_path)52            # Use shutil.copytree but ignore socket files if any53            import shutil54            shutil.copytree(sys_log_path, backup_log_path, ignore_dangling_symlinks=True)55            print(f"[save_to_dataset] Synced logs from {sys_log_path} to {backup_log_path}")56    except Exception as e:57        print(f"[save_to_dataset] Warning: Failed to sync logs: {e}")58 59    # Check for credentials60    creds_path = os.path.join(state_dir, "credentials/whatsapp/default/auth_info_multi.json")61    if os.path.exists(creds_path):62        print(f"[save_to_dataset] ✅ WhatsApp credentials found at {creds_path}")63    else:64        print(f"[save_to_dataset] ⚠️  WhatsApp credentials NOT found (user might need to login)")65 66    # Generate timestamped filename67    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")68    filename = f"state/backup-{timestamp}.tar.gz"69 70    with tempfile.TemporaryDirectory() as tmpdir:71        tar_path = os.path.join(tmpdir, "openclaw.tar.gz")72 73        try:74            with tarfile.open(tar_path, "w:gz") as tf:75                # Filter to exclude lock files or temp files if needed, but allow extensions76                def exclude_filter(info: tarfile.TarInfo) -> tarfile.TarInfo | None:77                    if info.name.endswith(".lock"):78                        return None79                    return info80                81                tf.add(state_dir, arcname=".", filter=exclude_filter)82        except Exception as e:83            print(f"[save_to_dataset] Failed to compress: {e}", file=sys.stderr)84            return85 86        print(f"[save_to_dataset] Uploading backup: {filename}")87        try:88            api.upload_file(89                path_or_fileobj=tar_path,90                path_in_repo=filename,91                repo_id=repo_id,92                repo_type="dataset",93            )94        except Exception as e:95            print(f"[save_to_dataset] Upload failed: {e}", file=sys.stderr)96            return97 98    # 2. Rotation: Delete old backups, keep last 599    try:100        files = api.list_repo_files(repo_id=repo_id, repo_type="dataset")101        # Match both .tar and .tar.gz for backward compatibility during transition102        backups = sorted([f for f in files if f.startswith("state/backup-") and (f.endswith(".tar") or f.endswith(".tar.gz"))])103        104        if len(backups) > 5:105            # Delete oldest106            to_delete = backups[:-5]107            print(f"[save_to_dataset] Rotating backups, deleting: {to_delete}")108            for old_backup in to_delete:109                api.delete_file(110                    path_in_repo=old_backup,111                    repo_id=repo_id,112                    repo_type="dataset",113                    token=token114                )115    except Exception as e:116        print(f"[save_to_dataset] Rotation failed (non-fatal): {e}", file=sys.stderr)117 118