影刀RPA进阶教程：SQLite/MySQL数据库读写与批量数据存储

系列前面的文章讲数据存储，基本都是在说Excel。Excel适合做报表和看数据，但作为数据仓库有天然短板——数据量大了会卡、多人同时写入会冲突、查询不便。

当数据量超过1万行，或者需要做复杂查询时，就该考虑用数据库了。

今天讲影刀如何操作数据库：SQLite（轻量级，单机）和MySQL（服务器级，多用户）。提供完整的读写模板，以及批量写入的优化方案。

---

先说选型：SQLite vs MySQL

对比维度	SQLite	MySQL
部署难度	无（Python内置）	需安装数据库服务

| 适用数据量 | <10万行 | 百万级以上 | | 并发写入 | 单线程 | 支持多用户 | | 使用场景 | 单机流程、本地缓存 | 团队共享、生产环境 | | 影刀支持 | Python sqlite3 | Python pymysql |

我的建议：

• 个人/小团队、数据量<5万行 → SQLite
• 多用户共享、数据量大 → MySQL

---

第一部分：SQLite数据库操作

SQLite是Python自带的，不需要额外安装。影刀的“执行Python代码”可以直接用。

1. 创建数据库与表

# ===== Python代码：创建SQLite数据库和表 =====

import sqlite3
import json

# 连接数据库（如果不存在会自动创建）
conn = sqlite3.connect('C:/RPA_Data/products.db')
cursor = conn.cursor()

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/3b150e68f9834dafb6a4aeb1c948df0a~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783493750&x-signature=OSj4MEvo46%2B9FlNAoJBeuEYc8QI%3D)

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/dfd37ca88ede49338b941719f52e58cf~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783493750&x-signature=y4qyk4XflLKEM9oQWTXSKK5Vouo%3D)
# 创建表（如果不存在）
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    platform TEXT NOT NULL,
    product_name TEXT NOT NULL,
    price REAL,
    sales INTEGER,
    collect_time TEXT
)
''')

conn.commit()
conn.close()

print("数据库和表创建成功")

2. 插入单条数据

# ===== 插入单条数据 =====
import sqlite3

def insert_product(platform, name, price, sales, collect_time):

![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/17d262885c1347ed9d608cbbaf120ffc~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783493750&x-signature=H8A%2BkIekl2a0THy90%2BlART74i%2Bk%3D)
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    
    cursor.execute('''
    INSERT INTO products (platform, product_name, price, sales, collect_time)
    VALUES (?, ?, ?, ?, ?)
    ''', (platform, name, price, sales, collect_time))
    
    conn.commit()
    conn.close()
    return cursor.lastrowid  # 返回插入的ID

# 影刀调用时，传入参数
product_id = insert_product('淘宝', 'iPhone 15', 5999.00, 1000, '2026-07-01 10:00:00')
print(f'插入成功，ID: {product_id}')

3. 批量插入（性能优化）

逐条插入很慢，批量插入速度可提升10倍以上。

# ===== 批量插入 =====
import sqlite3


![picture.image](https://p3-volc-community-sign.byteimg.com/tos-cn-i-tlddhu82om/600f65fff39b4418bd927809819ec456~tplv-tlddhu82om-image.image?=&rk3s=8031ce6d&x-expires=1783493750&x-signature=R0n9lDQhHBRUKYAOsUoKsxY2DL0%3D)
def batch_insert_products(products_list):
    """
    products_list: 列表，每个元素是 (platform, name, price, sales, collect_time)
    """
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    
    # 批量插入
    cursor.executemany('''
    INSERT INTO products (platform, product_name, price, sales, collect_time)
    VALUES (?, ?, ?, ?, ?)
    ''', products_list)
    
    conn.commit()
    count = cursor.rowcount
    conn.close()
    return count

# 影刀中调用
# 先从Excel或采集结果中准备数据列表
# 调用Python执行批量插入

4. 查询数据

# ===== 查询数据 =====
import sqlite3

def query_products_by_platform(platform):
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    
    cursor.execute('''
    SELECT product_name, price, sales, collect_time
    FROM products
    WHERE platform = ?
    ORDER BY collect_time DESC
    LIMIT 100
    ''', (platform,))
    
    rows = cursor.fetchall()
    conn.close()
    return rows

# 返回结果给影刀

5. 更新与删除

# ===== 更新数据 =====
def update_price(product_id, new_price):
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    cursor.execute('UPDATE products SET price = ? WHERE id = ?', (new_price, product_id))
    conn.commit()
    conn.close()

# ===== 删除数据 =====
def delete_old_data(days):
    import datetime
    cutoff = (datetime.datetime.now() - datetime.timedelta(days=days)).strftime('%Y-%m-%d')
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    cursor.execute('DELETE FROM products WHERE collect_time < ?', (cutoff,))
    conn.commit()
    conn.close()

---

第二部分：MySQL数据库操作

MySQL适合多人共享和生产环境。需要在服务器上安装MySQL，并在Python中安装 pymysql 库。

在影刀中安装 pymysql： 在“执行Python代码”指令详情面板中，点击“安装依赖”，输入 pymysql。

1. 连接MySQL与建表

# ===== 连接MySQL =====
import pymysql

# 数据库连接配置（从加密变量或配置文件中读取）
config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'rpa_user',
    'password': 'rpa_password',  # 建议用加密变量
    'database': 'rpa_data',
    'charset': 'utf8mb4'
}

conn = pymysql.connect(**config)
cursor = conn.cursor()

# 建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    platform VARCHAR(50) NOT NULL,
    product_name VARCHAR(255) NOT NULL,
    price DECIMAL(10,2),
    sales INT,
    collect_time DATETIME,
    INDEX idx_platform (platform),
    INDEX idx_collect_time (collect_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
''')

