影刀RPA 数据库数据导出：多格式导出与报表生成

影刀RPA 数据库数据导出多格式导出与报表生成署名林焱什么情况用RPA采集了三天数据存到SQLite/MySQL里老板说给我个Excel看看。你不能让他去看数据库吧数据导出是RPA流程的最后一公里——处理完的数据得变成人能看懂的格式。典型场景每日采集结果导出Excel日报自动发邮件给负责人数据库数据按条件筛选后导出CSV供其他系统导入多维度统计结果分Sheet导出生成月度报表按客户/部门/日期分文件导出每个文件一份怎么做1. 基础导出SQL查询结果导出Excelimportsqlite3importpandasaspd db_pathrC:\RPAData\products.dboutput_pathrC:\RPAData\导出\商品数据.xlsxconnsqlite3.connect(db_path)# 查询并直接导出[video(video-cl6Nej5o-1784264844890)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)]dfpd.read_sql_query(SELECT product_name, price, shop, crawl_time FROM products ORDER BY price DESC,conn)df.columns[商品名称,价格,店铺,采集时间]# 重命名列df.to_excel(output_path,indexFalse,sheet_name商品列表)conn.close()print(f导出完成{output_path}共{len(df)}条)2. 多Sheet导出importsqlite3importpandasaspd db_pathrC:\RPAData\products.dboutput_pathrC:\RPAData\导出\商品统计报表.xlsxconnsqlite3.connect(db_path)withpd.ExcelWriter(output_path,engineopenpyxl)aswriter:# Sheet1: 全部商品df_allpd.read_sql_query(SELECT * FROM products ORDER BY price DESC,conn)df_all.to_excel(writer,indexFalse,sheet_name全部商品)# Sheet2: 按店铺统计df_shoppd.read_sql_query( SELECT shop as 店铺, COUNT(*) as 商品数, ROUND(AVG(price), 2) as 均价, MAX(price) as 最高价, MIN(price) as 最低价 FROM products GROUP BY shop ORDER BY 商品数 DESC ,conn)df_shop.to_excel(writer,indexFalse,sheet_name店铺统计)# Sheet3: 价格区间分布df_rangepd.read_sql_query( SELECT CASE WHEN price 50 THEN 0-50元 WHEN price 100 THEN 50-100元 WHEN price 200 THEN 100-200元 WHEN price 500 THEN 200-500元 ELSE 500元以上 END as 价格区间, COUNT(*) as 商品数, ROUND(AVG(price), 2) as 区间均价 FROM products GROUP BY 价格区间 ORDER BY 区间均价 ,conn)df_range.to_excel(writer,indexFalse,sheet_name价格分布)# Sheet4: TOP20热销df_toppd.read_sql_query(SELECT product_name, price, shop FROM products ORDER BY price DESC LIMIT 20,conn)df_top.to_excel(writer,indexFalse,sheet_nameTOP20)conn.close()print(f多Sheet报表导出完成{output_path})3. 按条件分文件导出importsqlite3importpandasaspdimportos db_pathrC:\RPAData\products.dboutput_dirrC:\RPAData\导出\按店铺os.makedirs(output_dir,exist_okTrue)connsqlite3.connect(db_path)# 获取所有店铺列表cursorconn.cursor()cursor.execute(SELECT DISTINCT shop FROM products)shops[row[0]forrowincursor.fetchall()ifrow[0]]forshopinshops:dfpd.read_sql_query(SELECT * FROM products WHERE shop ? ORDER BY price DESC,conn,params(shop,))# 文件名清理去掉非法字符safe_nameshop.replace(/,_).replace(\\,_).replace(:,_).replace(*,_).replace(?,_).replace(,_).replace(,_).replace(,_).replace(|,_)file_pathos.path.join(output_dir,f{safe_name}_商品列表.xlsx)df.to_excel(file_path,indexFalse,sheet_name商品列表)print(f导出{file_path}{len(df)}条)conn.close()print(f\n按店铺分文件导出完成共{len(shops)}个文件)影刀操作配合导出后用【发送邮件】组件把Excel作为附件发送给对应店铺负责人。4. 导出CSV大数量量importsqlite3importcsv db_pathrC:\RPAData\products.dboutput_pathrC:\RPAData\导出\商品数据.csvconnsqlite3.connect(db_path)cursorconn.cursor()# 分批读取写入CSV避免内存溢出BATCH_SIZE5000cursor.execute(SELECT product_name, price, shop, crawl_time FROM products ORDER BY id)withopen(output_path,w,newline,encodingutf-8-sig)asf:# utf-8-sig解决Excel中文乱码writercsv.writer(f)writer.writerow([商品名称,价格,店铺,采集时间])# 表头total0whileTrue:rowscursor.fetchmany(BATCH_SIZE)ifnotrows:breakwriter.writerows(rows)totallen(rows)print(f已导出{total})conn.close()print(fCSV导出完成{output_path}共{total}条)5. 导出JSON对接APIimportsqlite3importjsonfromdatetimeimportdatetime db_pathrC:\RPAData\products.dboutput_pathrC:\RPAData\导出\商品数据.jsonconnsqlite3.connect(db_path)conn.row_factorysqlite3.Row# 返回可按列名访问的行对象cursorconn.cursor()cursor.execute(SELECT * FROM products ORDER BY price DESC LIMIT 100)rowscursor.fetchall()# 转为JSONdata[]forrowinrows:itemdict(row)# 处理特殊类型ifcrawl_timeinitemanditem[crawl_time]:item[crawl_time]str(item[crawl_time])data.append(item)withopen(output_path,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)conn.close()print(fJSON导出完成{output_path}共{len(data)}条)有什么坑坑1Excel单Sheet最多1048576行如果数据超过100万行Excel放不下会报错或截断。解决分文件或分Sheet。importsqlite3importpandasaspdimportos db_pathrC:\RPAData\big_data.dboutput_dirrC:\RPAData\导出connsqlite3.connect(db_path)MAX_ROWS1000000# 每个文件最多100万行cursorconn.cursor()cursor.execute(SELECT COUNT(*) FROM big_table)totalcursor.fetchone()[0]file_count(totalMAX_ROWS-1)//MAX_ROWSforiinrange(file_count):offseti*MAX_ROWS dfpd.read_sql_query(fSELECT * FROM big_table LIMIT{MAX_ROWS}OFFSET{offset},conn)file_pathos.path.join(output_dir,f数据_part{i1}.xlsx)df.to_excel(file_path,indexFalse)print(f导出{file_path}{len(df)}条)conn.close()坑2CSV中文乱码用Excel打开CSV文件中文全是乱码。原因Excel默认用ANSI编码读取CSV而Python默认用UTF-8写入。# 错误用utf-8withopen(path,w,encodingutf-8)asf:# Excel打开乱码writercsv.writer(f)# 正确用utf-8-sig带BOM头withopen(path,w,encodingutf-8-sig)asf:# Excel打开正常writercsv.writer(f)坑3pandas to_excel的日期格式数据库里的日期字段导出后变成2024-01-15 10:30:00这种字符串格式Excel不认为是日期。解决TEMU店群如何管理运营importsqlite3importpandasaspd connsqlite3.connect(db_path)dfpd.read_sql_query(SELECT * FROM orders,conn)# 转换日期列为datetime类型df[create_time]pd.to_datetime(df[create_time])# 导出到Excel时格式化withpd.ExcelWriter(output_path,engineopenpyxl)aswriter:df.to_excel(writer,indexFalse,sheet_name订单)# 获取worksheet对象设置日期格式worksheetwriter.sheets[订单]# 找到日期列的列号date_coldf.columns.get_loc(create_time)1# 1因为从1开始forrowinrange(2,len(df)2):# 从第2行开始第1行是表头cellworksheet.cell(rowrow,columndate_col)cell.number_formatYYYY-MM-DD HH:MM:SSconn.close()坑4导出文件被占用导致写入失败Excel正打开着导出文件RPA再次导出会报Permission denied。解决importosimporttimedefsafe_export(df,file_path,max_retries3):安全导出文件被占用时自动重试forattemptinrange(max_retries):try:df.to_excel(file_path,indexFalse)returnTrueexceptPermissionError:ifattemptmax_retries-1:print(f文件被占用{5}秒后重试...{attempt1}/{max_retries})time.sleep(5)else:# 重命名文件避免冲突base,extos.path.splitext(file_path)file_pathf{base}_{int(time.time())}{ext}df.to_excel(file_path,indexFalse)print(f原文件被占用已另存为{file_path})returnTruereturnFalse坑5大数据量导出内存溢出pandas的read_sql_query会把全部结果加载到内存。50万行数据可能占用2GB内存导致OOM。# 错误一次性加载全部dfpd.read_sql_query(SELECT * FROM huge_table,conn)# 内存爆炸# 正确用fetchmany分批读取cursor.execute(SELECT * FROM huge_table)withopen(output_path,w,newline,encodingutf-8-sig)asf:writercsv.writer(f)writer.writerow([desc[0]fordescincursor.description])# 表头whileTrue:rowscursor.fetchmany(5000)ifnotrows:breakwriter.writerows(rows)或者用SQLAlchemy的chunksize参数fromsqlalchemyimportcreate_engineimportpandasaspd enginecreate_engine(fsqlite:///{db_path})# 每次只读10000行forchunkinpd.read_sql(SELECT * FROM huge_table,engine,chunksize10000):# 处理每批数据chunk.to_csv(output_path,modea,headernotos.path.exists(output_path),indexFalse,encodingutf-8-sig)