importosimportreimportjsonimporttracebackfrompathlibimportPathimportsqlglotfromsqlglotimportexp# # 1. 核心解析引擎 (保持最强平铺逻辑)# classSQLDeepResolver:def__init__(self,dialectteradata):self.dialectdialect self.reset()defreset(self):self.internal_cte_map{}self.variables_map{}self.physical_sources[]self.global_varsset()def_pre_clean(self,sql_text):sqlre.sub(r^\s*[\.\\].*$,,sql_text,flagsre.MULTILINE)sqlre.sub(r/\*.*?\*/,,sql,flagsre.DOTALL)sqlre.sub(r--.*,,sql)vars_foundre.findall(r(#[A-Za-z0-9_]#|\$\{[A-Za-z0-9_]\}),sql)forvinvars_found:safe_namefVAR_TOKEN_{re.sub(r[^A-Za-z0-9_],,v)}self.variables_map[safe_name]v self.global_vars.add(v)sqlsql.replace(v,safe_name)returnsql.strip()def_restore(self,text):forsafe,rawinself.variables_map.items():texttext.replace(safe,raw)returntextdef_get_logic_and_refs(self,node,scope_tables):def_expand(n):ifisinstance(n,exp.Column):col_namen.name.upper()table_aliasn.table.upper()ifn.tableelsereal_sourcescope_tables.get(table_alias,table_alias)ifnotreal_sourceandlen(scope_tables)1:real_sourcelist(scope_tables.values())[0]ifreal_sourceinself.internal_cte_map:cte_infoself.internal_cte_map[real_source]ifcol_nameincte_info:returnexp.Paren(thiscte_info[col_name][ast].copy())ifreal_source:n.set(table,exp.Identifier(thisreal_source))returnn flattened_nodenode.transform(_expand)s_refs{self._restore(f{c.table.upper()}.{c.name.upper()}ifc.tableelsec.name.upper())forcinflattened_node.find_all(exp.Column)}logic_sqlself._restore(flattened_node.sql(dialectself.dialect))p_refs[vfork,vinself.variables_map.items()ifkinlogic_sql]returnlogic_sql,list(s_refs),p_refs,flattened_nodedef_parse_select_block(self,select_node,block_alias):scope_tables{}fortableinselect_node.find_all(exp.Table):t_nameself._restore(table.name.upper())t_aliastable.alias.upper()iftable.aliaselset_name scope_tables[t_alias]t_nameift_namenotinself.internal_cte_mapandt_name!TARGET:ifnotany(s[table]t_nameforsinself.physical_sources):self.physical_sources.append({table:t_name,alias:t_alias})results,output{},[]foridx,exprinenumerate(select_node.expressions):aliasexpr.alias.upper()ifisinstance(expr,exp.Alias)else(expr.name.upper()ifhasattr(expr,name)elsefCOL_{idx})coreexpr.thisifisinstance(expr,exp.Alias)elseexpr logic,s_refs,p_refs,final_astself._get_logic_and_refs(core,scope_tables)win_depNoneiffinal_ast.find(exp.Window):wfinal_ast.find(exp.Window)win_dep{partition_by:[self._restore(c.sql())forcinw.find_all(exp.PartitionBy)],order_by:[self._restore(c.sql())forcinw.find_all(exp.Order)]}item{target_col:alias,data_format:(expr.find(exp.Cast).args.get(to).sql().upper()ifexpr.find(exp.Cast)elseINFERRED),is_constant:nots_refsandnotp_refs,logic:logic,source_refs:s_refs,parameter_refs:p_refs,window_dependency:win_dep}output.append(item);results[alias]{ast:final_ast}ifblock_alias:self.internal_cte_map[block_alias]resultsreturnoutputdefresolve(self,sql_content):self.reset()clean_sqlself._pre_clean(sql_content)try:astsqlglot.parse_one(clean_sql,readself.dialect)except:astsqlglot.parse_one(clean_sql)forcteinast.find_all(exp.CTE):self._parse_select_block(cte.this,cte.alias.upper())subquerieslist(ast.find_all(exp.Subquery));subqueries.reverse()forsqinsubqueries:ifsq.alias:self._parse_select_block(sq.this,sq.alias.upper())main_queryast.expressionifhasattr(ast,expression)elseastifisinstance(main_query,exp.Union):def_walk(u):res[]ifisinstance(u,exp.Union):res.extend(_walk(u.left));res.extend(_walk(u.right))else:res.append(self._parse_select_block(u))returnres branches_walk(main_query)lineage[{target_col:branches[0][i][target_col],is_union:True,branches:[b[i]forbinbranches]}foriinrange(len(branches[0]))]else:lineageself._parse_select_block(main_query)filters[]forwinast.find_all(exp.Where):f_log,f_s,f_p,_self._get_logic_and_refs(w.this,{})filters.append({logic:f_log,predicate_refs:f_s,parameter_refs:f_p})return{target_table:self._restore(ast.find(exp.Table).sql().upper())ifast.find(exp.Table)elseUNKNOWN,global_variables:list(self.global_vars),sources:self.physical_sources,column_lineage:lineage,filter_lineage:filters}# # 2. 批量处理逻辑 (新增错误日志导出)# # --- 配置区 ---SQL_DIR./your_sql_folder# SQL 源码目录OUTPUT_DIR./json_results# 成功 JSON 目录ERROR_LOG_PATH./failed_files.txt# 【新增】解析失败的文件清单defmain():ifnotos.path.exists(OUTPUT_DIR):os.makedirs(OUTPUT_DIR)resolverSQLDeepResolver(dialectteradata)sql_files[fforfinos.listdir(SQL_DIR)iff.lower().endswith(.sql)]print(f 开始处理{len(sql_files)}个文件...)# 清空旧的错误日志withopen(ERROR_LOG_PATH,w,encodingutf-8)aslog:log.write( SQL 解析失败清单 \n\n)success_count0fail_count0forfile_nameinsql_files:file_pathos.path.join(SQL_DIR,file_name)output_pathos.path.join(OUTPUT_DIR,file_name.rsplit(.,1)[0].json)try:withopen(file_path,r,encodingutf-8)asf:contentf.read()# 执行解析resultresolver.resolve(content)# 保存成功结果withopen(output_path,w,encodingutf-8)asf:json.dump(result,f,indent4,ensure_asciiFalse)success_count1print(f✅ [{success_countfail_count}] 成功:{file_name})exceptExceptionase:fail_count1# 【核心改进】将失败文件名和报错原因写入文件withopen(ERROR_LOG_PATH,a,encodingutf-8)aslog:log.write(f文件名:{file_name}\n)log.write(f错误原因:{str(e)}\n)log.write(-*50\n)print(f❌ [{success_countfail_count}] 失败:{file_name}(已记录至错误清单))print(\n*30)print(f 任务结束)print(f 成功:{success_count}个)print(f 失败:{fail_count}个)iffail_count0:print(f⚠️ 请查看错误清单:{os.path.abspath(ERROR_LOG_PATH)})print(*30)if__name____main__:main()