chronaxis.planning.diagnostic_collection
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""一次收集独立配置错误，并阻断依赖坏声明的检查。23输入是已解析配置，输出为定位诊断或完整计划。Schema只负责结构与单字段约束；4语义仍由原解析器检查。只跳过无效声明，不改配置、不补假值，不创建可运行残缺计划。5上下文仅覆盖本次同步构建，退出后恢复；schema来自唯一维护的Docs/Schema，安装时随包提供。6"""7from contextlib import contextmanager8from contextvars import ContextVar9from functools import lru_cache, wraps10from pathlib import Path11import json1213from jsonschema import Draft202012Validator, validators14from ..configuration.models import field_path_text15from .checks import PlanValidationError161718_ACTIVE = ContextVar('chronaxis_configuration_diagnostics', default=None)192021@lru_cache(maxsize=4)22def schema_validator(name):23 """加载只读结构规范。参数: 固定规范名。返回: validator。异常: 安装缺资源或规范损坏。2425 原生整数检查与typed API一致，不让bool或1.0冒充整数；不访问网络引用。26 """27 package = Path(__file__).resolve().parents[1]28 path = package / 'schemas' / name29 if not path.is_file():30 path = package.parents[1] / 'Docs' / 'Schema' / name31 schema = json.loads(path.read_text(encoding='utf-8-sig'))32 checker = Draft202012Validator.TYPE_CHECKER.redefine('integer',33 lambda checker, value: type(value) is int)34 return validators.extend(Draft202012Validator, type_checker=checker)(schema)353637def _host_validator(validator, extensions):38 """在本次验证器中放行宿主已登记的操作名。参数: 原规范、注册。返回: 独立验证器。异常: 注册非法。3940 只复制变化的字典路径；内置操作和无注册的 CLI 始终使用原规范，不改变共享缓存。41 """42 schema = validator.schema43 sections = dict(schema['properties'])44 for section, names in (('calculations', extensions.reductions), ('transforms', extensions.transforms)):45 if not names:46 continue47 array = sections[section]48 item = array['items']49 operations = item['properties']['operation']50 properties = dict(item['properties'], operation=dict(operations, enum=[*operations['enum'], *names]))51 sections[section] = dict(array, items=dict(item, properties=properties))52 return type(validator)(dict(schema, properties=sections))535455def _specific_errors(error):56 """展开最适用联合分支，避免列出不相关格式的全部缺项。5758 参数: jsonschema错误。返回: 叶错误迭代器。异常: 无。59 优先匹配已有字段的const/enum判别，再选择错误最少的分支；未知判别值保留父错误。60 """61 if error.validator not in ('oneOf', 'anyOf') or not error.context:62 yield error63 return64 groups = {}65 for child in error.context:66 groups.setdefault(child.schema_path[0], []).append(child)67 instance = error.instance68 candidates = []69 for index, children in groups.items():70 branch = error.validator_value[index]71 score = 072 if 'type' in branch:73 # 先按输入真实类型选分支，不能把坏对象误诊为“应填写字符串”。74 types = branch['type'] if type(branch['type']) is list else [branch['type']]75 matching = any(Draft202012Validator.TYPE_CHECKER.is_type(instance, name) for name in types)76 score += 1000 if matching else -100077 for key, rule in branch.get('properties', {}).items():78 if type(instance) is dict and key in instance:79 if 'const' in rule:80 score += 10 if instance[key] == rule['const'] else -10081 elif 'enum' in rule:82 score += 5 if instance[key] in rule['enum'] else -10083 candidates.append((score, -len(children), index, children))84 chosen = max(candidates, key=lambda item: item[:3])[-1]85 for child in chosen:86 yield from _specific_errors(child)878889class DiagnosticCollection:90 """持有本次有限错误、阻断说明和无效对象；不保存输入或运行结果。"""9192 def __init__(self, config, schema_name='analysis.schema.json', *, extensions=None):93 """检查所有独立结构字段。参数: 原配置和规范名。返回: 新集合。异常: 资源错误。"""94 self.errors = []95 self.blocked = []96 self.invalid_names = set()97 self.invalid_nodes = set()98 self.invalid_sources = set()99 self.checking_derived = False100 validator = schema_validator(schema_name)101 if extensions is not None:102 from ..extensions import ExtensionRegistry103 if type(extensions) is not ExtensionRegistry or type(config) is not dict or config.get('config_version') != 4:104 raise PlanValidationError('$.config_version','宿主扩展需要版本4和ExtensionRegistry')105 validator = _host_validator(validator, extensions)106 for error in validator.iter_errors(config):107 for issue in _specific_errors(error):108 path = tuple(issue.absolute_path)109 code = 'invalid_config'110 # 未专门处理的Schema约束也使用项目文案，不透传校验器内部英文。111 message = '值或字段组合不符合当前配置要求'112 hint = '按字段参考修正此值'113 if issue.validator == 'required':114 missing = sorted(key for key in issue.validator_value if key not in issue.instance)115 for key in missing:116 self.add(PlanValidationError(field_path_text((*path, key)),117 '缺少必填字段：' + key, 'invalid_config' if 'field' in path else 'missing_field', hint='在所属对象中补充 ' + key))118 continue119 if issue.validator == 'additionalProperties' and type(issue.instance) is dict:120 from difflib import get_close_matches121 unknown = set(issue.instance) - set(issue.schema.get('properties', {}))122 for key in sorted(unknown):123 candidate = get_close_matches(key, list(issue.schema.get('properties', {})), n=1, cutoff=0.65)124 if key in ('workers', 'max_workers') and 'process_count' in issue.schema.get('properties', {}):125 candidate = ['process_count']126 if key == 'cache_policy' and path and path[0] == 'message_calculations':127 self.add(PlanValidationError(field_path_text((*path, key)),128 '旧缓存策略尚未接线；当前仅支持current_message-v1', 'capability_unavailable'))129 else:130 known_in_other_mode = path and path[-1] == 'time' and key in ('clock', 'timezone')131 self.add(PlanValidationError(field_path_text((*path, key)),132 ('此时间模式不适用字段：' if known_in_other_mode else '未知字段：') + key,133 'invalid_config' if 'field' in path or known_in_other_mode else 'unknown_field',134 hint=('是否要写 ' + candidate[0] + '？' if candidate else '检查拼写或删除不适用字段')))135 continue136 if issue.validator == 'type':137 type_names = {'object': '对象', 'array': '列表', 'string': '字符串',138 'integer': '整数', 'number': '数值', 'boolean': '布尔值', 'null': '空值'}139 expected = issue.validator_value if type(issue.validator_value) is list else [issue.validator_value]140 code, message = 'invalid_type', '值类型应为' + '或'.join(type_names.get(value, '规定类型') for value in expected)141 if path[-2:] == ('conversion', 'input_unit') and issue.instance is None:142 message = 'input_unit不能填写null；需要自动确定输入单位时请省略此字段'143 hint = '删除input_unit字段；如需手动指定，请填写非空单位字符串'144 elif issue.validator in ('enum', 'const'):145 code = 'invalid_choice'146 message = '值必须属于 ' + str(issue.validator_value)147 values = issue.validator_value if issue.validator == 'enum' else [issue.validator_value]148 if len(values) == 1 and path:149 label = '.'.join(str(part) for part in path[-2:]) if path[-1] == 'mode' else str(path[-1])150 message = label + '必须为' + str(values[0])151 if path and path[-1] == 'source_type':152 code = 'capability_unavailable'153 elif issue.validator in ('minimum', 'maximum', 'exclusiveMinimum', 'exclusiveMaximum'):154 code = 'out_of_range'155 relation = {'minimum': '大于或等于', 'maximum': '小于或等于',156 'exclusiveMinimum': '大于', 'exclusiveMaximum': '小于'}[issue.validator]157 message = '数值必须' + relation + str(issue.validator_value)158 elif issue.validator in ('minLength', 'maxLength', 'minItems', 'maxItems'):159 subject = '文本长度' if issue.validator.endswith('Length') else '列表项数'160 relation = '至少为' if issue.validator.startswith('min') else '最多为'161 message = subject + relation + str(issue.validator_value)162 elif issue.validator == 'not':163 forbidden = issue.validator_value.get('required', ()) if type(issue.validator_value) is dict else ()164 if type(issue.instance) is dict and len(forbidden) == 1:165 path = (*path, forbidden[0])166 message = '当前操作或配置下不能填写字段：' + str(forbidden[0])167 elif type(issue.instance) is dict and forbidden:168 message = '不能同时填写字段：' + '、'.join(forbidden)169 elif issue.validator_value == {} and path:170 message = '当前操作或配置下不能填写字段：' + str(path[-1])171 else:172 message = '此值或字段组合不适用于当前配置'173 hint = '删除不适用的字段，或按所选操作的字段说明调整配置'174 elif issue.validator == 'pattern':175 message = '文本格式不符合字段要求：' + str(issue.validator_value)176 if path and (path[-1].endswith('duration') or path[-1] in ('valid_for','period','pre','post','min_delay','max_delay')):177 code = 'invalid_duration'178 if path and path[-1] == 'config_version':179 code = 'invalid_choice'180 if path and path[-1] == 'anchor':181 code = 'invalid_config'182 # 原定位模型使用统一invalid_config；保留机器码，位置细化到真实坏字段。183 if code not in ('unknown_field', 'missing_field') and (184 'field' in path or 'update_message' in path or 'dbc_by_channel' in path or185 (path and path[0] == 'message_calculations' and 'parameters' in path)):186 code = 'invalid_config'187 if path and path[-1] == 'channel':188 message = '通道必须为从1开始的原生整数'189 self.add(PlanValidationError(field_path_text(path), message, code, hint=hint))190191 def add(self, error):192 """登记唯一错误。参数: PlanValidationError。返回: None。异常: 无。"""193 if not any(item.path == error.path and item.code == error.code for item in self.errors):194 self.errors.append(error)195196 def usable(self, path):197 """判断对象是否有结构错误。参数: JSON路径。返回: bool。异常: 无。"""198 return not any(item.path == path or item.path.startswith(path + '.') or199 item.path.startswith(path + '[') for item in self.errors)200201 def block(self, path, reason):202 """记录依赖阻断，避免派生假错误。参数: 字段路径/原因。返回: None。异常: 无。"""203 record = dict(path=path, reason=reason)204 if record not in self.blocked:205 self.blocked.append(record)206207 @contextmanager208 def stage(self, path, *, requires=True, check=True):209 """隔离一个有明确依赖的检查。参数: 路径/依赖是否有效。返回: 是否执行。异常: 非配置错误传播。"""210 if check and not self.usable(path):211 yield False212 return213 if not requires:214 self.block(path, '依赖的声明无效，修复上游后继续检查')215 yield False216 return217 try:218 yield True219 except PlanValidationError as error:220 if error.code == 'blocked_dependency':221 self.block(error.path, error.message)222 else:223 self.add(error)224225 def finish(self):226 """有错则统一抛出，不返回残缺计划。参数: 无。返回: None。异常: PlanValidationError。"""227 if self.errors:228 # 拼写建议先于其缺项；操作名决定后续填写规则，先于其他独立字段错误。229 def priority(error):230 """参数: 单条诊断。返回: 根因优先级。异常: 无。"""231 if error.code == 'unknown_field':232 return 0233 if error.path.endswith('.operation') and error.code in ('missing_field', 'invalid_choice', 'invalid_type'):234 return 1235 return 3 if error.code == 'missing_field' else 2236 self.errors.sort(key=priority)237 raise PlanValidationError.from_diagnostics(self.errors, self.blocked)238239240def current_collection():241 """返回当前构建的集合。参数: 无。返回: 集合或None。异常: 无。"""242 return _ACTIVE.get()243244245@contextmanager246def collecting(collection):247 """临时接入本次构建。参数: 集合。返回: 无。异常: 原异常传播，始终恢复上下文。"""248 token = _ACTIVE.set(collection)249 try:250 yield251 finally:252 _ACTIVE.reset(token)253254255def usable_node(path, raw=None):256 """跳过坏节点且记录名称供依赖判断。参数: 路径/原声明。返回: bool。异常: 无。"""257 active = current_collection()258 if active is None:259 return True260 if active.usable(path):261 # 引用坏规则或窗口时，只说明未检查，不再伪报该名称不存在。262 if type(raw) is dict:263 for key in ('expr', 'condition'):264 try:265 require_expression_dependencies(raw.get(key), path + '.' + key, ())266 except PlanValidationError as error:267 active.block(error.path, error.message)268 if type(raw.get('name')) is str:269 active.invalid_nodes.add(raw['name'])270 return False271 references = [raw.get(key) for key in ('rule', 'window', 'start', 'response')]272 scope = raw.get('scope')273 if type(scope) is str and ':' in scope:274 references.append(scope.split(':', 1)[1])275 if any(type(name) is str and name in active.invalid_nodes for name in references):276 active.block(path, '引用的规则或窗口无效')277 else:278 return True279 else:280 return True281 if path.startswith('$.sources['):282 import re283 index = re.match(r'\$\.sources\[(\d+)\]', path)284 name = raw.get('name') if type(raw) is dict else None285 active.invalid_sources.add(name if type(name) is str else 'source_' + str(int(index[1]) + 1))286 elif type(raw) is dict and type(raw.get('name')) is str:287 names = active.invalid_names if path.startswith(('$.signals[', '$.derived.')) else active.invalid_nodes288 names.add(raw['name'])289 return False290291292def collect_node_error(error, path, raw):293 """收集单节点失败；非汇总模式保持原抛出行为。参数: 异常/路径/声明。返回: None。异常: 原错误。"""294 active = current_collection()295 if active is None:296 raise error297 if error.code == 'blocked_dependency':298 active.block(error.path, error.message)299 if type(raw) is dict and type(raw.get('name')) is str:300 names = active.invalid_names if path.startswith(('$.signals[', '$.derived.')) else active.invalid_nodes301 names.add(raw['name'])302 return303 active.add(error)304 usable_node(path, raw)305306307def require_expression_dependencies(text, path, derived):308 """遇到已知坏信号只阻断当前表达式。参数: 文本/位置/派生定义。返回: None。异常: 依赖阻断。309310 只读取AST名称，不执行表达式；真正的语法、dtype和单位仍由表达式编译器检查。311 """312 import ast313 active = current_collection()314 if active is None or not active.invalid_names or type(text) is not str:315 return316 definitions = {item.name: item.expr for item in derived}317 visited = set()318 pending = [text]319 while pending:320 value = pending.pop()321 try:322 names = {node.id for node in ast.walk(ast.parse(value, mode='eval')) if isinstance(node, ast.Name)}323 except SyntaxError:324 return325 bad = names & active.invalid_names326 if bad:327 raise PlanValidationError(path, '依赖的信号或派生项无效：' + ', '.join(sorted(bad)), 'blocked_dependency')328 for name in names - visited:329 visited.add(name)330 if name in definitions:331 pending.append(definitions[name])332333334def expression_scope(text, bindings, derived):335 """给单项静态诊断选择真实依赖，避免无关坏派生项遮蔽此项。336337 参数: 表达式/声明绑定/全部派生。返回: 仅用于静态诊断的绑定和派生。异常: 原模型错误。338 全计划仍逐项检查所有派生；此处不创建运行计划或修改输入配置。339 """340 active = current_collection()341 if active is None or (not active.checking_derived and not active.errors):342 return bindings, derived343 import ast344 from dataclasses import replace345 definitions = {item.name: item for item in derived}346 selected = set()347 pending = [text]348 while pending:349 value = pending.pop()350 if type(value) is not str:351 continue352 try:353 names = {node.id for node in ast.walk(ast.parse(value, mode='eval')) if isinstance(node, ast.Name)}354 except SyntaxError:355 continue356 for name in names & definitions.keys() - selected:357 selected.add(name)358 pending.append(definitions[name].expr)359 required = tuple(item for item in derived if item.name in selected)360 return replace(bindings, derived_names=tuple(item.name for item in required)), required361362363def diagnose_plan(function):364 """为计划入口建立唯一诊断上下文。参数: 构建方法。返回: 包装方法。异常: 完整配置错误。365366 合法输入只构建一次；错误输入从不返回局部计划，且始终恢复上层上下文。367 """368 @wraps(function)369 def checked(cls, config, *args, **kwargs):370 """调用一次构建。参数: 原配置及路径参数。返回: 完整计划。异常: 汇总配置错误。"""371 from .checks import require_json_value372 require_json_value(config, '$')373 collection = DiagnosticCollection(config, extensions=kwargs.get('extensions'))374 with collecting(collection):375 try:376 result = function(cls, config, *args, **kwargs)377 except PlanValidationError as error:378 if tuple(collection.errors) == error.diagnostics:379 raise380 for item in error.diagnostics:381 collection.add(item)382 collection.finish()383 collection.finish()384 return result385 return checked386