华为OD机试真题 新系统 2026-07-12 JavaGoC 实现【软件依赖树】
2026/7/17 16:14:33
影刀RPA 数据字典自动生成：从数据库到文档
作者：林焱 | 分类：影刀RPA新手教程 | 难度：★★
什么情况用
项目做了半年，数据库有100+张表、几千个字段，没有任何文档说明每张表是干嘛的、每个字段是什么意思。新人入职看代码靠猜，数据分析和报表开发到处问人。
用影刀RPA自动连接数据库，读取表结构、字段信息、注释、索引，生成结构化的数据字典（Excel或Markdown）。数据库结构变了自动更新文档。
怎么做
拼多多店群自动化上架方案
第一步：连接数据库获取表结构
# 影刀Python节点：获取MySQL表结构importpymysqlimportpandasaspd DB_CONFIG={"host":"192.168.1.100","port":3306,"user":"readonly_user","password":"***",# 从影刀凭证管理读取"database":"your_database","charset":"utf8mb4",}defget_all_tables(conn):"""获取所有表名和注释"""sql=""" SELECT TABLE_NAME, TABLE_COMMENT, TABLE_ROWS, CREATE_TIME, UPDATE_TIME FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s AND TABLE_TYPE = 'BASE TABLE' ORDER BY TABLE_NAME """withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute(sql,(DB_CONFIG["database"],))returncursor.fetchall()defget_table_columns(conn,table_name):"""获取指定表的所有字段信息"""sql=f""" SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY, -- PRI=主键, UNI=唯一索引, MUL=普通索引 EXTRA, -- auto_increment 等 COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute(sql,(DB_CONFIG["database"],table_name))returncursor.fetchall()defget_table_indexes(conn,table_name):"""获取表的索引信息"""sql=f"SHOW INDEX FROM `{table_name}`"withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute(sql)returncursor.fetchall()# 连接数据库conn=pymysql.connect(**DB_CONFIG)tables=get_all_tables(conn)# 遍历收集所有信息data_dict=[]fortableintables:table_name=table["TABLE_NAME"]columns=get_table_columns(conn,table_name)indexes=get_table_indexes(conn,table_name)forcolincolumns:# 判断是否为主键is_pk="是"ifcol["COLUMN_KEY"]=="PRI"else"否"# 是否为索引is_indexed="是"ifcol["COLUMN_KEY"]in("PRI","UNI","MUL")else"否"# 是否自增is_auto="是"if"auto_increment"incol.get("EXTRA","")else"否"data_dict.append({"表名":table_name,"表注释":table["TABLE_COMMENT"],"字段名":col["COLUMN_NAME"],"字段类型":col["COLUMN_TYPE"],"允许空":col["IS_NULLABLE"],"默认值":col.get("COLUMN_DEFAULT",""),"主键":is_pk,"有索引":is_indexed,"自增":is_auto,"字段注释":col.get("COLUMN_COMMENT",""),"表行数":table.get("TABLE_ROWS",0),"创建时间":str(table.get("CREATE_TIME","")),"更新时间":str(table.get("UPDATE_TIME","")),})conn.close()df=pd.DataFrame(data_dict)print(f"共{len(tables)}张表，{len(data_dict)}个字段")第二步：生成Excel数据字典
# 影刀Python节点：生成格式化的ExcelfromopenpyxlimportWorkbookfromopenpyxl.stylesimportFont,PatternFill,Alignment,Border,Sidefromopenpyxl.utilsimportget_column_letterdefgenerate_excel_dictionary(dataframe,output_path):"""生成带样式格式的数据字典Excel"""wb=Workbook()ws=wb.active ws.title="数据字典"# 表头样式header_font=Font(bold=True,color="FFFFFF",size=11)header_fill=PatternFill(start_color="2F5496",end_color="2F5496",fill_type="solid")pk_fill=PatternFill(start_color="FFF2CC",end_color="FFF2CC",fill_type="solid")# 主键高亮thin_border=Border(left=Side(style="thin"),right=Side(style="thin"),top=Side(style="thin"),bottom=Side(style="thin"),)# 写表头headers=list(dataframe.columns)forcol_idx,headerinenumerate(headers,1):cell=ws.cell(row=1,column=col_idx,value=header)cell.font=header_font cell.fill=header_fill cell.alignment=Alignment(horizontal="center",vertical="center")cell.border=thin_border# 写数据forrow_idx,rowinenumerate(dataframe.itertuples(index=False),2):forcol_idx,valueinenumerate(row,1):cell=ws.cell(row=row_idx,column=col_idx,value=value)cell.border=thin_border cell.alignment=Alignment(vertical="center")# 主键行黄色高亮ifcol_idx==7andvalue=="是":# 第7列是"主键"cell.fill=pk_fill# 自适应列宽forcol_idxinrange(1,len(headers)+1):max_width=max(len(str(ws.cell(row=row_idx,column=col_idx).valueor""))forrow_idxinrange(1,len(dataframe)+2))ws.column_dimensions[get_column_letter(col_idx)].width=min(max_width+4,40)# 冻结首行ws.freeze_panes="A2"# 添加筛选ws.auto_filter.ref=f"A1:{get_column_letter(len(headers))}{len(dataframe)+1}"wb.save(output_path)print(f"数据字典已生成：{output_path}")generate_excel_dictionary(df,"数据字典.xlsx")第三步：生成Markdown文档
# 影刀Python节点：生成Markdown格式（方便放wiki）defgenerate_markdown_dictionary(dataframe,output_path):"""生成Markdown格式的数据字典"""md="# 数据库字典\n\n"md+=f"> 自动生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}\n"md+=f"> 数据库：{DB_CONFIG['database']}\n"md+=f"> 总表数：{len(dataframe['表名'].unique())}| 总字段数：{len(dataframe)}\n\n"# 按表分组fortable_name,groupindataframe.groupby("表名"):table_comment=group.iloc[0]["表注释"]row_count=group.iloc[0]["表行数"]md+=f"##{table_name}—{table_comment}\n\n"md+=f"数据行数：{row_count:,}\n\n"md+="| 字段名 | 类型 | 允许空 | 默认值 | 主键 | 索引 | 自增 | 注释 |\n"md+="|--------|------|--------|--------|------|------|------|------|\n"for_,rowingroup.iterrows():pk_mark="🔑"ifrow["主键"]=="是"else""idx_mark="📌"ifrow["有索引"]=="是"else""md+=f"|{row['字段名']}|{row['字段类型']}|{row['允许空']}|{row['默认值']}|{pk_mark}|{idx_mark}|{row['自增']}|{row['字段注释']}|\n"md+="\n"withopen(output_path,"w",encoding="utf-8")asf:f.write(md)print(f"Markdown字典已生成：{output_path}")generate_markdown_dictionary(df,"数据字典.md")第四步：检测变更并更新
# 影刀Python节点：对比新旧字典defcompare_schemas(old_dict,new_dict):"""检测数据库结构变化"""changes=[]# 新增表old_tables=set(old_dict["表名"].unique())new_tables=set(new_dict["表名"].unique())fortinnew_tables-old_tables:changes.append(f"➕ 新增表：{t}")fortinold_tables-new_tables:changes.append(f"➖ 删除表：{t}")# 新增/删除/变更字段fortableinold_tables&new_tables:old_fields=set(old_dict[old_dict["表名"]==table]["字段名"])new_fields=set(new_dict[new_dict["表名"]==table]["字段名"])forfinnew_fields-old_fields:changes.append(f"➕{table}.{f}：新增字段")forfinold_fields-new_fields:changes.append(f"➖{table}.{f}：字段已删除")returnchanges# 影刀发现变更后【企微推送】通知DBA/开发团队有什么坑
坑1：需要只读数据库账号
数据字典读取需要的是SELECT权限，绝对不能给RPA创建/修改的权限。建一个专用的只读账号，即使流程出bug也不会误操作数据。
TEMU店群如何管理运营？
坑2：大库扫描可能很慢
如果数据库有500+张表，information_schema查询可能比较慢。建议在从库上执行，别在生产主库上扫描。
坑3：字段注释经常为空
开发写表结构时很少有人认真填COMMENT。如果大部分字段注释是空的，数据字典的价值就大打折扣。建议生成字典后推动团队补充注释——没有注释的字段黄色高亮标记"待补充"。
坑4：不同数据库语法差异
MySQL用information_schema，PostgreSQL用information_schema也兼容但细节不同，Oracle用ALL_TAB_COLUMNS。如果公司多种数据库并存，需要分别适配。
坑5：视图、存储过程、触发器也需要文档
数据字典通常只覆盖表和字段，但完整的数据库文档还应该包括视图定义、存储过程说明、触发器逻辑。这些需要单独写SQL提取。
总结：数据字典自动化的核心价值不是"一次性生成文档"，而是持续同步——数据库在变、文档自动跟。把字典生成做成定时任务（每周日更新一次），团队永远看到最新的数据库文档。