chronaxis.configuration.yaml_parser
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""通过独立纯Python安全解析器读取YAML1.2受限子集，不构造任意标签对象。"""2from math import isfinite34from .models import ConfigurationError, ConfigurationLocation5from .json_parser import MAX_DEPTH678def parse_yaml(text, source):9 """用事件流构造一次原生值树。参数: 文本/来源。返回: 数据和位置表。异常: 受限语法错误。"""10 # JSON入口不依赖YAML导入；每次新实例，失败或并发加载不共享解析器状态。11 from ruamel.yaml import YAML12 from ruamel.yaml.events import (StreamStartEvent,StreamEndEvent,DocumentStartEvent,13 DocumentEndEvent,ScalarEvent,SequenceStartEvent,SequenceEndEvent,14 MappingStartEvent,MappingEndEvent,AliasEvent)15 from ruamel.yaml.nodes import ScalarNode16 from ruamel.yaml.error import YAMLError1718 class Yaml12(YAML):19 """版本在交给依赖校验前受限，避免依赖对1.3等版本抛出AssertionError。"""2021 @YAML.version.setter22 def version(self, selected):23 """只允许默认或1.2。参数: 解析器版本。返回: None。异常: 明确配置版本错误。"""24 if selected is not None and tuple(selected)!=(1,2):25 # 依赖在文档起点事件前设版本；没有可靠mark时只报告文件，不猜行号。26 raise ConfigurationError(source,(),None,'yaml_version','只支持YAML 1.2')27 YAML.version.fset(self,selected)2829 parser=Yaml12(typ='safe',pure=True)30 parser.version=(1,2)31 parser.allow_duplicate_keys=False32 positions={}33 events=parser.parse(text)3435 def location(event):36 """复制事件位置，不让结果引用整份输入文本。参数: 事件。返回: 标量位置。异常: 无。"""37 return ConfigurationLocation(event.start_mark.line+1,event.start_mark.column+1)3839 def reject(event,path,code,message):40 """抛定位错误。参数: 事件/路径/代码/原因。返回: 无。异常: ConfigurationError。"""41 raise ConfigurationError(source,path,location(event),code,message)4243 def check(event,path):44 """在构造前拒绝别名、锚点和显式标签。返回: None。异常: 定位配置错误。"""45 if isinstance(event,AliasEvent):reject(event,path,'yaml_alias','不支持YAML别名；请直接写出配置内容')46 if getattr(event,'anchor',None) is not None:reject(event,path,'yaml_anchor','不支持YAML锚点')47 if getattr(event,'tag',None) is not None:reject(event,path,'yaml_tag','不支持显式标签；字符串请加引号')4849 def scalar(event,path):50 """仅转换JSON原生标量；日期时间保文本。返回: 原生值。异常: 不支持类型或非有限数。"""51 check(event,path)52 tag=str(parser.resolver.resolve(ScalarNode,event.value,event.implicit))53 if tag in ('tag:yaml.org,2002:str','tag:yaml.org,2002:timestamp',54 'tag:yaml.org,2002:merge','tag:yaml.org,2002:value'):55 # <<只在未加引号的映射键位置有合并含义；普通值仍是原始文本。56 return event.value57 if tag not in {'tag:yaml.org,2002:null','tag:yaml.org,2002:bool',58 'tag:yaml.org,2002:int','tag:yaml.org,2002:float'}:59 reject(event,path,'yaml_scalar','仅支持文本、布尔、有限数字和null')60 # 报文ID只接受约定的三种进制；不能让YAML的0o或下划线数字先转换后绕过检查。61 if path and path[-1] == 'frame_id' and tag == 'tag:yaml.org,2002:int':62 from ..legacy.records import normalize_frame_id63 try:64 return normalize_frame_id(event.value)65 except (TypeError, ValueError) as error:66 reject(event, path, 'invalid_frame_id', str(error))67 # 直接调用安全标量构造，不创建带缓存的YAML节点树或任意Python对象。68 try:69 value=parser.constructor.yaml_constructors[tag](parser.constructor,ScalarNode(tag,event.value))70 except (ValueError,OverflowError) as error:71 raise ConfigurationError(source,path,location(event),'invalid_number',72 '数字超出当前Python可解析的范围') from error73 if type(value) is float and not isfinite(value):74 reject(event,path,'nonfinite_number','数值必须有限，不能使用NaN或Infinity')75 return value7677 def value(event,path=(),depth=0):78 """递归消费一次事件流并保位置。返回: 原生值。异常: 受限语法或嵌套超限。"""79 check(event,path)80 if depth>MAX_DEPTH:reject(event,path,'nesting_limit','配置嵌套不能超过64层')81 positions[path]=location(event)82 if isinstance(event,ScalarEvent):return scalar(event,path)83 if isinstance(event,SequenceStartEvent):84 result=[]85 for child in events:86 if isinstance(child,SequenceEndEvent):return result87 result.append(value(child,(*path,len(result)),depth+1))88 elif isinstance(event,MappingStartEvent):89 result={}90 for key_event in events:91 if isinstance(key_event,MappingEndEvent):return result92 check(key_event,path)93 if not isinstance(key_event,ScalarEvent):reject(key_event,path,'non_string_key','配置对象的键必须为字符串')94 if key_event.value=='<<' and key_event.style is None:95 reject(key_event,path,'yaml_merge','不支持YAML合并键')96 key=scalar(key_event,path)97 if type(key) is not str:reject(key_event,path,'non_string_key','配置对象的键必须为字符串；数字键请加引号')98 child_path=(*path,key)99 if key in result:reject(key_event,child_path,'duplicate_key',f'重复键：{key}')100 result[key]=value(next(events),child_path,depth+1)101 reject(event,path,'invalid_yaml','配置值结构不完整')102103 try:104 if not isinstance(next(events),StreamStartEvent):raise AssertionError('YAML解析器缺少流起点')105 start=next(events)106 if isinstance(start,StreamEndEvent):return None,{():ConfigurationLocation(1,1)}107 if not isinstance(start,DocumentStartEvent):reject(start,(),'invalid_yaml','需要一个YAML文档')108 if start.version not in (None,(1,2)):reject(start,(),'yaml_version','只支持YAML 1.2')109 if start.tags:reject(start,(),'yaml_tag','不支持标签指令')110 data=value(next(events))111 end=next(events)112 if not isinstance(end,DocumentEndEvent):reject(end,(),'invalid_yaml','文档结束位置不正确')113 end=next(events)114 if not isinstance(end,StreamEndEvent):reject(end,(),'multiple_documents','一个配置文件只能包含一个YAML文档')115 return data,positions116 except YAMLError as error:117 mark=getattr(error,'problem_mark',None)118 place=None if mark is None else ConfigurationLocation(mark.line+1,mark.column+1)119 offset=getattr(error,'position',None)120 if place is None and type(offset) is int:121 place=ConfigurationLocation(text.count('\n',0,offset)+1,offset-text.rfind('\n',0,offset))122 message=getattr(error,'problem',None) or getattr(error,'reason',None) or 'YAML语法无效'123 raise ConfigurationError(source,(),place,'invalid_yaml',message) from error124 finally:125 events.close()126