影刀RPA 数据字典自动生成：从数据库到文档
影刀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):获取指定表的所有字段信息sqlf 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):获取表的索引信息sqlfSHOW INDEX FROM {table_name}withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute(sql)returncursor.fetchall()# 连接数据库connpymysql.connect(**DB_CONFIG)tablesget_all_tables(conn)# 遍历收集所有信息data_dict[]fortableintables:table_nametable[TABLE_NAME]columnsget_table_columns(conn,table_name)indexesget_table_indexes(conn,table_name)forcolincolumns:# 判断是否为主键is_pk是ifcol[COLUMN_KEY]PRIelse否# 是否为索引is_indexed是ifcol[COLUMN_KEY]in(PRI,UNI,MUL)else否# 是否自增is_auto是ifauto_incrementincol.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()dfpd.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):生成带样式格式的数据字典ExcelwbWorkbook()wswb.active ws.title数据字典# 表头样式header_fontFont(boldTrue,colorFFFFFF,size11)header_fillPatternFill(start_color2F5496,end_color2F5496,fill_typesolid)pk_fillPatternFill(start_colorFFF2CC,end_colorFFF2CC,fill_typesolid)# 主键高亮thin_borderBorder(leftSide(stylethin),rightSide(stylethin),topSide(stylethin),bottomSide(stylethin),)# 写表头headerslist(dataframe.columns)forcol_idx,headerinenumerate(headers,1):cellws.cell(row1,columncol_idx,valueheader)cell.fontheader_font cell.fillheader_fill cell.alignmentAlignment(horizontalcenter,verticalcenter)cell.borderthin_border# 写数据forrow_idx,rowinenumerate(dataframe.itertuples(indexFalse),2):forcol_idx,valueinenumerate(row,1):cellws.cell(rowrow_idx,columncol_idx,valuevalue)cell.borderthin_border cell.alignmentAlignment(verticalcenter)# 主键行黄色高亮ifcol_idx7andvalue是:# 第7列是主键cell.fillpk_fill# 自适应列宽forcol_idxinrange(1,len(headers)1):max_widthmax(len(str(ws.cell(rowrow_idx,columncol_idx).valueor))forrow_idxinrange(1,len(dataframe)2))ws.column_dimensions[get_column_letter(col_idx)].widthmin(max_width4,40)# 冻结首行ws.freeze_panesA2# 添加筛选ws.auto_filter.reffA1:{get_column_letter(len(headers))}{len(dataframe)1}wb.save(output_path)print(f数据字典已生成{output_path})generate_excel_dictionary(df,数据字典.xlsx)第三步生成Markdown文档# 影刀Python节点生成Markdown格式方便放wikidefgenerate_markdown_dictionary(dataframe,output_path):生成Markdown格式的数据字典md# 数据库字典\n\nmdf 自动生成时间{datetime.now().strftime(%Y-%m-%d %H:%M)}\nmdf 数据库{DB_CONFIG[database]}\nmdf 总表数{len(dataframe[表名].unique())}| 总字段数{len(dataframe)}\n\n# 按表分组fortable_name,groupindataframe.groupby(表名):table_commentgroup.iloc[0][表注释]row_countgroup.iloc[0][表行数]mdf##{table_name}—{table_comment}\n\nmdf数据行数{row_count:,}\n\nmd| 字段名 | 类型 | 允许空 | 默认值 | 主键 | 索引 | 自增 | 注释 |\nmd|--------|------|--------|--------|------|------|------|------|\nfor_,rowingroup.iterrows():pk_markifrow[主键]是elseidx_markifrow[有索引]是elsemdf|{row[字段名]}|{row[字段类型]}|{row[允许空]}|{row[默认值]}|{pk_mark}|{idx_mark}|{row[自增]}|{row[字段注释]}|\nmd\nwithopen(output_path,w,encodingutf-8)asf:f.write(md)print(fMarkdown字典已生成{output_path})generate_markdown_dictionary(df,数据字典.md)第四步检测变更并更新# 影刀Python节点对比新旧字典defcompare_schemas(old_dict,new_dict):检测数据库结构变化changes[]# 新增表old_tablesset(old_dict[表名].unique())new_tablesset(new_dict[表名].unique())fortinnew_tables-old_tables:changes.append(f➕ 新增表{t})fortinold_tables-new_tables:changes.append(f➖ 删除表{t})# 新增/删除/变更字段fortableinold_tablesnew_tables:old_fieldsset(old_dict[old_dict[表名]table][字段名])new_fieldsset(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_schemaPostgreSQL用information_schema也兼容但细节不同Oracle用ALL_TAB_COLUMNS。如果公司多种数据库并存需要分别适配。坑5视图、存储过程、触发器也需要文档数据字典通常只覆盖表和字段但完整的数据库文档还应该包括视图定义、存储过程说明、触发器逻辑。这些需要单独写SQL提取。总结数据字典自动化的核心价值不是一次性生成文档而是持续同步——数据库在变、文档自动跟。把字典生成做成定时任务每周日更新一次团队永远看到最新的数据库文档。
