影刀RPA进阶教程：低代码平台与RPA混合开发搭建业务系统实战

影刀RPA进阶教程：低代码平台与RPA混合开发搭建业务系统实战

痛点切入

中小企业信息化的最大困局：买不起商业软件，雇不起开发团队。

一个典型的场景：公司需要一个供应商管理系统，IT预算只有5万。买SAP要50万起，自研前端+后端+数据库最少两个程序员三个月。最后只能继续用Excel，每月靠行政手动更新几十个表格，表格一多就乱、一乱就错。

直到我们发现了一条新路：低代码平台搭壳，RPA做采集和操作。

低代码平台（明道云/简道云/轻流/得帆）负责数据管理、表单、流程审批、报表展示。影刀RPA负责对接老旧系统（那些没有API的ERP、网页版进销存），做数据采集和写入。两者配合，搭建速度极快，成本低到可以忽略。

本文通过一个真实的供应商管理系统案例，展示如何搭建这套混合架构。

架构设计

┌─────────────────────────────────────────────────┐
│ 低代码平台（明道云） │
│ │
│ ┌──────────┐ ┌──────────┐ ┌───────────────┐ │
│ │ 供应商表单 │ │ 合同管理 │ │ 绩效评分看板 │ │
│ └──────────┘ └──────────┘ └───────────────┘ │
│ ┌──────────┐ ┌──────────┐ ┌───────────────┐ │
│ │ 审批流程 │ │ 数据报表 │ │ 自动通知提醒 │ │
│ └──────────┘ └──────────┘ └───────────────┘ │
│ │
│ ▲ API接收 │ API调用 ▼ │
└─────────┼─────────────────┼────────────────────┘
│ │
┌─────────▼─────────────────▼────────────────────┐
│ RPA层（影刀RPA） │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 数据采集流程 │ │ 数据回写流程 │ │
│ │ ERP → 低代码 │ │ 低代码 → ERP │ │
│ └──────────────┘ └──────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 定时同步任务 │ │ 异常处理流程 │ │
│ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────┘
│ ▲
▼ │
┌─────────────────────────────────────────────────┐
│ 老旧系统（没有API） │
│ 财务ERP │ 进销存系统 │ 网页版审批平台 │
└─────────────────────────────────────────────────┘

分工原则：

• 低代码平台负责：数据模型、表单UI、审批流、权限管理、报表Dashboard
• RPA负责：对接无API的老旧系统、批量数据导入导出、跨系统操作串联

完整案例：供应商管理系统

系统需求

第一步：在明道云中搭建数据模型

表结构设计：

1. 供应商信息表 (supplier)
– 供应商编号 (文本, 唯一)
– 供应商名称 (文本, 必填)
– 供应商类别 (单选: 原材料/辅料/服务/其他)
– 联系人 (文本)
– 联系电话 (电话)
– 地址 (地址)
– 状态 (单选: 正常/停用/黑名单)
– 创建时间 (日期时间, 自动生成)

2. 合同表 (contract)
– 关联供应商 (关联记录 → supplier)
– 合同编号 (文本, 唯一)
– 合同名称 (文本)
– 合同金额 (数字)
– 签订日期 (日期)
– 到期日期 (日期)
– 合同附件 (附件)
– 续签状态 (单选: 正常/即将到期/已到期/已续签)

3. 绩效评分表 (performance)
– 关联供应商 (关联记录 → supplier)
– 评分周期 (文本, 如 "2024-Q1")
– 交货及时率 (数字, %)
– 产品合格率 (数字, %)
– 服务满意度 (数字, 1-5)
– 综合评分 (公式: 自动计算)
– 评分来源 (单选: 自动采集/人工评分)

在明道云操作：应用管理 → 新建工作表 → 按上述字段逐个添加 → 设置关联关系。

第二步：配置审批流程

在明道云工作流中创建"合同续签审批"流程：

