chronaxis.results.publication
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""给同次输出分配时间文件名，并在已有结果目录中不覆盖地发布。23只检查本次候选路径，不扫描旧结果。平台不覆盖移动保护竞争期间出现的文件；4临时锁仅协调使用相同时间后缀的发布者，完成清单最后移动。失败回滚本次移动，5身份发生变化的路径保留给调用者处理，不删除其他运行或外部替换的文件。6"""78from datetime import datetime9import os10from pathlib import Path11from secrets import randbits12from .platform_files import move_no_replace, require_publication_platform13from .owned_directory import directory_identity141516def output_root(value):17 """检查可复用结果根。参数: value路径。返回: 绝对Path。异常: 路径/平台错误。"""18 root = Path(value).absolute()19 require_publication_platform()20 if root.is_symlink() or root.is_junction() or (root.exists() and not root.is_dir()):21 raise FileExistsError('输出目标必须为目录：' + str(root))22 return root232425def _identity(path):26 """取得文件身份。参数: path实际路径。返回: 设备和inode。异常: 原stat错误。"""27 value = path.lstat()28 return value.st_dev, value.st_ino293031class PublicationNames:32 """独占本次时间后缀，保留有限文件映射及已移动文件身份。"""3334 def __init__(self, root, names):35 """分配不冲突名称。参数: root输出根、names逻辑文件名。返回: 实例。异常: IO错。"""36 self.root = output_root(root)37 self.root.mkdir(parents=True, exist_ok=True)38 self._root_identity=directory_identity(self.root)39 self.lock = None40 self.moved = []41 self.completed = False42 names = tuple(names)43 if any(Path(n).name != n or n in ('.', '..') for n in names):44 raise ValueError('发布名称必须是单层文件或目录名')45 base = datetime.now().astimezone().strftime('%Y%m%d_%H%M%S_%f')46 index = 047 permission_retry = False48 # 同刻、时钟回拨及已有输出均只影响本组候选，不遍历目录。49 while True:50 stamp = base if index == 0 else f'{base}_{index}'51 lock = self.root / ('.publishing-' + stamp)52 try:53 stream = lock.open('xb')54 except FileExistsError:55 index += 156 continue57 except PermissionError:58 # Windows正在删除同刻锁时也会返回拒绝访问。只换一次独立候选；59 # 真正的目录权限错误仍立即传播，不能无限重试或删除别人的锁。60 if os.name!='nt' or permission_retry:61 raise62 index = randbits(64) + 163 permission_retry = True64 continue65 except BaseException:66 self.close()67 raise68 with stream:69 value = os.fstat(stream.fileno())70 self.lock = lock71 self.lock_identity = value.st_dev, value.st_ino72 try:73 self.names = {n: f'{Path(n).stem}_{stamp}{Path(n).suffix}' for n in names}74 if any(os.path.lexists(self.root / n) for n in self.names.values()):75 self.close()76 index += 177 continue78 except BaseException as error:79 try:80 self.close()81 except BaseException as cleanup:82 error.add_note('命名锁释放失败：' + str(cleanup))83 raise84 self.stamp = stamp85 return8687 def publish(self, temporary, last, installer=None, *, only=None):88 """先移数据再移清单。参数: temporary临时目录、last完成文件、installer可选rename、only本次子集。返回: None。异常: IO/取消。"""89 move = move_no_replace if installer is None else installer90 if directory_identity(self.root)!=self._root_identity:91 raise ValueError('发布根目录身份已变化')92 temporary_identity=directory_identity(Path(temporary))93 # 同一报告再次写出时，既有成功移动不属于此次失败的回滚范围。94 self.moved = []95 self.completed = False96 selected = tuple(self.names) if only is None else tuple(only)97 if last not in selected or any(name not in self.names for name in selected):98 raise ValueError('发布子集必须属于本次命名并包含完成清单')99 try:100 for name in (*[n for n in selected if n != last], last):101 if directory_identity(Path(temporary))!=temporary_identity or directory_identity(self.root)!=self._root_identity:102 raise ValueError('发布父目录身份已变化')103 source, target = Path(temporary) / name, self.root / self.names[name]104 identity = _identity(source)105 # 先登记候选身份，覆盖系统调用已完成却在返回前取消的情况。106 self.moved.append((source, target, identity))107 move(source, target)108 if _identity(target)!=identity:109 raise ValueError('移动后的发布对象身份已变化')110 # 释放失败也进入回滚，不能留下completed却报告本次没有发布。111 self.close()112 self.completed = True113 except BaseException as error:114 # 回滚只接管身份仍与移动前一致的对象；其他文件绝不覆盖。115 for source, target, identity in reversed(self.moved):116 rollback_started = False117 try:118 if _identity(target) == identity:119 rollback_started = True120 move_no_replace(target, source)121 elif target.name == self.names[last]:122 # 外部替换的清单可能仍引用本次数据；不能删除其依赖。123 error.add_note('完成清单已被替换，保留已发布文件供调用者核对')124 break125 except BaseException as cleanup:126 if not rollback_started and isinstance(cleanup, FileNotFoundError):127 continue128 error.add_note('发布回滚未完成：' + str(cleanup))129 if target.name == self.names[last]:130 # rename可能已完成，随后目录fd关闭才报错；按实际文件身份判断。131 try:132 try:133 target_identity = _identity(target)134 except FileNotFoundError:135 target_identity = None136 try:137 source_identity = _identity(source)138 except FileNotFoundError:139 source_identity = None140 except BaseException as inspection:141 error.add_note('无法确认完成清单位置，保留数据：' + str(inspection))142 break143 if target_identity is None and source_identity == identity:144 error.add_note('完成清单已撤回临时目录，继续撤回其数据文件')145 continue146 self.completed = target_identity == identity147 if self.completed:148 error.add_note('完成清单仍已发布，其数据文件完整保留')149 else:150 error.add_note('完成清单位置或身份已变化，保留数据供核对')151 break152 try:153 self.close()154 except BaseException as cleanup:155 error.add_note('发布锁释放失败：' + str(cleanup))156 raise157158 def close(self):159 """释放本次命名锁。参数: 无。返回: None。异常: IO错，不删除其他身份的文件。"""160 if self.lock is not None:161 try:162 if _identity(self.lock) == self.lock_identity:163 self.lock.unlink()164 except FileNotFoundError:165 pass166 self.lock = None167