Spaces:
Sleeping
Sleeping
| """Versioned metadata with optimistic concurrency; audio and index commit together.""" | |
| import copy | |
| import json | |
| import os | |
| import threading | |
| from pathlib import Path | |
| from huggingface_hub import CommitOperationAdd, CommitOperationDelete, HfApi, hf_hub_download | |
| from huggingface_hub.errors import EntryNotFoundError, HfHubHTTPError | |
| INDEX = 'v1/indice.json' | |
| class Conflict(Exception): | |
| pass | |
| class Store: | |
| def __init__(self, repo=None, token=None, directory='data'): | |
| self.repo, self.token = repo, token | |
| self.api = HfApi(token=token) if repo and token else None | |
| self.directory = Path(directory) | |
| self.lock = threading.Lock() | |
| def mode(self): | |
| return 'huggingface' if self.api else 'local' | |
| def read(self): | |
| if self.api: | |
| revision = self.api.repo_info(self.repo, repo_type='dataset').sha | |
| try: | |
| path = hf_hub_download(self.repo, INDEX, repo_type='dataset', revision=revision, token=self.token) | |
| return json.loads(Path(path).read_text()), revision | |
| except EntryNotFoundError: | |
| return empty_index(), revision | |
| path = self.directory / INDEX | |
| return (json.loads(path.read_text()) if path.exists() else empty_index()), None | |
| def commit(self, data, revision, files=None): | |
| encoded = json.dumps(data, ensure_ascii=False, indent=2).encode() | |
| if self.api: | |
| operations = [CommitOperationAdd(path_in_repo=INDEX, path_or_fileobj=encoded)] | |
| operations += [CommitOperationDelete(path_in_repo=p) if b is None else | |
| CommitOperationAdd(path_in_repo=p, path_or_fileobj=b) for p, b in (files or {}).items()] | |
| try: | |
| self.api.create_commit(self.repo, repo_type='dataset', operations=operations, | |
| commit_message='Actualizar colección de voz', parent_commit=revision) | |
| except HfHubHTTPError as exc: | |
| if exc.response.status_code in (409, 412): | |
| raise Conflict() from exc | |
| raise | |
| else: | |
| for name, content in (files or {}).items(): | |
| path = self.directory / name | |
| if content is None: | |
| path.unlink(missing_ok=True) | |
| else: | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_bytes(content) | |
| path = self.directory / INDEX | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| temp = path.with_suffix('.tmp') | |
| temp.write_bytes(encoded) | |
| os.replace(temp, path) | |
| def legacy_manifests(self): | |
| if self.api: | |
| revision = self.api.repo_info(self.repo, repo_type='dataset').sha | |
| names = set(self.api.list_repo_files(self.repo, repo_type='dataset', revision=revision)) | |
| result = {} | |
| for name in sorted(names): | |
| if name.endswith('/manifest.json') and not name.startswith('v1/'): | |
| path = hf_hub_download(self.repo, name, repo_type='dataset', revision=revision, token=self.token) | |
| result[name] = json.loads(Path(path).read_text()) | |
| return result, names | |
| if not self.directory.exists(): | |
| return {}, set() | |
| names = {str(p.relative_to(self.directory)) for p in self.directory.rglob('*') if p.is_file()} | |
| return {name: json.loads((self.directory / name).read_text()) for name in sorted(names) | |
| if name.endswith('/manifest.json') and not name.startswith('v1/')}, names | |
| def mutate(self, change): | |
| with self.lock: | |
| for _ in range(4): | |
| data, revision = self.read() | |
| result, files = change(copy.deepcopy(data)) | |
| try: | |
| self.commit(result, revision, files) | |
| return result | |
| except Conflict: | |
| continue | |
| raise Conflict('Otra sesión está actualizando el dataset. Intenta de nuevo.') | |
| def empty_index(): | |
| return {'schema_version': 1, 'collections': [{'id': 'principal', 'name': 'Mi colección'}], | |
| 'items': {}, 'recordings': [], 'sources': []} | |