影刀RPA 数据字典管理：字段含义与数据标准

影刀RPA 数据字典管理字段含义与数据标准署名林焱什么情况用你的RPA流程采了500个字段过了三个月回来维护哪个字段是啥意思prd_price_v2是什么usr_lvl_cd代表什么没人记得了。或者新同事接手项目看数据库里几十张表几百个字段完全不知道该怎么用。数据字典就是解决这个问题的——一份说明书记录每个表、每个字段是什么意思、什么类型、有什么规则。典型场景RPA采集项目字段多、命名不直观需要文档说明多人协作开发需要统一字段定义避免歧义数据交付给业务方时附带字段说明数据库表结构变更后同步更新文档怎么做拼多多店群自动化报活动上架1. 手动创建数据字典importpandasaspdfromdatetimeimportdatetime# 定义数据字典data_dictionary[{表名:products,字段名:product_id,数据类型:INTEGER,是否主键:是,是否可空:否,默认值:自增,字段说明:商品唯一ID自增主键,取值范围:正整数,示例值:10001,关联表:,更新时间:datetime.now().strftime(%Y-%m-%d)},{表名:products,字段名:product_name,数据类型:TEXT,是否主键:否,是否可空:否,默认值:,字段说明:商品名称采集自商品标题,取值范围:文本最长200字符,示例值:无线蓝牙耳机5.0,关联表:,更新时间:datetime.now().strftime(%Y-%m-%d)},{表名:products,字段名:price,数据类型:REAL,是否主键:否,是否可空:是,默认值:NULL,字段说明:商品价格单位元采集时的标价,取值范围:0.01 - 999999.99,示例值:129.90,关联表:,更新时间:datetime.now().strftime(%Y-%m-%d)},{表名:products,字段名:shop_id,数据类型:INTEGER,是否主键:否,是否可空:是,默认值:NULL,字段说明:店铺ID关联shops表,取值范围:正整数,示例值:205,关联表:shops(id),更新时间:datetime.now().strftime(%Y-%m-%d)},]# 导出为Exceldfpd.DataFrame(data_dictionary)output_pathrC:\RPAData\数据字典.xlsxdf.to_excel(output_path,indexFalse,sheet_name数据字典)print(f数据字典已生成{output_path})2. 从数据库自动提取表结构importsqlite3importpandasaspdfromdatetimeimportdatetime db_pathrC:\RPAData\business.dboutput_pathrC:\RPAData\数据字典_自动生成.xlsxconnsqlite3.connect(db_path)cursorconn.cursor()# 获取所有表名cursor.execute(SELECT name FROM sqlite_master WHERE typetable AND name NOT LIKE sqlite_%)tables[row[0]forrowincursor.fetchall()]all_dict[]fortableintables:# 获取表结构cursor.execute(fPRAGMA table_info({table}))columnscursor.fetchall()# 获取表的创建语句含注释信息cursor.execute(fSELECT sql FROM sqlite_master WHERE name {table})create_sqlcursor.fetchone()[0]forcolincolumns:col_id,col_name,col_type,not_null,default_val,is_pkcol all_dict.append({表名:table,字段名:col_name,数据类型:col_type,是否主键:是ifis_pkelse否,是否可空:否ifnot_nullelse是,默认值:str(default_val)ifdefault_valisnotNoneelse,字段说明:,# SQLite不存储字段注释需手动补充示例值:,# 可从数据中抽样更新时间:datetime.now().strftime(%Y-%m-%d)})conn.close()# 导出dfpd.DataFrame(all_dict)# 按表分Sheet导出withpd.ExcelWriter(output_path,engineopenpyxl)aswriter:# 总览Sheetdf.to_excel(writer,indexFalse,sheet_name全部字段)# 每个表一个Sheetfortableintables:table_dfdf[df[表名]table]table_df.to_excel(writer,indexFalse,sheet_nametable[:31])# Sheet名最长31字符print(f数据字典已生成{output_path}{len(tables)}张表{len(all_dict)}个字段)3. 从数据中抽样示例值importsqlite3importpandasaspd db_pathrC:\RPAData\business.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 获取表结构cursor.execute(SELECT name FROM sqlite_master WHERE typetable)tables[row[0]forrowincursor.fetchall()]dict_entries[]fortableintables:cursor.execute(fPRAGMA table_info({table}))columnscursor.fetchall()forcolincolumns:col_namecol[1]col_typecol[2]# 抽样取3个非空值作为示例try:cursor.execute(fSELECT DISTINCT{col_name}FROM{table}WHERE{col_name}IS NOT NULL LIMIT 3)samples[str(row[0])forrowincursor.fetchall()]sample_str | .join(samples)except:sample_str# 统计空值比例try:cursor.execute(fSELECT COUNT(*) FROM{table})totalcursor.fetchone()[0]cursor.execute(fSELECT COUNT(*) FROM{table}WHERE{col_name}IS NULL)null_countcursor.fetchone()[0]null_ratef{null_count}/{total}({null_count/total*100:.1f}%)iftotal0elseN/Aexcept:null_rateN/Adict_entries.append({表名:table,字段名:col_name,类型:col_type,示例值:sample_str,空值率:null_rate})conn.close()dfpd.DataFrame(dict_entries)df.to_excel(rC:\RPAData\数据字典_含示例.xlsx,indexFalse,sheet_name数据字典)print(f含示例值的数据字典已生成)4. 字段命名规范校验importreimportpandasaspd# 读取数据字典dfpd.read_excel(rC:\RPAData\数据字典.xlsx)# 命名规范规则naming_rules{前缀规范:{pattern:r^(is_|has_|num_|dt_|txt_|amt_),description:布尔用is_/has_数量用num_日期用dt_文本用txt_金额用amt_},禁止缩写:[prd,usr,lvl,cd,nm],# 不允许的缩写长度限制:30,# 字段名最长30字符}violations[]for_,rowindf.iterrows():field_namestr(row[字段名])issues[]# 检查长度iflen(field_name)naming_rules[长度限制]:issues.append(f字段名过长({len(field_name)}{naming_rules[长度限制]}))# 检查缩写forabbrinnaming_rules[禁止缩写]:ifabbrinfield_name.lower():issues.append(f包含禁止缩写{abbr})# 检查命名规范可选不是所有字段都需要前缀# if not re.match(naming_rules[前缀规范][pattern], field_name):# issues.append(不符合前缀规范)ifissues:violations.append({表名:row[表名],字段名:field_name,问题:.join(issues)})ifviolations:vio_dfpd.DataFrame(violations)print(f发现{len(violations)}个命名不规范的字段)print(vio_df.to_string(indexFalse))vio_df.to_excel(rC:\RPAData\命名规范检查.xlsx,indexFalse)else:print(所有字段命名符合规范)有什么坑坑1数据字典和实际表结构不同步改了数据库表结构加字段、改类型但忘了更新数据字典文档和实际不一致。# 解决方案自动比对数据字典和实际表结构importsqlite3importpandasaspd connsqlite3.connect(rC:\RPAData\business.db)cursorconn.cursor()# 读取现有数据字典dict_dfpd.read_excel(rC:\RPAData\数据字典.xlsx)# 获取实际表结构actual_fieldsset()fortableindict_df[表名].unique():cursor.execute(fPRAGMA table_info({table}))forcolincursor.fetchall():actual_fields.add((table,col[1]))# 数据字典中的字段dict_fieldsset(zip(dict_df[表名],dict_df[字段名]))# 差异missing_in_dictactual_fields-dict_fields# 表里有但字典里没有missing_in_dbdict_fields-actual_fields# 字典里有但表里没有ifmissing_in_dict:print(f字典中缺失的字段表里有但字典没记录)fortable,fieldinmissing_in_dict:print(f{table}.{field})ifmissing_in_db:print(f字典中多余的字段字典有但表里没有可能已删除)fortable,fieldinmissing_in_db:print(f{table}.{field})坑2字段说明写得太笼统“商品价格”——是含税还是不含税是标价还是到手价单位是元还是分好的字段说明应该包括含义单位数据来源注意事项。TEMU店群矩阵自动化运营核价报活动# 差的字段说明 price: 商品价格 # 好的字段说明 price: 商品标价不含运费单位元采集自商品详情页价格区域。注意促销时此价格可能不等于到手价到手价需减去优惠券。坑3枚举值没有记录status字段有pending、processing、shipped、delivered、cancelled五个值但数据字典里只写了订单状态。新来的人根本不知道每个值代表什么。# 数据字典中应该记录枚举值{字段名:status,字段说明:订单状态,枚举值:pending待处理; processing处理中; shipped已发货; delivered已签收; cancelled已取消}# 或单独维护枚举值表enum_dfpd.DataFrame([{表名:orders,字段名:status,枚举值:pending,含义:待处理,备注:新订单默认状态},{表名:orders,字段名:status,枚举值:processing,含义:处理中,备注:已分配仓库},{表名:orders,字段名:status,枚举值:shipped,含义:已发货,备注:已生成物流单},])坑4敏感字段没有标注手机号、身份证号、银行卡号等敏感字段在数据字典中必须标注提醒使用时注意脱敏。# 敏感字段标注sensitive_fields[phone,id_card,bank_card,email,address]forentryindict_entries:ifentry[字段名]insensitive_fields:entry[敏感级别]高entry[脱敏规则]phone: 显示前3后4 (138****5678); id_card: 显示前6后4else:entry[敏感级别]低entry[脱敏规则]坑5数据字典文件版本混乱数据字典.xlsx、数据字典_v2.xlsx、数据字典_最新.xlsx、数据字典_最终版.xlsx——到底哪个是最新的解决方案用数据库管理数据字典本身不要用Excel文件# 把数据字典也存在数据库里cursor.execute( CREATE TABLE IF NOT EXISTS _data_dictionary ( id INTEGER PRIMARY KEY, table_name TEXT, field_name TEXT, data_type TEXT, description TEXT, is_sensitive INTEGER DEFAULT 0,   update_time TEXT, updated_by TEXT ) )# 查字典直接查数据库cursor.execute(SELECT field_name, description FROM _data_dictionary WHERE table_name ?,(products,))