影刀RPA进阶教程：MySQL数据库读写与Pandas数据互通实战

影刀RPA进阶教程：MySQL数据库读写与Pandas数据互通实战

Excel文件越来越多，一个日报汇总表三个月就膨胀到50MB，打开都要等十几秒。店群数据做到一定规模，必须要上数据库——MySQL是最合适的选择。但影刀RPA本身没有内置数据库操作指令，数据怎么落地成了难题。

用Python桥接影刀RPA和MySQL，Pandas做中间层，数据流就能跑通。

我也是非技术背景，第一次装MySQL折腾了两天。但配置好后，数据从Excel到数据库再到飞书看板的全链路就打通了。每天上千条店铺数据自动入库，查询和分析效率比Excel高了至少10倍。

MySQL环境准备

安装MySQL

推荐用MySQL Community Server 8.0。安装时注意：

安装完成后验证：

# 打开命令行，测试连接
mysql -u root -p
# 输入密码后看到 mysql> 提示符即安装成功

# 创建电商数据库
CREATE DATABASE ecommerce_data CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE ecommerce_data;

安装Python连接库

影刀RPA的Python环境需要安装 pymysql 和 pandas：

# 在影刀RPA的Python环境中安装依赖
pip install pymysql pandas sqlalchemy

影刀RPA使用的Python解释器路径在安装目录下。可以在影刀RPA的「执行Python脚本」指令中先跑一段检测代码确认库是否安装成功：

# Python环境检测
try:
import pymysql
import pandas
from sqlalchemy import create_engine
print("所有依赖库已安装")
print(f"pymysql版本: {pymysql.__version__}")
print(f"pandas版本: {pandas.__version__}")
except ImportError as e:
print(f"缺少依赖库: {e}")

数据库表设计

电商数据表结构