触发器：合同表.到期日期 < TODAY() + 30天 且 续签状态 = "正常"
│
▼
通知节点：发送通知给采购主管
│
▼
填写节点：采购主管填写续签意见和金额
│
▼
审批节点：部门经理审批 (金额<10万自动通过)
│
├── 通过 → 更新合同.续签状态 = "已续签" → 通知供应商联系人
│
└── 拒绝 → 更新合同.续签状态 = "即将到期" → 通知采购主管重新谈判

第三步：RPA数据采集流程

这是混合架构最核心的部分。老旧ERP系统没有API，影刀RPA通过Selenium自动化操作网页来提取数据。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
影刀RPA + 明道云 — 供应商绩效数据采集
用途：从ERP系统采集供应商交货数据，同步到明道云
作者：林焱
"""

import requests
import json
from datetime import datetime, timedelta
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait, Select
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException

class MingdaoAPIClient:
"""
明道云API客户端

前提：在明道云中创建API密钥（应用管理 → API文档 → 获取API Key）
"""

def __init__(self, app_key, app_secret, host="https://api.mingdao.com"):
self.app_key = app_key
self.app_secret = app_secret
self.host = host
self.access_token = None
self._authenticate()

def _authenticate(self):
"""获取访问令牌"""
url = f"{self.host}/oauth2/token"
data = {
"grant_type": "client_credentials",
"client_id": self.app_key,
"client_secret": self.app_secret
}
resp = requests.post(url, data=data)
if resp.status_code == 200:
self.access_token = resp.json()["access_token"]
print("[明道云] 认证成功")
else:
raise Exception(f"认证失败: {resp.text}")

def _headers(self):
"""请求头"""
return {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json"
}

def get_worksheet_rows(self, worksheet_id, view_id=None, filter_json=None):
"""
查询工作表数据

参数:
worksheet_id: 工作表ID
view_id: 视图ID（可选）
filter_json: 过滤条件（可选）
"""
url = f"{self.host}/api/v2/open/worksheet/getFilterRows"
params = {
"worksheetId": worksheet_id,
"pageSize": 500,
"pageIndex": 1
}
if view_id:
params["viewId"] = view_id
if filter_json:
params["filter"] = json.dumps(filter_json, ensure_ascii=False)

resp = requests.post(url, headers=self._headers(), json=params)
if resp.status_code == 200:
data = resp.json()
return data.get("data", {}).get("rows", [])
else:
print(f"[明道云] 查询失败: {resp.text}")
return []

def add_worksheet_row(self, worksheet_id, row_data):
"""
新增工作表行

参数:
worksheet_id: 工作表ID
row_data: 行数据字典
例如: {
"supplier_name": "XX科技",
"delivery_rate": 95.5,
"quality_rate": 98.2
}
"""
url = f"{self.host}/api/v2/open/worksheet/addRow"
controls = []
for field_name, value in row_data.items():
controls.append({
"controlId": field_name, # 明道云的字段控件ID
"value": str(value) if value is not None else ""
})

payload = {
"worksheetId": worksheet_id,
"controls": controls
}

resp = requests.post(url, headers=self._headers(), json=payload)
if resp.status_code == 200:
result = resp.json()
return result.get("data", {}).get("rowId")
else:
print(f"[明道云] 新增失败: {resp.text}")
return None

def update_worksheet_row(self, worksheet_id, row_id, update_data):
"""
更新工作表行

参数:
worksheet_id: 工作表ID
row_id: 行ID
update_data: 要更新的字段
"""
url = f"{self.host}/api/v2/open/worksheet/editRow"
controls = []
for field_name, value in update_data.items():
controls.append({
"controlId": field_name,
"value": str(value) if value is not None else ""
})

payload = {
"worksheetId": worksheet_id,
"rowId": row_id,
"controls": controls
}

resp = requests.post(url, headers=self._headers(), json=payload)
return resp.status_code == 200

class ERPSupplierDataCollector:
"""
从老旧ERP网页采集供应商绩效数据

