RPA:代码到低代码的实践
本文正在完善中
本文章给出基于代码的和基于RPA软件实现的RPA,在具体实施上的常见策略。本文不是技术介绍,涉及的技术是为解决问题为主,并不追求所谓“先进”或者“强大”。
提示词:我计划写一个RPA帖子,考虑给出基于代码的和基于RPA软件(如影刀)在具体实施上的常见策略。你需要列举在RPA具体实施步骤中常见的问题、改进方式、高级能力。我想到的基于代码的:成熟库、安全文件操作、健壮性如异常处理,重试和断点恢复、自动化流程的编程框架等。你需要基于想到的和我提出的点,进行合理补全精简和组织结构,给出良好的markdown文章结构。
从代码到低代码: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 示例
给出一个代码示例,涉及常见库、安全文件操作、进度记录和恢复、日志和异常处理。
点击展开 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')
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler())
DATA_FILE = "data.json"
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.unlink(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(r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe').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()
logging.info("登录成功,正在获取标签列表...")
eles = tab('@class:col-md-4 tags-box').eles('@class:tag-item')
tags = [ele.text for ele in eles]
logging.info(f"获取到{len(tags)}个标签")
all_data = load_saved(DATA_FILE)
saved_tags = {item['标签'] for item in all_data} if all_data else set()
tags = [tag for tag in tags if tag not in saved_tags]
logger.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}/")
logging.info(f"正在操作标签: {tag}")
text = tab.s_ele('@class:quote').text
data_buffer.append({'标签': tag, '内容': text})
logging.info("已获取内容。")
except Exception as e:
logging.error(f"操作标签时发生错误: {e}")
raise e
finally:
if data_buffer:
def writer(src, tmp):
combined = load_saved(src) # 读旧
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文件
all_data = load_saved(DATA_FILE)
wb = Workbook()
ws = wb.active
headers = all_data[0].keys() if all_data else []
ws.append(list(headers))
# 3. 写入数据行
for row in data_buffer:
ws.append(list(row.values()))
wb.save("帖子数据.xlsx")
logging.info("数据已成功保存至帖子数据.xlsx")
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); - 设置备用识别方案(如主路径失败后启用OCR或图像匹配)。
3.2 流程模块化与参数化
- 将通用操作(如登录、文件上传)抽象为子流程;
- 通过输入参数驱动流程行为,实现复用;
- 利用Excel或数据库作为外部数据源,支持批量执行。
3.3 稳健控制与安全机制
- 在操作间插入随机延迟(如1.5–4秒),模拟人类行为;
- 控制请求频率,避免触发目标系统的风控策略;
- 在独立浏览器Profile或虚拟机中运行,实现环境隔离。
3.4 错误处理与审计合规
- 配置Try-Catch块捕获异常;
- 自动保存失败时的截图与日志;
- 利用平台内置的加密凭证库管理账号密码,禁止明文存储;
- 启用操作审计日志,满足企业合规要求。
优势:开发效率高、内置监控与权限管理、支持业务人员参与。
挑战:定制能力受限、存在厂商锁定风险、黑盒调试困难。
4. 路径对比与选型建议
| 评估维度 | 基于代码的RPA | 基于RPA软件的RPA |
|---|---|---|
| 开发效率 | 低(需编码) | 高(可视化拖拽) |
| 维护成本 | 高(需DevOps支持) | 中(平台托管) |
| 灵活性 | 极高(可深度定制) | 受限于平台功能 |
| 安全与合规 | 自主可控 | 依赖厂商安全模型 |
| 适用场景 | 复杂逻辑、系统集成、高并发任务 | 标准化流程、跨部门协作 |
实施建议:
- 技术团队主导、流程复杂度高的场景:优先采用代码实现;
- 业务部门主导、流程标准化程度高的场景:选用成熟RPA平台;
- 混合模式:核心逻辑由代码实现,外围交互通过RPA软件调用,形成协同自动化体系。
5. 结语:RPA的本质是流程工程
无论采用何种技术路径,RPA的成功实施始终依赖于对业务流程的深刻理解、对异常场景的充分预判以及对安全合规的严格遵循。未来,随着AI Agent与RPA的深度融合(如通过大模型生成自动化脚本),自动化将从“规则驱动”迈向“意图驱动”。但在此之前,夯实工程基础、建立健壮的实施规范,仍是每一位自动化实践者的首要任务。
附录
- 示例代码:使用
Playwright实现安全登录与文件下载 - 影刀RPA流程图:展示模块化子流程设计
- 推荐工具链:
Pandoc(文档转换)、Mermaid(流程图绘制)、Vault(密钥管理)