如何用Python解析SQL脚本生成字段-业务名称-来源表映射表

sqlparse完全可以满足你的需求，它能帮你拆解SQL的语法结构，提取SELECT子句里的字段、业务别名以及关联的来源表信息，进而生成符合要求的映射表。下面是具体的实现思路和示例代码：

核心实现步骤 #

• 解析SQL语句：通过sqlparse.parse()将SQL字符串转换为可遍历的语法树对象
• 定位关键子句：遍历语法树，筛选出SELECT字段列表和FROM关联表部分
• 提取字段映射关系：对每个SELECT字段，拆分原始字段名、业务别名（含AS定义的情况），同时匹配对应的来源表
• 整合结构化结果：把提取到的信息整理成统一格式的映射表，方便业务人员查询使用

示例代码 #

假设你的视图SQL如下：

SELECT 
    act.id AS 活动ID,
    act.name AS 活动名称,
    act.start_date AS 开始时间,
    org.org_name AS 所属组织
FROM ods_ACT_Master act
LEFT JOIN ods_ORG_Master org ON act.org_id = org.id

对应的解析代码：

import sqlparse
from sqlparse.sql import IdentifierList, Identifier
from sqlparse.tokens import Keyword, DML

def extract_field_mapping(sql):
    parsed = sqlparse.parse(sql)[0]
    mapping = []
    select_clause = None
    from_clause = None

    # 遍历语法树定位SELECT和FROM子句
    for token in parsed.tokens:
        if token.ttype == DML and token.value.upper() == 'SELECT':
            select_clause = token.next_sibling
        elif token.ttype == Keyword and token.value.upper() == 'FROM':
            from_clause = token.next_sibling

    # 提取SELECT字段的映射关系
    if isinstance(select_clause, IdentifierList):
        for item in select_clause.get_identifiers():
            # 处理带别名的字段
            if item.has_alias():
                original_field = item.value.split(' AS ')[0].strip()
                # 拆分来源表前缀和字段名
                if '.' in original_field:
                    source_table_alias = original_field.split('.')[0]
                    field_name = original_field.split('.')[1]
                else:
                    # 默认取FROM子句第一个表的别名/表名
                    first_table = from_clause.get_identifiers()[0]
                    source_table_alias = first_table.get_alias() or first_table.value
                    field_name = original_field
                business_name = item.get_alias()
                mapping.append({
                    '字段名': field_name,
                    '业务名称': business_name,
                    '来源表别名': source_table_alias
                })
            # 处理无别名的字段
            else:
                original_field = item.value.strip()
                if '.' in original_field:
                    source_table_alias = original_field.split('.')[0]
                    field_name = original_field.split('.')[1]
                else:
                    first_table = from_clause.get_identifiers()[0]
                    source_table_alias = first_table.get_alias() or first_table.value
                    field_name = original_field
                mapping.append({
                    '字段名': field_name,
                    '业务名称': field_name,
                    '来源表别名': source_table_alias
                })

    # 建立表别名与真实表名的映射
    table_alias_map = {}
    if isinstance(from_clause, IdentifierList):
        for table in from_clause.get_identifiers():
            if table.has_alias():
                real_table_name = table.value.split(' ')[0]
                table_alias_map[table.get_alias()] = real_table_name
            else:
                table_alias_map[table.value] = table.value

    # 替换来源表别名为真实表名
    for item in mapping:
        if item['来源表别名'] in table_alias_map:
            item['来源表'] = table_alias_map.pop(item['来源表别名'])
            del item['来源表别名']

    return mapping

# 测试示例
sample_sql = """
SELECT 
    act.id AS 活动ID,
    act.name AS 活动名称,
    act.start_date AS 开始时间,
    org.org_name AS 所属组织
FROM ods_ACT_Master act
LEFT JOIN ods_ORG_Master org ON act.org_id = org.id
"""

result = extract_field_mapping(sample_sql)
for item in result:
    print(item)

适配复杂场景的注意点 #

• 函数计算字段：如果SQL包含COUNT(act.id) AS 活动数量这类带函数的字段，需要额外解析函数内部的原始字段，可通过正则匹配或递归遍历语法树实现
• 嵌套子查询：若FROM子句包含子查询，需要先递归解析子查询的结构，再提取字段的最终来源表
• 大小写兼容：建议统一将SQL转换为大写（或小写）后再解析，避免关键字大小写不一致导致的识别错误
• 批量处理：可遍历所有视图的SQL脚本文件，批量调用上述函数，将结果整合到Excel或数据库表中，形成业务人员的统一查询工具

内容的提问来源于stack exchange，提问作者snow