— 店铺日销售数据表
CREATE TABLE shop_daily_sales (
id INT AUTO_INCREMENT PRIMARY KEY,
platform VARCHAR(20) NOT NULL COMMENT '平台：拼多多/淘宝/TEMU/TikTok',
shop_name VARCHAR(100) NOT NULL COMMENT '店铺名称',
report_date DATE NOT NULL COMMENT '数据日期',
sales DECIMAL(12,2) DEFAULT 0 COMMENT '销售额(元)',
orders INT DEFAULT 0 COMMENT '订单数',
visitors INT DEFAULT 0 COMMENT '访客数',
conversion_rate DECIMAL(5,2) DEFAULT 0 COMMENT '转化率(%)',
refund_amount DECIMAL(12,2) DEFAULT 0 COMMENT '退款金额(元)',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_shop_date (platform, shop_name, report_date) COMMENT '同店铺同日期唯一'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='店铺日销售数据';

— 商品维度数据表
CREATE TABLE product_daily (
id INT AUTO_INCREMENT PRIMARY KEY,
platform VARCHAR(20) NOT NULL,
shop_name VARCHAR(100) NOT NULL,
product_id VARCHAR(50) NOT NULL COMMENT '商品ID',
product_title VARCHAR(500) COMMENT '商品标题',
report_date DATE NOT NULL,
sales DECIMAL(12,2) DEFAULT 0 COMMENT '商品销售额',
orders INT DEFAULT 0 COMMENT '商品订单数',
stock INT DEFAULT 0 COMMENT '库存数',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_product_date (platform, shop_name, product_id, report_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品维度日数据';

索引优化

电商数据查询频繁，需要合理建索引：

— 按日期查询（日报场景最频繁）
CREATE INDEX idx_report_date ON shop_daily_sales(report_date);

— 按平台+日期查询
CREATE INDEX idx_platform_date ON shop_daily_sales(platform, report_date);

— 组合索引：多平台数据看板最常用
CREATE INDEX idx_platform_shop_date ON shop_daily_sales(platform, shop_name, report_date);

影刀RPA写入MySQL

数据采集后自动入库

# 影刀RPA流程中的MySQL数据写入
# 在「执行Python脚本」指令中运行

import pymysql
import pandas as pd
from datetime import datetime

# 1. 数据库连接配置（从影刀变量读取）
db_config = {
"host": "127.0.0.1",
"port": 3306,
"user": flow.get_variable("db_user"),
"password": flow.get_variable("db_password"),
"database": "ecommerce_data",
"charset": "utf8mb4"
}

# 2. 从影刀变量获取采集到的数据
data_json = flow.get_variable("pdd_collected_data")
df = pd.read_json(data_json)

# 3. 数据清洗：确保字段类型正确
df['report_date'] = pd.to_datetime(df['report_date']).dt.date
df['platform'] = df['platform'].astype(str)
df['shop_name'] = df['shop_name'].astype(str)
df['sales'] = pd.to_numeric(df['sales'], errors='coerce').fillna(0)
df['orders'] = pd.to_numeric(df['orders'], errors='coerce').fillna(0).astype(int)

# 4. 创建数据库连接
conn = pymysql.connect(**db_config)
cursor = conn.cursor()

# 5. 使用 INSERT … ON DUPLICATE KEY UPDATE 实现插入或更新
# 如果同店铺同日期已有数据，则更新；否则插入
insert_sql = """
INSERT INTO shop_daily_sales
(platform, shop_name, report_date, sales, orders, visitors, conversion_rate, refund_amount)
VALUES
(%s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
sales = VALUES(sales),
orders = VALUES(orders),
visitors = VALUES(visitors),
conversion_rate = VALUES(conversion_rate),
refund_amount = VALUES(refund_amount),
updated_at = NOW()
"""

# 6. 逐行写入
success = 0
fail = 0
for idx, row in df.iterrows():
try:
cursor.execute(insert_sql, (
row['platform'],
row['shop_name'],
row['report_date'],
row['sales'],
row['orders'],
row['visitors'],
row['conversion_rate'],
row['refund_amount']
))
success += 1
except Exception as e:
fail += 1
print(f"第{idx}行写入失败: {str(e)}")
print(f"数据: {row.to_dict()}")

# 7. 提交事务并关闭连接
conn.commit()
cursor.close()
conn.close()

print(f"写入完成! 成功: {success}条, 失败: {fail}条")

批量写入优化

单条INSERT在大数据量下效率很低。改用批量方式：

# 批量写入（一次提交多条，效率提升10倍+）
# 每500条一批
batch_size = 500
data_tuples = []

for idx, row in df.iterrows():
data_tuples.append((
row['platform'], row['shop_name'], row['report_date'],
row['sales'], row['orders'], row['visitors'],
row['conversion_rate'], row['refund_amount']
))

if len(data_tuples) >= batch_size:
cursor.executemany(insert_sql, data_tuples)
conn.commit()
print(f"已写入{len(data_tuples)}条")
data_tuples = []

# 处理剩余不足一批的数据
if data_tuples:
cursor.executemany(insert_sql, data_tuples)
conn.commit()
print(f"已写入最后{len(data_tuples)}条")

Pandas从MySQL读取数据做分析

数据入库后，影刀RPA可以直接用Pandas从MySQL读取数据做分析：

# 从MySQL读取近7天数据做汇总分析
import pandas as pd
from sqlalchemy import create_engine

# SQLAlchemy连接串
db_url = "mysql+pymysql://root:password@127.0.0.1:3306/ecommerce_data?charset=utf8mb4"
engine = create_engine(db_url)

# 读取近7天数据
sql = """
SELECT
platform,
shop_name,
report_date,
sales,
orders,
conversion_rate,
refund_amount
FROM shop_daily_sales
WHERE report_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
ORDER BY report_date DESC, sales DESC
"""

df = pd.read_sql(sql, engine)

# Pandas分析：各平台汇总
summary = df.groupby('platform').agg({
'sales': 'sum',
'orders': 'sum',
'conversion_rate': 'mean'
}).round(2)

print("近7天各平台汇总:")
print(summary)

# Pandas分析：每日趋势
daily_trend = df.groupby('report_date')['sales'].sum()
print("\\n每日销售额趋势:")
print(daily_trend)

# 输出到影刀变量（供后续飞书推送使用）
flow.set_variable("analysis_result", summary.to_json(force_ascii=False))
flow.set_variable("daily_trend", daily_trend.to_json(force_ascii=False))

Pandas与影刀Excel双向互通

很多时候数据在Excel和MySQL之间来回倒。Pandas是最方便的中间层：

# Excel → Pandas → MySQL
df = pd.read_excel("导出数据.xlsx")
# 清洗后写入数据库…（代码同上）

# MySQL → Pandas → Excel
df = pd.read_sql("SELECT * FROM shop_daily_sales WHERE report_date = CURDATE()", engine)
df.to_excel("今日数据导出.xlsx", index=False)

# MySQL → Pandas → 飞书多维表格
df = pd.read_sql(sql, engine)
# 转JSON写入飞书Base…（调用飞书API）

定时数据清理与备份

数据库不能无限增长，需要定期清理和备份：

# 数据归档脚本：保留近90天数据，超过90天的导出为CSV备份
import pandas as pd
import pymysql
from datetime import datetime, timedelta

conn = pymysql.connect(host="127.0.0.1", user="root", password="password", database="ecommerce_data")

# 查询90天前的数据
cutoff_date = (datetime.now() – timedelta(days=90)).strftime('%Y-%m-%d')
sql = f"SELECT * FROM shop_daily_sales WHERE report_date < '{cutoff_date}'"

df = pd.read_sql(sql, conn)

if len(df) > 0:
# 备份为CSV
backup_file = f"C:\\\\RPA_Backup\\\\shop_sales_before_{cutoff_date}.csv"
df.to_csv(backup_file, index=False, encoding='utf-8-sig')
print(f"已备份{len(df)}条记录到: {backup_file}")

# 删除已备份数据
cursor = conn.cursor()
delete_sql = f"DELETE FROM shop_daily_sales WHERE report_date < '{cutoff_date}'"
cursor.execute(delete_sql)
conn.commit()
print(f"已删除{len(df)}条旧记录")
else:
print("无需要清理的数据")

conn.close()

在影刀RPA中把这个脚本设为每月1号自动执行，避免数据库膨胀。

常见问题/易错速查

推荐资源

• MySQL官方文档——建表语句和索引优化参考
• Pandas中文文档 read_sql 章节——了解数据库查询的完整参数
• DBeaver（免费数据库管理工具）——可视化查看MySQL数据，比命令行方便
• 

#影刀RPA #RPA自动化 #MySQL #数据库 #Pandas #数据清洗 #电商数据 #数据存储

作者：林焱

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