chronaxis.configuration.origins
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按请求说明定义或叶字段来自哪个配置文件；不作为执行计划的持久状态。23只读主配置与明确的一层公共定义，先复用静态校验再生成小型来源记录。4默认报告定义位置及来源文件路径；fields模式列全部显式叶字段，不虚构默认值来源。5不读取ASC/BLF/CSV中的数据记录；校验可能读取显式DBC以补齐信号元数据。6不写报告文件；调用者决定显示或重定向。7"""8from pathlib import Path910from . import load_configuration11from .imports import expand_document_imports12from .file_tools import validate_document13from .models import field_path_text141516def _at(data, path):17 """读取存在的字段，跳过展开后已删除的imports位置。参数: 原生树/tuple路径。返回: (存在,值)。异常: 无。"""18 current=data19 for key in path:20 if type(current) is dict and key in current:current=current[key]21 elif type(current) is list and type(key) is int and 0<=key<len(current):current=current[key]22 else:return False,None23 return True,current242526def _record(document, path, name=None):27 """复制一项有限来源事实。参数: 文档、路径、可选定义名。返回: dict。异常: 无。"""28 origin=document.origin_for(path)29 location=origin.location if origin else document.location_for(path)30 return dict(name=name,effective_path=field_path_text(path),31 declared_path=field_path_text(origin.field_path if origin else path),32 file=origin.source if origin else document.source,33 line=None if location is None else location.line,34 column=None if location is None else location.column)353637def configuration_origins(source, *, fields=False):38 """读取实际定义或叶字段的声明位置。参数: 配置路径、是否逐字段。返回: 来源报告dict。异常: 配置/读取错误。3940 分析配置列其定义；批量只列本文件的任务，不读取子计划。省略的默认字段41 不在报告内。fields只保位置而不复制值；路径声明单独说明最终指向。42 """43 path=Path(source).absolute()44 document=expand_document_imports(load_configuration(path))45 kind=validate_document(document,path.parent)46 items=[]47 if fields:48 for parts in document.locations:49 exists,value=_at(document.data,parts)50 if exists and type(value) not in (dict,list):items.append(_record(document,parts))51 else:52 analysis=document.data53 if kind=='batch' and 'template' in analysis:54 item=_record(document,('template',),'template')55 item['resolved_paths']=[str((path.parent/analysis['template']).absolute())]56 items.append(item)57 items.append(_record(document,('inputs',),'inputs'))58 sections=('jobs',) if kind=='batch' else ('sources','signals','derived')59 for section in sections:60 values=analysis.get(section,{} if section=='derived' else [])61 entries=values.items() if section=='derived' else enumerate(values)62 for key,value in entries:63 if section=='signals' and 'signals' in value:64 for child_index,child in enumerate(value['signals']):65 items.append(_record(document,(section,key,'signals',child_index),child['name']))66 continue67 name=key if section=='derived' else value.get('name')68 item=_record(document,(section,key),name)69 # 文件位置只按已知路径字段说明；metadata内路径文字不解释。70 if section=='sources':71 from ..planning.input_paths import source_path_fields72 declared=[owner[key] for owner,key in source_path_fields(value)]73 declared.extend(value.get('dbc_by_channel',{}).values())74 item['resolved_paths']=[str((path.parent/entry).absolute()) for entry in declared]75 elif section=='jobs':item['resolved_paths']=[str((path.parent/value['plan']).absolute())]76 items.append(item)77 return dict(configuration=str(path),configuration_type=kind,78 scope='fields' if fields else 'definitions',items=items)798081def render_origins(report):82 """将有限来源记录显示为中文文本。参数: configuration_origins报告。返回: str。异常: 内部报告结构错误。"""83 lines=['配置来源：'+report['configuration'],84 '只列显式声明；未读取数据记录。批量任务的子配置需单独查看。']85 for item in report['items']:86 place=item['file']87 if item['line'] is not None:place+=f":{item['line']}:{item['column']}"88 lines.append(item['effective_path']+' ← '+place+' '+item['declared_path'])89 for path in item.get('resolved_paths',[]):lines.append(' 文件路径：'+path)90 return '\n'.join(lines)+'\n'91