影刀RPA 数据库数据导入Excel与CSV批量导入署名林焱什么情况用老板给你一个Excel文件5000条客户信息说把这些导入系统。你总不能手动一条一条录吧或者ERP每天导出一个CSV你需要自动导入到分析库里做统计。数据导入是数据处理的入口环节——格式不对、数据不干净、主键冲突每个问题都能让导入卡住。掌握正确的导入姿势才能让RPA流程跑得稳。典型场景Excel客户名单批量导入CRM系统每日CSV销售数据导入分析数据库多个Excel文件合并后导入数据库旧系统数据迁移到新系统怎么做1. Excel数据导入SQLite拼多多店群自动化报活动上架importsqlite3importpandasaspdfromdatetimeimportdatetime# 读取Excelexcel_pathrC:\RPAData\客户名单.xlsxdfpd.read_excel(excel_path,dtype{phone:str,id_card:str})# 文本列防止被转成数字# 数据清洗dfdf.dropna(subset[customer_name])# 删除姓名为空的行df[phone]df[phone].fillna().astype(str).str.replace(.0,)# 处理电话号码df[create_time]datetime.now().strftime(%Y-%m-%d %H:%M:%S)# 连接数据库db_pathrC:\RPAData\business.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 确保表存在cursor.execute( CREATE TABLE IF NOT EXISTS customers ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_name TEXT NOT NULL, phone TEXT, email TEXT, address TEXT, create_time TEXT ) )# 批量插入records[tuple(x)forxindf[[customer_name,phone,email,address,create_time]].values]cursor.executemany(INSERT INTO customers (customer_name, phone, email, address, create_time) VALUES (?,?,?,?,?),records)conn.commit()print(f导入完成{len(records)}条客户数据)conn.close()2. CSV数据导入MySQLimportpymysqlimportcsvimportos db_config{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:sales_db,charset:utf8mb4}csv_pathrC:\RPAData\每日销售数据.csv# 读取CSVwithopen(csv_path,r,encodingutf-8-sig)asf:readercsv.DictReader(f)records[(row[order_id],row[product_name],float(row[quantity]),float(row[unit_price]),float(row[quantity])*float(row[unit_price]),row[sale_date])forrowinreader]# 连接MySQL并批量导入connpymysql.connect(**db_config)cursorconn.cursor()BATCH_SIZE1000foriinrange(0,len(records),BATCH_SIZE):batchrecords[i:iBATCH_SIZE]cursor.executemany( INSERT INTO sales (order_id, product_name, quantity, unit_price, total_amount, sale_date) VALUES (%s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE quantity VALUES(quantity), unit_price VALUES(unit_price), total_amount VALUES(total_amount) ,batch)conn.commit()print(f导入进度{min(iBATCH_SIZE,len(records))}/{len(records)})print(fCSV导入完成{len(records)}条)conn.close()3. 多Excel文件合并导入importsqlite3importpandasaspdimportosimportglob db_pathrC:\RPAData\business.dbexcel_dirrC:\RPAData\待导入connsqlite3.connect(db_path)cursorconn.cursor()cursor.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT, price REAL, shop TEXT, source_file TEXT, import_time TEXT ) )# 遍历所有Excel文件excel_filesglob.glob(os.path.join(excel_dir,*.xlsx))print(f发现{len(excel_files)}个Excel文件)total_imported0forfile_pathinexcel_files:try:dfpd.read_excel(file_path)df[source_file]os.path.basename(file_path)df[import_time]pd.Timestamp.now().strftime(%Y-%m-%d %H:%M:%S)records[tuple(x)forxindf[[product_name,price,shop,source_file,import_time]].values]cursor.executemany(INSERT INTO products (product_name, price, shop, source_file, import_time) VALUES (?,?,?,?,?),records)conn.commit()total_importedlen(records)print(f{os.path.basename(file_path)}:{len(records)}条)exceptExceptionase:print(f{os.path.basename(file_path)}导入失败{e})conn.close()print(f\n全部导入完成共{total_imported}条)影刀操作配合用【获取文件夹下所有文件】组件获取文件列表 → 【循环-列表】遍历每个文件 → 【执行Python代码】读取并导入。4. 数据校验后导入importsqlite3importpandasaspdfromdatetimeimportdatetime db_pathrC:\RPAData\business.dbexcel_pathrC:\RPAData\客户名单.xlsx# 读取数据dfpd.read_excel(excel_path,dtype{phone:str})# 数据校验errors[]valid_records[]foridx,rowindf.iterrows():row_errors[]# 校验必填字段ifnotrow.get(customer_name)orpd.isna(row[customer_name]):row_errors.append(姓名为空)# 校验手机号phonestr(row.get(phone,))ifphoneandphone!nan:phonephone.replace(.0,)# 处理Excel数字格式iflen(phone)!11ornotphone.startswith(1):row_errors.append(f手机号格式错误{phone})# 校验邮箱emailstr(row.get(email,))ifemailandemail!nanandnotinemail:row_errors.append(f邮箱格式错误{email})ifrow_errors:errors.append({行号:idx2,姓名:row.get(customer_name,),错误:.join(row_errors)})else:valid_records.append((row[customer_name],phoneifphone!nanelse,emailifemail!nanelse,row.get(address,),datetime.now().strftime(%Y-%m-%d %H:%M:%S)))# 输出错误报告iferrors:error_dfpd.DataFrame(errors)error_pathexcel_path.replace(.xlsx,_错误报告.xlsx)error_df.to_excel(error_path,indexFalse)print(f发现{len(errors)}条错误数据错误报告{error_path})# 只导入有效数据connsqlite3.connect(db_path)cursorconn.cursor()ifvalid_records:cursor.executemany(INSERT INTO customers (customer_name, phone, email, address, create_time) VALUES (?,?,?,?,?),valid_records)conn.commit()print(f成功导入{len(valid_records)}条有效数据)conn.close()5. MySQL LOAD DATA方式最快importpymysqlimportcsvimportos db_config{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:sales_db,charset:utf8mb4}csv_pathrC:\RPAData\sales.csv# 先把CSV转为MySQL兼容格式clean_csv_pathcsv_path.replace(.csv,_clean.csv)withopen(csv_path,r,encodingutf-8-sig)asinfile:readercsv.reader(infile)headernext(reader)withopen(clean_csv_path,w,newline,encodingutf-8)asoutfile:writercsv.writer(outfile,terminator\n)writer.writerow(header)writer.writerows(reader)# 用LOAD DATA导入比INSERT快10-100倍connpymysql.connect(**db_config)cursorconn.cursor()cursor.execute( LOAD DATA LOCAL INFILE %s INTO TABLE sales FIELDS TERMINATED BY , ENCLOSED BY LINES TERMINATED BY \\n IGNORE 1 ROWS (order_id, product_name, quantity, unit_price, sale_date) SET total_amount quantity * unit_price, import_time NOW() ,(clean_csv_path.replace(\\,/)))# MySQL需要正斜杠路径conn.commit()print(fLOAD DATA导入完成{cursor.rowcount}条)# 清理临时文件os.remove(clean_csv_path)conn.close()有什么坑坑1Excel数字列被自动转为浮点数手机号13812345678读进来变成1.3812345678E10身份证号变成科学计数法。# 错误直接读取dfpd.read_excel(path)# phone列变成float# 正确指定dtype为strdfpd.read_excel(path,dtype{phone:str,id_card:str,zip_code:str})# 或读取后转换df[phone]df[phone].astype(str).str.replace(r\.0$,,regexTrue).str.strip()坑2Excel日期变成数字Excel里的日期2024-01-15读取后可能变成45306Excel日期序列号。importpandasaspd# 方案1用parse_dates参数dfpd.read_excel(path,parse_dates[order_date,create_time])# 方案2手动转换序列号defexcel_date_to_datetime(serial):[video(video-9WW4CL4R-1784264804778)(type-csdn)(url-https://live.csdn.net/v/embed/526817)(image-https://v-blog.csdnimg.cn/asset/1d3c3709da119dd8c13ab01e9b282520/cover/Cover0.jpg)(title-TEMU店群矩阵自动化运营核价报活动)]ifisinstance(serial,(int,float)):returnpd.Timestamp(1899-12-30)pd.Timedelta(daysserial)returnserial df[order_date]df[order_date].apply(excel_date_to_datetime)坑3主键冲突导致导入中断Excel里有重复数据或者数据库已有部分数据INSERT到冲突行就报错中断。# 方案1INSERT OR REPLACE覆盖旧数据cursor.executemany(INSERT OR REPLACE INTO customers VALUES (?,?,?,?,?),records)# 方案2INSERT OR IGNORE跳过冲突数据cursor.executemany(INSERT OR IGNORE INTO customers VALUES (?,?,?,?,?),records)# 方案3先查重再导入existingset()cursor.execute(SELECT phone FROM customers)forrowincursor.fetchall():existing.add(row[0])new_records[rforrinrecordsifr[1]notinexisting]# r[1]是phonecursor.executemany(INSERT INTO customers VALUES (?,?,?,?,?),new_records)print(f跳过{len(records)-len(new_records)}条重复导入{len(new_records)}条新数据)坑4MySQL的local_infile未开启使用LOAD DATA LOCAL INFILE时报错The used command is not allowed with this MySQL version。# 需要在连接时启用local_infileconnpymysql.connect(**db_config,local_infileTrue)# 或者执行SQL开启cursor.execute(SET GLOBAL local_infile 1)服务端也需要在my.cnf中配置local_infile1然后重启MySQL。坑5编码问题导致中文乱码CSV文件用GBK编码Windows默认Python用UTF-8读取中文全乱码。importchardetdefdetect_and_read_csv(file_path):自动检测编码并读取CSVwithopen(file_path,rb)asf:rawf.read(10000)encodingchardet.detect(raw)[encoding]# 常见编码回退ifencodingisNone:encodingutf-8elifencodingGB2312:encodinggbk# GBK兼容GB2312try:returnpd.read_csv(file_path,encodingencoding)exceptUnicodeDecodeError:# 尝试其他编码forencin[utf-8,gbk,gb18030,latin1]:try:returnpd.read_csv(file_path,encodingenc)except:continueraiseValueError(f无法识别文件编码{file_path})dfdetect_and_read_csv(csv_path)