File size: 7,467 Bytes
0790cf1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
#!/usr/bin/env python3
"""
Atomic Dataset Restore for HermesFace
Restore state from HF Dataset with integrity validation and local backup.

Usage:
    python3 restore_from_dataset_atomic.py <repo_id> <target_dir> [--force]
"""
import hashlib
import json
import logging
import shutil
import sys
import tempfile
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from huggingface_hub import HfApi, hf_hub_download

logging.basicConfig(
    level=logging.INFO,
    format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", "module": "atomic-restore", "message": "%(message)s"}',
)
logger = logging.getLogger(__name__)


class AtomicDatasetRestorer:
    def __init__(self, repo_id: str, dataset_path: str = "state"):
        self.repo_id = repo_id
        self.dataset_path = Path(dataset_path)
        self.api = HfApi()

    def calculate_checksum(self, file_path: Path) -> str:
        h = hashlib.sha256()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                h.update(chunk)
        return h.hexdigest()

    def validate_integrity(self, metadata: Dict[str, Any], state_files: List[Path]) -> bool:
        try:
            if "checksum" not in metadata:
                logger.warning("no_checksum_in_metadata")
                return True
            calculated = hashlib.sha256(
                json.dumps(metadata.get("state_data", {}), sort_keys=True).encode()
            ).hexdigest()
            expected = metadata["checksum"]
            valid = calculated == expected
            logger.info(f"integrity_check expected={expected} calculated={calculated} valid={valid}")
            return valid
        except Exception as e:
            logger.error(f"integrity_validation_failed error={e}")
            return False

    def create_backup_before_restore(self, target_dir: Path) -> Optional[Path]:
        try:
            if not target_dir.exists():
                return None
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            backup_dir = target_dir.parent / f"state_backup_{timestamp}"
            logger.info(f"creating_local_backup source={target_dir} backup={backup_dir}")
            shutil.copytree(target_dir, backup_dir)
            return backup_dir
        except Exception as e:
            logger.error(f"local_backup_failed error={e}")
            return None

    def restore_from_commit(
        self, commit_sha: str, target_dir: Path, force: bool = False
    ) -> Dict[str, Any]:
        operation_id = f"restore_{int(time.time())}"
        logger.info(f"starting_atomic_restore op={operation_id} commit={commit_sha}")

        try:
            self.api.repo_info(repo_id=self.repo_id, repo_type="dataset", revision=commit_sha)
        except Exception as e:
            return {"success": False, "operation_id": operation_id,
                    "error": f"Invalid commit: {e}", "timestamp": datetime.now().isoformat()}

        backup_dir = self.create_backup_before_restore(target_dir)

        try:
            with tempfile.TemporaryDirectory() as _tmpdir:
                files = self.api.list_repo_files(
                    repo_id=self.repo_id, repo_type="dataset", revision=commit_sha
                )
                state_files = [f for f in files if f.startswith(str(self.dataset_path))]
                if not state_files:
                    return {"success": False, "operation_id": operation_id,
                            "error": "No state files found in commit",
                            "timestamp": datetime.now().isoformat()}

                downloaded_files: List[Path] = []
                metadata = None
                for file_path in state_files:
                    try:
                        local = hf_hub_download(
                            repo_id=self.repo_id,
                            repo_type="dataset",
                            filename=file_path,
                            revision=commit_sha,
                        )
                        if local:
                            downloaded_files.append(Path(local))
                            if file_path.endswith("metadata.json"):
                                with open(local, "r") as f:
                                    metadata = json.load(f)
                    except Exception as e:
                        logger.error(f"file_download_failed file={file_path} error={e}")

                if not metadata:
                    return {"success": False, "operation_id": operation_id,
                            "error": "Metadata not found in state files",
                            "timestamp": datetime.now().isoformat()}

                if not self.validate_integrity(metadata, downloaded_files):
                    return {"success": False, "operation_id": operation_id,
                            "error": "Data integrity validation failed",
                            "timestamp": datetime.now().isoformat()}

                target_dir.mkdir(parents=True, exist_ok=True)
                restored = []
                for f in downloaded_files:
                    if f.name != "metadata.json":
                        dst = target_dir / f.name
                        shutil.copy2(f, dst)
                        restored.append(str(dst))

                result = {
                    "success": True,
                    "operation_id": operation_id,
                    "commit_sha": commit_sha,
                    "backup_dir": str(backup_dir) if backup_dir else None,
                    "timestamp": datetime.now().isoformat(),
                    "restored_files": restored,
                    "metadata": metadata,
                }
                logger.info(f"atomic_restore_completed {result}")
                return result
        except Exception as e:
            logger.error(f"atomic_restore_failed error={e}")
            return {"success": False, "operation_id": operation_id,
                    "error": str(e), "timestamp": datetime.now().isoformat()}

    def restore_latest(self, target_dir: Path, force: bool = False) -> Dict[str, Any]:
        try:
            repo_info = self.api.repo_info(repo_id=self.repo_id, repo_type="dataset")
            if not repo_info.sha:
                return {"success": False, "error": "No commit found in repository",
                        "timestamp": datetime.now().isoformat()}
            return self.restore_from_commit(repo_info.sha, target_dir, force)
        except Exception as e:
            return {"success": False, "error": f"Failed to get latest commit: {e}",
                    "timestamp": datetime.now().isoformat()}


def main() -> None:
    if len(sys.argv) < 3:
        print(json.dumps({
            "error": "Usage: python restore_from_dataset_atomic.py <repo_id> <target_dir> [--force]",
            "status": "error",
        }, indent=2))
        sys.exit(1)

    repo_id = sys.argv[1]
    target_dir = sys.argv[2]
    force = "--force" in sys.argv

    try:
        restorer = AtomicDatasetRestorer(repo_id)
        result = restorer.restore_latest(Path(target_dir), force)
        print(json.dumps(result, indent=2))
        if not result.get("success", False):
            sys.exit(1)
    except Exception as e:
        print(json.dumps({"error": str(e), "status": "error"}, indent=2))
        sys.exit(1)


if __name__ == "__main__":
    main()