影刀RPA SQLite数据库入门轻量级数据存储署名林焱什么情况用做RPA采集的时候数据存哪存Excel吧数据量一大就卡存MySQL吧还得装数据库服务。如果你只是想存几万条数据做做查询统计SQLite就是最佳选择——一个文件就是一个数据库不需要安装服务Python自带sqlite3模块影刀里直接用。典型场景采集的商品数据需要去重、按价格排序后再导出Excel多次运行的采集结果需要累积存储每次只取增量需要给数据打标签、做分类查询Excel筛选太慢做一个小型的客户管理系统增删改查怎么做1. 建库建表在影刀中拖入一个【执行Python代码】组件写入以下代码拼多多店群自动化上架方案importsqlite3importos# 数据库文件路径放在流程同级目录下db_pathrC:\RPAData\products.db# 确保目录存在os.makedirs(os.path.dirname(db_path),exist_okTrue)# 连接数据库不存在则自动创建connsqlite3.connect(db_path)cursorconn.cursor()# 建表cursor.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT NOT NULL,  price REAL, shop TEXT, url TEXT UNIQUE, crawl_time TEXT, tags TEXT ) )# 创建索引加速查询cursor.execute(CREATE INDEX IF NOT EXISTS idx_price ON products(price))cursor.execute(CREATE INDEX IF NOT EXISTS idx_shop ON products(shop))conn.commit()conn.close()print(f数据库已创建{db_path})2. 插入数据采集到商品数据后批量插入importsqlite3fromdatetimeimportdatetime db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 单条插入cursor.execute( INSERT OR IGNORE INTO products (product_name, price, shop, url, crawl_time, tags) VALUES (?, ?, ?, ?, ?, ?) ,(无线蓝牙耳机,129.9,京东自营,https://item.jd.com/123.html,datetime.now().strftime(%Y-%m-%d %H:%M:%S),数码))# 批量插入采集结果列表products[(机械键盘,299.0,天猫旗舰店,https://detail.tmall.com/456.html,datetime.now().strftime(%Y-%m-%d %H:%M:%S),数码),(鼠标垫,19.9,拼多多,https://mobile.yangkeduo.com/789.html,datetime.now().strftime(%Y-%m-%d %H:%M:%S),办公),(USB hubs,49.0,京东,https://item.jd.com/012.html,datetime.now().strftime(%Y-%m-%d %H:%M:%S),数码),]cursor.executemany( INSERT OR IGNORE INTO products (product_name, price, shop, url, crawl_time, tags) VALUES (?, ?, ?, ?, ?, ?) ,products)conn.commit()print(f插入完成当前总记录数{cursor.execute(SELECT COUNT(*) FROM products).fetchone()[0]})conn.close()影刀操作配合用【采集网页数据】组件获取商品列表 → 结果传入【执行Python代码】组件写入SQLite。3. 查询数据importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 基础查询价格降序前10cursor.execute(SELECT product_name, price, shop FROM products ORDER BY price DESC LIMIT 10)top10cursor.fetchall()print(价格TOP10)forrowintop10:print(f{row[0]}- ¥{row[1]}-{row[2]})# 条件查询某店铺商品cursor.execute(SELECT product_name, price FROM products WHERE shop ? AND price 50 ORDER BY price,(京东自营,))resultscursor.fetchall()print(f\n京东自营50元以上商品{len(results)}条)# 聚合统计按店铺分组cursor.execute( SELECT shop, COUNT(*) as count, AVG(price) as avg_price, MAX(price) as max_price FROM products GROUP BY shop ORDER BY count DESC )statscursor.fetchall()print(\n店铺统计)forrowinstats:print(f{row[0]}:{row[1]}条, 均价¥{row[2]:.1f}, 最高¥{row[3]})conn.close()4. 更新和删除importsqlite3 db_pathrC:\RPAData\products.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 更新给数码类商品打上促销标签cursor.execute(UPDATE products SET tags ? WHERE tags ?,(数码-促销,数码))print(f更新了{cursor.rowcount}条记录)# 删除清理30天前的旧数据cursor.execute(DELETE FROM products WHERE crawl_time date(now, -30 day))print(f删除了{cursor.rowcount}条过期记录)conn.commit()conn.close()5. 查询结果导出Excelimportsqlite3importpandasaspd db_pathrC:\RPAData\products.dboutput_pathrC:\RPAData\商品统计报表.xlsxconnsqlite3.connect(db_path)# 直接用pandas读取SQL结果dfpd.read_sql_query(SELECT product_name, price, shop, crawl_time FROM products ORDER BY price DESC,conn)df.to_excel(output_path,indexFalse,sheet_name商品列表)# 按店铺分Sheet导出shopspd.read_sql_query(SELECT DISTINCT shop FROM products,conn)[shop].tolist()withpd.ExcelWriter(output_path,engineopenpyxl,modea)aswriter:forshopinshops:shop_dfpd.read_sql_query(fSELECT * FROM products WHERE shop ?,conn,params(shop,))shop_df.to_excel(writer,indexFalse,sheet_nameshop[:30])conn.close()print(f报表已导出{output_path})有什么坑坑1INSERT OR IGNORE的UNIQUE约束只对指定了UNIQUE的列生效建表时如果url列没有UNIQUE约束INSERT OR IGNORE就不会去重数据会重复插入。必须在建表时明确指定url TEXT UNIQUE。如果建表时忘了加可以事后补cursor.execute(CREATE UNIQUE INDEX IF NOT EXISTS idx_url_unique ON products(url))但如果已有重复数据这一步会报错。得先去重再加索引cursor.execute(DELETE FROM products WHERE id NOT IN (SELECT MIN(id) FROM products GROUP BY url))conn.commit()cursor.execute(CREATE UNIQUE INDEX IF NOT EXISTS idx_url_unique ON products(url))[video(video-y2FLriYe-1783623894691)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]坑2sqlite3连接不自动提交事务很多新手执行完INSERT后发现数据库里没数据。原因sqlite3默认开启事务必须手动conn.commit()。如果中间出异常数据全部回滚。正确写法try:cursor.executemany(INSERT INTO products VALUES (?,?,?,?,?,?),data)conn.commit()exceptExceptionase:conn.rollback()print(f插入失败已回滚{e})finally:conn.close()坑3数据库文件被锁定SQLite是文件锁如果影刀流程A正在写流程B也想写就会报database is locked错误。解决办法设置超时conn sqlite3.connect(db_path, timeout10)— 等待10秒再报错用WAL模式conn.execute(PRAGMA journal_modeWAL)— 允许读写并发高并发场景换MySQLSQLite不适合坑4路径中有中文导致找不到数据库Windows下如果数据库路径含中文如C:\数据\products.db某些情况下会报错。建议路径用英文或者用raw string并确保编码正确db_pathos.path.join(os.environ[USERPROFILE],RPAData,products.db)坑5pandas read_sql_query的params参数必须用元组或列表# 正确写法dfpd.read_sql_query(SELECT * FROM products WHERE shop ?,conn,params(京东,))# 错误写法直接拼接字符串有SQL注入风险dfpd.read_sql_query(fSELECT * FROM products WHERE shop {shop_name},conn)参数化查询不仅是安全要求还能避免特殊字符如单引号导致SQL语法错误。