这个ERP是10年前的Java Web系统，没有任何API，
只能通过网页操作来提取数据。
"""

def __init__(self):
self.driver = None

def start(self):
"""启动浏览器并登录ERP"""
from selenium.webdriver.chrome.options import Options
options = Options()
options.add_argument('–disable-gpu')
options.add_argument('–window-size=1366,768')
self.driver = webdriver.Chrome(options=options)

def login_erp(self, username, password):
"""登录ERP系统"""
self.driver.get("http://erp.internal/login")
# 填写用户名
self.driver.find_element(By.ID, "username").send_keys(username)
# 填写密码
self.driver.find_element(By.ID, "password").send_keys(password)
# 选择数据权限（某些ERP登录时需要选择公司/部门）
select = Select(self.driver.find_element(By.ID, "company"))
select.select_by_visible_text("总公司")
# 点击登录
self.driver.find_element(By.CSS_SELECTOR, "button[type='submit']").click()
# 等待登录成功（首页加载完成）
WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.ID, "mainMenu"))
)
print("[ERP] 登录成功")

def collect_supplier_performance(self, period):
"""
采集指定周期的供应商绩效数据

参数:
period: 统计周期，如 "2024-Q1"
返回:
[{"supplier_code": "S001", "delivery_rate": 95.5, …}, …]
"""
# 步骤1：导航到供应商绩效报表页面
# ERP的导航菜单通常是多层iframe嵌套的树形菜单
menu_frame = self.driver.find_element(By.NAME, "menuFrame")
self.driver.switch_to.frame(menu_frame)
self.driver.find_element(By.LINK_TEXT, "供应商管理").click()
self.driver.find_element(By.LINK_TEXT, "绩效考核报表").click()
self.driver.switch_to.default_content()

# 步骤2：在主内容区操作
content_frame = self.driver.find_element(By.NAME, "contentFrame")
self.driver.switch_to.frame(content_frame)

# 步骤3：设置查询条件
# 选择统计周期
period_select = Select(self.driver.find_element(By.ID, "period"))
period_select.select_by_visible_text(period)
# 点击查询按钮
self.driver.find_element(By.ID, "btnSearch").click()

# 步骤4：等待报表数据加载
WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.CSS_SELECTOR, "#reportTable tbody tr"))
)
print(f"[ERP] 报表数据已加载")

# 步骤5：提取表格数据
rows = self.driver.find_elements(By.CSS_SELECTOR, "#reportTable tbody tr")
results = []
for row in rows:
cells = row.find_elements(By.TAG_NAME, "td")
if len(cells) < 5:
continue

# 解析每一列的数据
supplier_code = cells[0].text.strip()
supplier_name = cells[1].text.strip()

# 处理百分比格式（ERP中可能是"95.5%"）
delivery_rate_str = cells[2].text.strip().replace('%', '')
quality_rate_str = cells[3].text.strip().replace('%', '')

try:
delivery_rate = float(delivery_rate_str)
quality_rate = float(quality_rate_str)
except ValueError:
print(f"[ERP] 解析失败: {supplier_code} {delivery_rate_str}/{quality_rate_str}")
continue

# 跳过无效数据
if delivery_rate <= 0 and quality_rate <= 0:
continue

results.append({
"supplier_code": supplier_code,
"supplier_name": supplier_name,
"delivery_rate": delivery_rate,
"quality_rate": quality_rate,
"period": period,
"collected_at": datetime.now().isoformat()
})

self.driver.switch_to.default_content()
print(f"[ERP] 采集完成: {len(results)} 条供应商数据")
return results

def stop(self):
"""关闭浏览器"""
if self.driver:
self.driver.quit()

class SupplierDataSync:
"""
供应商数据同步器
负责协调ERP采集和明道云写入
"""

def __init__(self, mingdao_client: MingdaoAPIClient):
self.mingdao = mingdao_client
self.collector = ERPSupplierDataCollector()

def sync_performance_data(self, period: str):
"""
同步供应商绩效数据
完整流程：ERP采集 → 匹配明道云供应商 → 写入绩效表
"""
print(f"\\n[同步] 开始同步供应商绩效数据 – {period}")

# 阶段1：从ERP采集数据
try:
self.collector.start()
self.collector.login_erp("rpa_bot", "rpa_pass_2024")
erp_data = self.collector.collect_supplier_performance(period)
finally:
self.collector.stop()

if not erp_data:
print("[同步] 未采集到数据，终止")
return {"total": 0, "synced": 0, "skipped": 0}

# 阶段2：从明道云读取供应商列表（用于匹配）
suppliers = self.mingdao.get_worksheet_rows(
worksheet_id="supplier_worksheet_id" # 替换为实际ID
)

# 构建供应商编号 -> 明道云行ID的映射
supplier_map = {}
for row in suppliers:
code = row.get("supplier_code") # 字段控件ID
if code:
supplier_map[code] = row.get("rowId")

# 阶段3：写入绩效数据
synced = 0
skipped = 0
for item in erp_data:
supplier_code = item["supplier_code"]

# 检查供应商是否已在明道云中
if supplier_code not in supplier_map:
print(f"[同步] 跳过未知供应商: {supplier_code} – {item['supplier_name']}")
skipped += 1
continue

# 构建写入数据
row_data = {
"supplier_relation": supplier_map[supplier_code], # 关联供应商记录
"period": period,
"delivery_rate": item["delivery_rate"],
"quality_rate": item["quality_rate"],
"data_source": "自动采集"
}

# 写入明道云绩效表
row_id = self.mingdao.add_worksheet_row(
worksheet_id="performance_worksheet_id", # 替换为实际ID
row_data=row_data
)
if row_id:
synced += 1
print(f"[同步] {supplier_code}: 绩效数据已写入 (rowId={row_id})")
else:
skipped += 1

result = {"total": len(erp_data), "synced": synced, "skipped": skipped}
print(f"[同步] 完成: 采集{result['total']}条 → 同步{result['synced']}条 → 跳过{result['skipped']}条")
return result

# ===== 在影刀RPA流程中的使用 =====
# 影刀流程编排：
# 1. Python节点：初始化
mingdao = MingdaoAPIClient(
app_key="your_app_key",
app_secret="your_app_secret"
)
sync = SupplierDataSync(mingdao)

# 2. Python节点：执行同步
result = sync.sync_performance_data("2024-Q1")

# 3. Python节点：将结果写入影刀变量
# 影刀变量: sync_result = json.dumps(result)

第四步：配置明道云工作流自动处理

当RPA写入绩效数据后，明道云自动触发：

触发器：performance表.新增记录 且 data_source = "自动采集"
│
▼
查询节点：查询 supplier 表中对应的供应商记录
│
▼
条件分支节点：
├── delivery_rate >= 95 且 quality_rate >= 98
│ → 更新 supplier.状态 = "正常"
│
├── delivery_rate < 80 或 quality_rate < 85
│ → 更新 supplier.状态 = "观察中"
│ → 发送飞书通知给采购主管
│
└── delivery_rate < 60 或 quality_rate < 70
→ 更新 supplier.状态 = "黑名单"
→ 创建审批任务：供应商淘汰审批

第五步：数据大屏与报表

在明道云中创建仪表盘（数据大屏）：

布局设计：
┌─────────────────────────────────────────────┐
│ 供应商管理数据大屏 – 实时 │
├─────────────────────┬───────────────────────┤
│ 供应商总数: 156 │ 正常: 142 观察中: 9 │
│ 即将到期合同: 3 │ 黑名单: 5 │
├─────────────────────┴───────────────────────┤
│ [供应商绩效排名 TOP10 柱状图] │
│ │
├─────────────────────────────────────────────┤
│ [供应商类别分布 饼图] [合同到期预警 甘特图] │
│ │
└─────────────────────────────────────────────┘

明道云仪表盘配置步骤：

常见问题速查

Q1: 明道云API返回的数据中文乱码

现象：通过API读取的中文字段显示乱码。

原因：明道云API默认使用UTF-8，但如果ERP系统使用了GBK编码，中间转换可能出现问题。

解决：

# 在使用明道云API前确保正确编码
resp = requests.post(url, headers={"Content-Type": "application/json; charset=utf-8"}, json=payload)
resp.encoding = 'utf-8'

# 从ERP采集时也要注意
text = driver.find_element(By.ID, "name").text
# Selenium自动处理编码，但如果遇到乱码：
text = text.encode('latin1').decode('gbk')

Q2: ERP系统太老，页面加载极慢

现象：登录ERP后等30秒页面才加载完，RPA经常超时。

原因：老系统服务器性能不足，前端大量使用过时的技术（如大量嵌套table布局）。

解决：

# 1. 增加通用等待超时
driver.set_page_load_timeout(60)

# 2. 用显式等待替代固定sleep
# 不好
time.sleep(30)
# 好
WebDriverWait(driver, 60).until(
EC.presence_of_element_located((By.ID, "reportTable"))
)

# 3. 如果页面有加载动画，等待动画消失
WebDriverWait(driver, 60).until_not(
EC.presence_of_element_located((By.CSS_SELECTOR, ".loading-spinner"))
)

Q3: 同步时发现ERP和明道云供应商名称不一致

现象：同一个供应商在ERP叫"XX科技有限公司"，在明道云叫"XX科技"，匹配失败。

原因：两个系统独立维护，名称标准不统一。

解决：使用供应商编号（唯一键）做匹配，而不是名称：

# 用编号匹配（推荐）
supplier_map = {row["code"]: row["rowId"] for row in suppliers}

# 如果实在没有编号，用模糊匹配兜底
from difflib import SequenceMatcher
def fuzzy_match(name, candidates, threshold=0.8):
for candidate in candidates:
ratio = SequenceMatcher(None, name, candidate["name"]).ratio()
if ratio >= threshold:
return candidate
return None

Q4: 明道云API调用频率限制

现象：批量写入时报 Too Many Requests。

原因：明道云免费版API调用频率限制为100次/分钟。

解决：

import time
def rate_limited_add(api, worksheet_id, rows, rate_limit=80):
"""带限速的批量写入"""
for i, row in enumerate(rows):
api.add_worksheet_row(worksheet_id, row)
if (i + 1) % rate_limit == 0:
print(f"已处理 {i+1}/{len(rows)}，限速等待60秒…")
time.sleep(60)

Q5: RPA流程中断后数据不同步

现象：RPA同步到一半中断了，ERP和明道云数据不一致。

原因：没有事务机制，部分数据已写入明道云。

解决：增量同步 + 幂等设计：

# 1. 用batch_id标记每次同步
batch_id = f"sync_{datetime.now().strftime('%Y%m%d%H%M%S')}"

# 2. 写入时带上batch_id
row_data["sync_batch_id"] = batch_id

# 3. 同步前查询上次成功同步的时间戳
last_sync = get_last_sync_time()
erp_data = collector.collect_supplier_performance(
period,
from_date=last_sync # 只采集上次同步之后的数据
)

# 4. 写入前检查是否已存在（幂等）
existing = api.get_worksheet_rows(
worksheet_id="performance",
filter_json={"sync_batch_id": batch_id}
)
if existing:
print("该批次已同步过，跳过")

推荐资源

• 明道云开放平台：https://open.mingdao.com — API文档
• 简道云开放平台：https://www.jiandaoyun.com/open — 简道云API（类似方案）
• 影刀RPA + 简道云案例：影刀官方社区 → 搜索"简道云"
• 得帆云：https://www.definesys.com — 企业级低代码平台

内容标签

#影刀RPA #低代码 #明道云 #混合开发 #企业信息化 #数据同步

作者：林焱