人体健康状态检测Python脚本
2026/7/17 19:29:27
影刀RPA 数据库数据导入：Excel与CSV批量导入
署名：林焱
什么情况用
老板给你一个Excel文件，5000条客户信息，说"把这些导入系统"。你总不能手动一条一条录吧？或者ERP每天导出一个CSV，你需要自动导入到分析库里做统计。
数据导入是数据处理的入口环节——格式不对、数据不干净、主键冲突，每个问题都能让导入卡住。掌握正确的导入姿势，才能让RPA流程跑得稳。
典型场景：
Excel客户名单批量导入CRM系统
每日CSV销售数据导入分析数据库
多个Excel文件合并后导入数据库
旧系统数据迁移到新系统
怎么做
1. Excel数据导入SQLite
拼多多店群自动化报活动上架！
importsqlite3importpandasaspdfromdatetimeimportdatetime# 读取Excelexcel_path=r"C:\RPAData\客户名单.xlsx"df=pd.read_excel(excel_path,dtype={'phone':str,'id_card':str})# 文本列防止被转成数字# 数据清洗df=df.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_path=r"C:\RPAData\business.db"conn=sqlite3.connect(db_path)cursor=conn.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数据导入MySQL
importpymysqlimportcsvimportos db_config={'host':'192.168.1.100','port':3306,'user':'rpa_user','password':'RPA@2024#secure','database':'sales_db','charset':'utf8mb4'}csv_path=r"C:\RPAData\每日销售数据.csv"# 读取CSVwithopen(csv_path,'r',encoding='utf-8-sig')asf:reader=csv.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并批量导入conn=pymysql.connect(**db_config)cursor=conn.cursor()BATCH_SIZE=1000foriinrange(0,len(records),BATCH_SIZE):batch=records[i:i+BATCH_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(i+BATCH_SIZE,len(records))}/{len(records)}")print(f"CSV导入完成：{len(records)}条")conn.close()3. 多Excel文件合并导入
importsqlite3importpandasaspdimportosimportglob db_path=r"C:\RPAData\business.db"excel_dir=r"C:\RPAData\待导入"conn=sqlite3.connect(db_path)cursor=conn.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_files=glob.glob(os.path.join(excel_dir,"*.xlsx"))print(f"发现{len(excel_files)}个Excel文件")total_imported=0forfile_pathinexcel_files:try:df=pd.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_imported+=len(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_path=r"C:\RPAData\business.db"excel_path=r"C:\RPAData\客户名单.xlsx"# 读取数据df=pd.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('姓名为空')# 校验手机号phone=str(row.get('phone',''))ifphoneandphone!='nan':phone=phone.replace('.0','')# 处理Excel数字格式iflen(phone)!=11ornotphone.startswith('1'):row_errors.append(f'手机号格式错误：{phone}')# 校验邮箱email=str(row.get('email',''))ifemailandemail!='nan'and'@'notinemail:row_errors.append(f'邮箱格式错误：{email}')ifrow_errors:errors.append({'行号':idx+2,'姓名':row.get('customer_name',''),'错误':'；'.join(row_errors)})else:valid_records.append((row['customer_name'],phoneifphone!='nan'else'',emailifemail!='nan'else'',row.get('address',''),datetime.now().strftime('%Y-%m-%d %H:%M:%S')))# 输出错误报告iferrors:error_df=pd.DataFrame(errors)error_path=excel_path.replace('.xlsx','_错误报告.xlsx')error_df.to_excel(error_path,index=False)print(f"发现{len(errors)}条错误数据，错误报告：{error_path}")# 只导入有效数据conn=sqlite3.connect(db_path)cursor=conn.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':'RPA@2024#secure','database':'sales_db','charset':'utf8mb4'}csv_path=r"C:\RPAData\sales.csv"# 先把CSV转为MySQL兼容格式clean_csv_path=csv_path.replace('.csv','_clean.csv')withopen(csv_path,'r',encoding='utf-8-sig')asinfile:reader=csv.reader(infile)header=next(reader)withopen(clean_csv_path,'w',newline='',encoding='utf-8')asoutfile:writer=csv.writer(outfile,terminator='\n')writer.writerow(header)writer.writerows(reader)# 用LOAD DATA导入（比INSERT快10-100倍）conn=pymysql.connect(**db_config)cursor=conn.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(f"LOAD DATA导入完成：{cursor.rowcount}条")# 清理临时文件os.remove(clean_csv_path)conn.close()有什么坑
坑1：Excel数字列被自动转为浮点数
手机号13812345678读进来变成1.3812345678E10，身份证号变成科学计数法。
# 错误：直接读取df=pd.read_excel(path)# phone列变成float# 正确：指定dtype为strdf=pd.read_excel(path,dtype={'phone':str,'id_card':str,'zip_code':str})# 或读取后转换df['phone']=df['phone'].astype(str).str.replace(r'\.0$','',regex=True).str.strip()坑2：Excel日期变成数字
Excel里的日期2024-01-15读取后可能变成45306（Excel日期序列号）。
importpandasaspd# 方案1：用parse_dates参数df=pd.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(days=serial)returnserial df['order_date']=df['order_date'].apply(excel_date_to_datetime)坑3：主键冲突导致导入中断
Excel里有重复数据，或者数据库已有部分数据，INSERT到冲突行就报错中断。
# 方案1：INSERT OR REPLACE（覆盖旧数据）cursor.executemany('INSERT OR REPLACE INTO customers VALUES (?,?,?,?,?)',records)# 方案2：INSERT OR IGNORE（跳过冲突数据）cursor.executemany('INSERT OR IGNORE INTO customers VALUES (?,?,?,?,?)',records)# 方案3：先查重再导入existing=set()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)}条新数据")坑4：MySQL的local_infile未开启
使用LOAD DATA LOCAL INFILE时报错The used command is not allowed with this MySQL version。
# 需要在连接时启用local_infileconn=pymysql.connect(**db_config,local_infile=True)# 或者执行SQL开启cursor.execute('SET GLOBAL local_infile = 1')服务端也需要在my.cnf中配置local_infile=1，然后重启MySQL。
坑5：编码问题导致中文乱码
CSV文件用GBK编码（Windows默认），Python用UTF-8读取，中文全乱码。
importchardetdefdetect_and_read_csv(file_path):"""自动检测编码并读取CSV"""withopen(file_path,'rb')asf:raw=f.read(10000)encoding=chardet.detect(raw)['encoding']# 常见编码回退ifencodingisNone:encoding='utf-8'elifencoding=='GB2312':encoding='gbk'# GBK兼容GB2312try:returnpd.read_csv(file_path,encoding=encoding)exceptUnicodeDecodeError:# 尝试其他编码forencin['utf-8','gbk','gb18030','latin1']:try:returnpd.read_csv(file_path,encoding=enc)except:continueraiseValueError(f"无法识别文件编码：{file_path}")df=detect_and_read_csv(csv_path)