chronaxis.planning.input_paths
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""将文件、目录或通配符展开为有序输入，不读取报文内容。23明确文件列表保持用户顺序；每个目录或通配符内部采用自然排序。4只枚举显式请求的目录，不递归追踪子目录；递归必须通过**通配符提出。5展开最多4096项，空匹配及无效路径报配置错误，不悄悄跳过缺失目录。6"""78from glob import iglob9from pathlib import Path10import re1112from .checks import PlanValidationError, absolute_path, array_at, coordinate_at, object_at, text_at1314# 明确后缀的文件只做词法定位，保持现有静态配置不访问业务文件的行为。15FILE_SUFFIXES = frozenset(('.asc', '.blf', '.csv', '.txt'))16SOURCE_SUFFIXES = {'can_blf': {'.blf'}, 'can_asc': {'.asc'},17 'raw_can_csv': {'.csv', '.txt'}, 'signal_csv': {'.csv', '.txt'}}18MAX_INPUT_FILES = 4096192021def natural_path_key(path):22 """产生稳定自然序键，使file2排在file10前。2324 参数: path为Path。返回: 数字/文字分段键及原路径平局键。25 异常: 无主动异常；不访问文件系统。26 """27 text = str(path)28 # 类型标签避免数字与文字直接比较；补原文本使01/1及大小写平局可复现。29 parts = tuple((1, int(part)) if part.isdigit() else (0, part.casefold())30 for part in re.split(r'(\d+)', text))31 return parts, text.casefold(), text323334def expand_source_files(raw, base, where):35 """按声明展开来源路径；目录只取该来源支持的格式。3637 参数: raw为来源字典，base为配置文件目录，where为来源错误位置。38 返回: (绝对Path元组, 可选逐文件偏移字符串元组)。异常: PlanValidationError表示缺项、空匹配或超限；39 OSError由目录读取产生并转换为带配置位置的错误。40 副作用: 仅目录/通配符声明枚举文件名；明确文件不stat、不读数据内容。41 """42 if ('path' in raw) == ('paths' in raw):43 raise PlanValidationError(where, 'path与paths必须且只能提供一个')44 key = 'path' if 'path' in raw else 'paths'45 value = raw[key]46 entries = [value] if type(value) in (str, dict) else array_at(value, where + '.' + key)47 if not entries:48 raise PlanValidationError(where + '.' + key, '至少需要一个输入文件或目录')49 allowed = SOURCE_SUFFIXES.get(raw.get('source_type'), {'.asc', '.blf'})50 result, shifts, has_shifts = [], [], False51 for index, entry in enumerate(entries):52 location = f'{where}.{key}[{index}]'53 # 文件自己的偏移跟在同一条声明内，展开目录时该偏移作用于其中各文件。54 shift = '0s'55 if type(entry) is dict:56 item = object_at(entry, location, ('path',), ('offset',))57 text = text_at(item['path'], location + '.path')58 if 'offset' in item:59 coordinate_at(item['offset'], location + '.offset')60 shift = item['offset']61 has_shifts = True62 else:63 text = text_at(entry, location)64 target = absolute_path(text, base, location)65 # 仅*和?启动通配符模式，普通文件名中的方括号保持字面含义。66 wildcard = any(char in text for char in '*?')67 # 常规文件不访问磁盘；目录名像文件时可用末尾分隔符明确目录含义。68 if not wildcard and target.suffix.lower() in FILE_SUFFIXES and not text.endswith(('/', '\\')):69 selected = [target]70 else:71 selected = []72 try:73 pattern = str(target)74 suffix = target.suffix75 if wildcard and suffix.lower() in FILE_SUFFIXES:76 # 只统一明确扩展名的大小写；主体、隐藏文件和**仍遵循原glob规则。77 extension = ''.join('[' + char.lower() + char.upper() + ']'78 if char.isalpha() else char for char in suffix)79 pattern = pattern[:-len(suffix)] + extension80 candidates = (Path(item) for item in iglob(pattern, recursive='**' in text)) if wildcard else target.iterdir()81 for candidate in candidates:82 if candidate.suffix.lower() in allowed and candidate.is_file():83 selected.append(candidate)84 if len(result) + len(selected) > MAX_INPUT_FILES:85 raise PlanValidationError(location, '输入文件超过4096项')86 except OSError as error:87 raise PlanValidationError(location, '无法枚举输入目录：' + str(error)) from error88 if not selected:89 raise PlanValidationError(location, '目录或通配符没有匹配的输入文件')90 selected.sort(key=natural_path_key)91 # 不重排显式列表、不合并不同声明项，也不因内容相同而删除真实文件。92 result.extend(selected)93 shifts.extend([shift] * len(selected))94 if len(result) > MAX_INPUT_FILES:95 raise PlanValidationError(location, '输入文件超过4096项')96 if has_shifts and 'offsets' in raw:97 raise PlanValidationError(where, '文件内offset与来源offsets不能同时填写')98 return tuple(result), tuple(shifts) if has_shifts else None99100101def expand_input_paths(raw, base, where):102 """只取展开文件路径，供无需文件偏移的调用方使用。103104 参数: 来源配置、基准目录、错误位置。返回: Path元组。105 异常: 与expand_source_files一致；目录扫描只执行一次。106 """107 return expand_source_files(raw, base, where)[0]108109110def source_path_fields(source, where='$.sources'):111 """枚举文件声明中的路径文字，供转换/导入工具固定路径基准。112113 参数: source为来源字典，where为错误位置。返回: (容器,键)迭代器。114 异常: PlanValidationError为错误结构。不扫描目录，不改变offset或其他元数据。115 """116 for key in ('path', 'paths'):117 if key not in source:118 continue119 value = source[key]120 if type(value) is str:121 yield source, key122 continue123 # 每项可以是路径文字或带自己偏移的文件声明，工具不能把对象转成字符串。124 entries = [value] if type(value) is dict else array_at(value, where + '.' + key)125 for index, item in enumerate(entries):126 if type(item) is dict:127 object_at(item, f'{where}.{key}[{index}]', ('path',), ('offset',))128 yield item, 'path'129 else:130 text_at(item, f'{where}.{key}[{index}]')131 yield entries, index132