RPA:代码到低代码的实践
本文正在完善中
本文章给出基于代码的和基于RPA软件实现的RPA,在具体实施上的常见策略。本文不是技术介绍,涉及的技术是为解决问题为主,并不追求所谓“先进”或者“强大”
从代码到低代码:RPA自动化的双轨策略与工程实践
1. 自动化路径的分化
在企业数字化进程中,机器人流程自动化(RPA)已成为提升运营效率的关键技术。然而,实施RPA常面临路径选择:技术团队倾向通过编程实现高度定制的自动化,而业务团队则偏好可视化、低代码的RPA软件以快速交付。
本文系统梳理基于代码与基于RPA软件两类RPA实施路径,在工程实践中的常用策略、技术选型与健壮性设计,并提供可操作的选型建议,助力读者合理决策。
2. 路径一:基于代码的RPA实现
考虑易用性以及对各类RPA的兼容性,此部分以 Python3 作为示例编程语言。
2.1 核心技术栈
| 自动化场景 | 推荐库/工具 | 说明 |
|---|---|---|
| 桌面应用自动化 | PyAutoGUI, pywinauto | 适用于面应用 |
| 浏览器自动化 | Playwright, Selenium, DrissionPage | Playwright 最为强大,DrissionPage 较均衡 |
| 光学字符识别(OCR) | PaddleOCR, Tesseract | 用于处理验证码、截图文本提取等非结构化输入 |
| 文件与系统操作 | pathlib, tempfile, shutil | 提供文件路径与临时文件管理 |
2.2 安全与健壮性保障
- 文件操作
- 资源释放:始终使用(
with open(...))操作文件,确保资源释放; - 原子操作:使用
tempfile生成临时文件,使用os.replace()进行原子写入,可保证文件完整性。
- 资源释放:始终使用(
- 超时和异常
- 等待机制:确保加载完成后操作,保证正确性和稳健性。常见的自动化库大多有自带等待控制机制,如
Selenium的WebDriverWait(driver, 5) - 异常处理:具备对错误的重试和降级流程。重试指多次执行失败的流程直至成功或达到次数上限,降级指失败后进行处理或直接进行下一步骤。
- 断点恢复:对于失败的流程,可以原地重试,或者广度遍历完成后重试,后者就需要记录状态。如果是多次运行程序,记录状态并在流程中对完成的部分跳过、对失败的部分重试执行,可实现断点恢复的功能,可以使流程更高效。状态记录可使用简单的
xlsx,csv,或者json,以及更强大的pickle和数据库引擎,如SQLite。
- 等待机制:确保加载完成后操作,保证正确性和稳健性。常见的自动化库大多有自带等待控制机制,如
2.3 示例
给出一个代码示例,涉及常见库、安全文件操作、进度记录和恢复、日志和异常处理。
演示在线非商业网页自动化: https://quotes.toscrape.com/ 进行登录;获取Top Ten tags,并获取每个标签第一个帖子的内容,保存至xlsx文件。
点击展开 demo.py
import json, os, tempfile, logging
from typing import Callable, Optional
from DrissionPage import ChromiumPage, ChromiumOptions
from openpyxl import Workbook
# 设置日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("app.log", encoding="utf-8"),
logging.StreamHandler()
]
)
DATA_FILE = "datacache.json"
OUTPUT_FILE = "帖子数据.xlsx"
EDGE_PATH = r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe'
def safe_modify_file(
target_path: str,
processor: Callable[[str, str], None]
) -> None:
"""
原子操作文件
Args:
target_path: 要修改的目标文件路径
processor: 业务逻辑函数 (源文件路径, 临时输出路径)
注意:源文件可能不存在,需自行处理 FileNotFoundError
"""
target_dir = os.path.dirname(os.path.abspath(target_path))
os.makedirs(target_dir, exist_ok=True) # 确保目标目录存在
tmp_path = None
try:
with tempfile.NamedTemporaryFile(dir=target_dir, delete=False) as f:
tmp_path = f.name
processor(target_path, tmp_path) # 调用业务逻辑
os.replace(tmp_path, target_path)
except Exception as e:
if tmp_path and os.path.exists(tmp_path): os.remove(tmp_path)
raise e
def load_saved(file):
data = []
if os.path.exists(file):
for line in open(file, 'r', encoding='utf-8'):
if line.strip():
data.append(json.loads(line))
return data
'''
演示在线非商业网页自动化: https://quotes.toscrape.com/
进行登录;
获取Top Ten tags,并获取每个标签第一个帖子的内容,保存至xlsx文件。
'''
def main():
co = ChromiumOptions().set_browser_path(EDGE_PATH).ignore_certificate_errors()
tab = ChromiumPage(addr_or_opts=co, timeout = 5).get_tab()
tab.get('https://quotes.toscrape.com/')
if tab.s_ele('Logout', timeout = 1) is None: # 如果没有找到Logout元素,说明未登录
tab('Login').click()
tab('#username').input('admin')
tab('#password').input('passwd')
tab('@class:btn btn-primary').click()
eles = tab('@class:col-md-4 tags-box').eles('@class:tag-item')
logging.info("获取标签列表")
tags = [ele.text for ele in eles]
logging.info(f"获取到{len(tags)}个标签")
saved = load_saved(DATA_FILE)
saved_tags = {item['标签'] for item in saved} if saved else set()
tags = [tag for tag in tags if tag not in saved_tags]
logging.info(f"已保存可跳过: {len(saved_tags)}。待处理: {len(tags)}")
data_buffer: List[Dict] = []
try:
for tag in tags:
tab.get(f"https://quotes.toscrape.com/tag/{tag}/")
text = tab.s_ele('@class:quote').text
data_buffer.append({'标签': tag, '内容': text})
except Exception as e:
logging.error(f"操作标签时发生错误: {e}")
raise e
finally:
if data_buffer:
def writer(src, tmp):
combined = saved.copy() # 读旧
combined.extend(data_buffer) # 加新
with open(tmp, 'w', encoding='utf-8') as f:
for d in combined:
print(json.dumps(d, ensure_ascii=False), file=f)
safe_modify_file(DATA_FILE, writer)
logging.info(f"追加 {len(data_buffer)} 条数据到 {DATA_FILE}")
logging.info("所有标签操作完成,正在保存数据...")
# 将数据保存至xlsx文件
data = load_saved(DATA_FILE)
wb = Workbook()
ws = wb.active
headers = data[0].keys() if data else []
ws.append(list(headers))
# 3. 写入数据行
for row in data:
ws.append(list(row.values()))
wb.save(OUTPUT_FILE)
logging.info(f"数据已成功保存至{os.path.abspath(OUTPUT_FILE)}")
logging.info(f"全部操作完成")
if __name__ == "__main__":
try:
main()
except Exception as e:
logging.error(f"程序发生错误: {e}")
import traceback
traceback.print_exc()
3. 路径二:基于RPA软件的实施策略
尽管RPA软件(如影刀RPA、UiPath)常被视为“低代码工具”,其成功实施仍高度依赖工程化思维与精细化配置。以下为关键实施策略:
3.1 元素定位的稳健性设计
- 避免依赖屏幕坐标或固定XPath;
- 优先使用多属性组合选择器(如ID + class + visible text);
- 配置动态等待(“等待元素出现”而非
sleep); - 设置备用识别方案(如基于XPATH失败后启用OCR)。
3.2 流程模块化与参数化
- 将通用操作(如登录、文件上传)抽象为子流程;
- 通过输入参数驱动流程行为,实现复用;
- 利用Excel或数据库作为外部数据源,支持批量执行。
3.3 稳健控制与安全机制
- 在操作间插入随机延迟(如1.5–4秒),模拟人类行为;
- 控制请求频率,避免触发目标系统的风控策略;
- 在独立浏览器Profile或虚拟机中运行,实现环境隔离。
3.4 错误处理与审计合规
- 配置捕获异常;
- 失败时的自动日志;
- 启用操作日志,满足企业合规;
- 加密凭证库管理账号密码;
优势:开发效率高、内置监控与权限管理、支持业务人员参与。
挑战:定制能力受限、存在厂商锁定风险、黑盒调试困难。
3.4 示例
给出一个基于影刀RPA的流程示例,涉及常见库、安全文件操作、进度记录和恢复、日志和异常处理。
演示在线非商业网页自动化: https://quotes.toscrape.com/ 进行登录;获取Top Ten tags,并获取每个标签第一个帖子的内容,保存至xlsx文件。
4. 路径对比与选型建议
在实现示例功能时,笔者对此场景下两种方式的特点有明显感知。
基于RPA软件的方式在元素定位、普通的数据转换和读写、预先配置上有明显的效率,但异常处理(如下图,没有finally块,可能是版本限制) 以及自定义数据类型、复杂数据转换部分,使用图形化界面完成非常繁琐,并且遇到错误时的排查稍困难(少有完整的报错信息)。
基于代码的方式在编程环境安装、虚拟环境创建、代码调试方面需要花时间,元素定位也至少需要前端理论基础且稳定性需要调试,故而门槛高、时间长,但该示例 比较精简的实现了进度记录和恢复,相比RPA软件使用了两个文件来简化流程设计明显是高效的。而且它实现安全的文件操作,相比影刀的掌控性更强(并不确定影刀 对于文件操作是否是原子性的,因此无法确定系统故障后文件能否安全保留)。还有更关键的一点,基于代码的方式足够轻量化、可导出、AI扩展友好。
| 评估维度 | 基于代码的RPA | 基于RPA软件的RPA |
|---|---|---|
| 开发效率 | 中等 | 高(调试效率低) |
| 维护成本 | 高 | 中(平台托管) |
| 灵活性 | 完全 | 受限于平台功能 |
| 安全与合规 | 自主可控 | 依赖厂商安全模型 |
| 适用场景 | 复杂逻辑、系统集成、高性能 | 基础流程、部门协作 |
实施建议:
- 技术团队主导、流程复杂度高的场景:优先采用代码实现;
- 业务部门主导、流程标准化程度高的场景:选用成熟RPA平台;
- 混合模式:核心逻辑由代码实现,外围交互RPA软件调用,形成协同体系。
5. 结语:RPA的本质是流程工程
无论采用何种技术路径,RPA的成功实施始终依赖于对业务流程的理解、对异常场景的预判以及对合规审计的遵循。随着AI Agent的不断进步,自动化也将从“规则驱动”迈向“意图驱动”。但在此之前,夯实工程基础、建立健壮的实施规范,仍是实践者的首要任务。

