Spaces:
Running
Running
| """同目录HF待同步写前记录；调用方持业务文件锁，提交方另持公共提交锁。""" | |
| import hashlib | |
| import json | |
| import os | |
| from pathlib import Path | |
| import stat | |
| import tempfile | |
| import uuid | |
| class SyncJournal: | |
| def __init__(self,directory): | |
| self.directory=Path(directory)/'.hf-pending' | |
| def validate_name(name): | |
| if (not isinstance(name,str) or not name or name in {'.','..'} | |
| or os.path.normcase(name)==os.path.normcase('__hf_dataset_sync__') | |
| or any(char in name for char in '/\\:') or any(ord(char)<32 for char in name)): | |
| raise ValueError('invalid sync file name') | |
| def path(self,name): | |
| self.validate_name(name) | |
| key=hashlib.sha256(os.path.normcase(name).encode('utf-8')).hexdigest() | |
| return self.directory/(key+'.json') | |
| def check_regular(path,directory=False): | |
| info=os.lstat(path) | |
| if (stat.S_ISLNK(info.st_mode) or getattr(info,'st_file_attributes',0)&1024 | |
| or not (stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode))): | |
| raise ValueError('invalid sync journal path') | |
| def mark(self,name): | |
| target=self.path(name) | |
| self.directory.mkdir(exist_ok=True) | |
| self.check_regular(self.directory,directory=True) | |
| if target.exists() or target.is_symlink(): self.check_regular(target) | |
| generation=uuid.uuid4().hex | |
| data=json.dumps({'schema':1,'file':name,'generation':generation},ensure_ascii=False).encode('utf-8') | |
| if len(data)>4096: raise ValueError('sync journal record too large') | |
| descriptor,temporary=tempfile.mkstemp(prefix='.pending-',suffix='.tmp',dir=self.directory) | |
| try: | |
| with os.fdopen(descriptor,'wb') as stream: | |
| stream.write(data); stream.flush(); os.fsync(stream.fileno()) | |
| os.replace(temporary,target) | |
| finally: | |
| try: os.unlink(temporary) | |
| except FileNotFoundError: pass | |
| return generation | |
| def read_record(self,path): | |
| self.check_regular(self.directory,directory=True) | |
| self.check_regular(path) | |
| descriptor=os.open(path,os.O_RDONLY|getattr(os,'O_NOFOLLOW',0)) | |
| with os.fdopen(descriptor,'rb') as stream: | |
| if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): raise ValueError('invalid sync journal file') | |
| content=stream.read(4097) | |
| if len(content)>4096: raise ValueError('sync journal record too large') | |
| data=json.loads(content) | |
| if (not isinstance(data,dict) or set(data)!={'schema','file','generation'} | |
| or type(data['schema']) is not int or data['schema']!=1): | |
| raise ValueError('invalid sync journal schema') | |
| self.validate_name(data['file']) | |
| generation=data['generation'] | |
| if (not isinstance(generation,str) or len(generation)!=32 | |
| or any(char not in '0123456789abcdef' for char in generation) | |
| or self.path(data['file'])!=path): | |
| raise ValueError('invalid sync journal identity') | |
| return data | |
| def read(self,name): | |
| try: return self.read_record(self.path(name))['generation'] | |
| except FileNotFoundError: return None | |
| def pending(self,on_invalid=None): | |
| """默认严格读取；同步方可隔离坏记录，目录本身故障仍整体拒绝。""" | |
| if not self.directory.exists(): | |
| if self.directory.is_symlink(): raise ValueError('invalid sync journal directory') | |
| return set() | |
| self.check_regular(self.directory,directory=True) | |
| names=set() | |
| for path in self.directory.glob('*.json'): | |
| try: names.add(self.read_record(path)['file']) | |
| except FileNotFoundError: pass # 其他提交方已确认该记录。 | |
| except (OSError,ValueError) as error: | |
| if on_invalid is None: raise | |
| on_invalid(error) | |
| return names | |
| def acknowledge(self,name,generation): | |
| if (not isinstance(generation,str) or len(generation)!=32 | |
| or any(char not in '0123456789abcdef' for char in generation)): | |
| return False | |
| if self.read(name)!=generation: return False | |
| os.unlink(self.path(name)) | |
| return True | |