conn.commit()
conn.close()

2. 批量插入MySQL

# ===== MySQL批量插入（高效） =====
import pymysql

def batch_insert_mysql(products_list):
    conn = pymysql.connect(**config)
    cursor = conn.cursor()
    
    # 使用executemany批量插入
    sql = '''
    INSERT INTO products (platform, product_name, price, sales, collect_time)
    VALUES (%s, %s, %s, %s, %s)
    '''
    cursor.executemany(sql, products_list)
    conn.commit()
    count = cursor.rowcount
    conn.close()
    return count

3. 查询与分页

# ===== MySQL分页查询 =====
def query_paginated(page, page_size=100):
    offset = (page - 1) * page_size
    conn = pymysql.connect(**config)
    cursor = conn.cursor()
    
    cursor.execute('''
    SELECT id, platform, product_name, price, sales, collect_time
    FROM products
    ORDER BY collect_time DESC
    LIMIT %s OFFSET %s
    ''', (page_size, offset))
    
    rows = cursor.fetchall()
    conn.close()
    return rows

---

第三部分：影刀中调用数据库操作的模板

场景1：采集后直接入库

# ===== 采集流程中直接入库 =====
# 影刀主流程采集数据，然后调用Python入库

# 1. 采集数据（影刀段）
获取相似元素列表("//div[@class='product']", 存入列表=产品卡片)
设置变量(入库数据 = [])

按列表循环(产品卡片, 循环项=card):
    平台 = "淘宝"
    商品名 = 获取元素文本(card + "//h3")
    价格 = 提取数字(获取元素文本(card + "//span[@class='price']"))
    销量 = 提取数字(获取元素文本(card + "//span[@class='sales']"))
    采集时间 = 获取当前时间()
    
    行数据 = [平台, 商品名, 价格, 销量, 采集时间]
    将 行数据 加入列表(入库数据)
结束循环

# 2. 批量入库（调用Python）
执行Python代码(
    代码=批量插入SQLite代码,
    输入参数: data_list=入库数据,
    输出: inserted_count
)

输出日志("入库完成，共 " + inserted_count + " 条")

场景2：从数据库读取配置

# ===== 从数据库读取采集关键词 =====
执行Python代码(
    代码="""
import sqlite3
conn = sqlite3.connect('C:/RPA_Data/config.db')
cursor = conn.cursor()
cursor.execute('SELECT keyword FROM keywords WHERE enabled=1')
rows = cursor.fetchall()
conn.close()
keywords = [row[0] for row in rows]
""",
    输出参数: keywords=keywords
)

# 影刀循环读取的关键词列表
按列表循环(keywords, 循环项=keyword):
    输出日志("采集关键词：" + keyword)
    # 执行搜索采集...
结束循环

场景3：数据去重（用数据库代替Excel）

# ===== 用数据库做去重，避免内存溢出 =====
# 当数据量很大时，逐条检查数据库是否存在

def insert_if_not_exists(platform, name, price, sales, collect_time):
    conn = sqlite3.connect('C:/RPA_Data/products.db')
    cursor = conn.cursor()
    
    # 检查是否已存在（按平台+商品名）
    cursor.execute('SELECT id FROM products WHERE platform=? AND product_name=?', (platform, name))
    existing = cursor.fetchone()
    
    if existing:
        # 已存在，更新价格
        cursor.execute('UPDATE products SET price=?, sales=?, collect_time=? WHERE id=?', 
                       (price, sales, collect_time, existing[0]))
        inserted = False
    else:
        # 不存在，插入
        cursor.execute('INSERT INTO products (platform, product_name, price, sales, collect_time) VALUES (?,?,?,?,?)',
                       (platform, name, price, sales, collect_time))
        inserted = True
    
    conn.commit()
    conn.close()
    return inserted

---

数据库 vs Excel 存储决策表

判断维度	使用Excel	使用SQLite	使用MySQL
数据量	<1万行	1-10万行	>10万行
查询需求	简单筛选	中等复杂度	复杂关联查询
多用户	❌	❌	✅
部署成本	低	低	中
备份方式	复制文件	复制文件	导出SQL

---

常见问题速查

问题	原因	解决方法
连接SQLite失败	路径不存在	确保目录存在
连接MySQL失败	服务未启动或配置错误	检查host/port/user/password
中文乱码	编码设置错误	使用 utf8mb4
批量插入速度慢	逐条插入	改用 executemany
数据库被锁（SQLite）	多线程同时写	用连接池或串行化写入

---

推荐资源

• SQLite官方文档
• MySQL官方文档
• Python sqlite3 模块文档
• Python pymysql 模块文档

---

#影刀RPA #RPA自动化 #SQLite #MySQL #数据库 #数据存储

作者：林焱

本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。