ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

TagUI与Robocorp:从快速原型到企业级RPA机器人的完整实践指南

TagUI与Robocorp:从快速原型到企业级RPA机器人的完整实践指南

1. 从零开始:为什么选择TagUI与Robocorp这对组合?

如果你正在寻找一个既能快速上手、又具备企业级扩展潜力的RPA(机器人流程自动化)入门方案,那么TagUI和Robocorp的组合绝对值得你花时间研究。我最初接触RPA时,市面上工具繁多,从闭源的商业软件到各种开源框架,让人眼花缭乱。商业软件功能强大但价格不菲,且存在厂商锁定风险;而一些纯代码框架,虽然灵活,但学习曲线陡峭,对非开发背景的同事不够友好。

TagUI和Robocorp的出现,恰好在这两者之间找到了一个绝佳的平衡点。TagUI(现已更名为RPA for Python的一部分)以其极低的入门门槛和直观的“自然语言”脚本语法著称。你可以用近乎口语化的英语(如click ‘login_button’)来编写自动化脚本,这让业务分析师或初涉自动化的朋友也能快速构建原型。而Robocorp则提供了一个基于Python的、更加强大和结构化的开发与运行环境,它集成了机器人开发、依赖管理、日志、错误处理以及云端部署等一系列生产级功能。

简单来说,TagUI是你的“快速原型画笔”,而Robocorp是你的“工业化生产车间”。用TagUI快速验证一个流程的可行性,捕捉网页元素,理清操作步骤;然后用Robocorp的框架将其重构,加入健壮的错误处理、清晰的任务结构、便于维护的配置管理,最终打包成一个可以在任何地方(本地、虚拟机、容器、云端)稳定运行的机器人。这个组合覆盖了从创意验证到生产部署的全链路,对于想深入RPA领域的开发者而言,是一个性价比极高的技术栈。

2. 环境搭建与核心工具链解析

在动手写代码之前,我们需要把“画板”和“车间”都准备好。这个环节的配置直接决定了后续开发的顺畅度,很多初学者遇到的“莫名其妙”的错误,往往都源于环境问题。

2.1 Python环境与Robocorp CLI的安装

Robocorp的核心是Python。首先,我强烈建议使用condavenv创建一个独立的Python虚拟环境。这能避免与你系统上其他项目的依赖发生冲突。这里以conda为例(如果你没有安装conda,可以直接使用Python自带的venv模块)。

# 创建一个名为rpa-demo的Python 3.9环境 conda create -n rpa-demo python=3.9 conda activate rpa-demo

接下来,安装Robocorp的命令行工具robocorp。这个工具是管理Robocorp项目的瑞士军刀。

pip install robocorp

安装完成后,运行robocorp -h验证是否成功。这个CLI工具将用于创建项目模板、运行机器人、管理依赖等。

2.2 创建你的第一个Robocorp机器人项目

Robocorp通过项目模板来规范结构,这能保证最佳实践。我们创建一个标准的机器人项目:

# 创建一个名为 `my_first_rpa` 的目录并初始化项目 robocorp init my_first_rpa

执行命令后,CLI会引导你选择模板。对于纯RPA任务,选择“Standard Robot”模板即可。完成后,进入项目目录,你会看到一个结构清晰的项目文件夹:

my_first_rpa/ ├── robot.yaml # 机器人配置文件,定义依赖、环境变量等 ├── conda.yaml # Conda环境依赖文件(Robocorp也支持) ├── tasks.py # 任务入口文件,定义机器人执行的工作流 ├── tasks.robot # (可选)Robot Framework格式的任务文件 ├── devdata/ # 开发环境数据 ├── output/ # 机器人运行输出(日志、截图等) └── src/ # 主要源代码目录 └── my_first_rpa/ # 你的Python包 ├── __init__.py └── tasks.py # 具体的任务实现(与根目录tasks.py可能关联)

这个结构是Robocorp的精华之一。robot.yaml是机器人的“身份证”和“说明书”,它声明了机器人需要什么环境、依赖哪些库、以及如何运行。我们稍后会修改它来引入TagUI。

2.3 集成TagUI(RPA for Python)

TagUI的核心能力已经集成到了rpaframework这个库中,具体是RPA.Browser.SeleniumRPA.Desktop等库。但为了使用TagUI原生的简洁语法,我们需要安装tagui包。不过,更现代且与Robocorp生态融合更好的方式是直接使用rpaframework

修改robot.yaml文件,在dependencies部分加入RPA框架的依赖:

dependencies: - rpaframework>=24.0.0 # 这是一个RPA工具集合,包含了基于Selenium的浏览器自动化等 - selenium>=4.0.0 # 明确的Selenium版本,确保兼容性

然后,在项目根目录下运行以下命令,让Robocorp根据配置文件安装所有依赖:

robocorp install

这个命令会读取robot.yamlconda.yaml,为你创建一个隔离的、包含所有指定依赖的运行时环境。至此,你的“车间”里已经备好了所有核心工具。

注意:网络环境可能会导致selenium下载浏览器驱动(如chromedriver)失败。rpaframeworkRPA.Browser.Selenium库通常会尝试自动下载匹配的驱动,但如果遇到问题,你可能需要手动下载对应版本的 ChromeDriver 并放置到系统PATH或项目指定目录下。这是RPA浏览器自动化中一个经典的“坑”。

3. 使用TagUI风格语法编写第一个自动化脚本

环境就绪,让我们先用TagUI直观的语法来感受一下快速原型的魅力。我们假设一个经典场景:打开浏览器,访问GitHub,搜索Robocorp仓库,并获取第一个结果的描述信息

src/my_first_rpa/目录下,我们创建一个新文件github_search.py。这里,我们将使用rpaframework中类似TagUI的简洁API风格。实际上,RPA.Browser.Selenium库的设计哲学与TagUI一脉相承,力求简化操作。

from RPA.Browser.Selenium import Selenium # 初始化浏览器库 browser_lib = Selenium() def open_github_and_search(): # 1. 打开浏览器并导航至GitHub browser_lib.open_available_browser("https://github.com") # 2. 定位搜索框并输入关键词 - 使用TagUI式的“自然”定位思路 # 我们可以用CSS选择器、XPath,或者更简单的:包含特定文本或属性的元素 search_selector = "input[placeholder='Search GitHub']" browser_lib.input_text(search_selector, "Robocorp") browser_lib.press_keys(search_selector, "ENTER") # 3. 等待结果加载 browser_lib.wait_until_element_is_visible("css:div.codesearch-results", timeout=10) # 4. 获取第一个仓库的链接和描述 # 定位第一个仓库条目。这里使用一个更具体的CSS选择器,避免抓到其他元素。 first_repo_selector = "css:div.codesearch-results ul.repo-list li:first-child a.v-align-middle" # 检查元素是否存在 if browser_lib.is_element_visible(first_repo_selector): repo_name = browser_lib.get_text(first_repo_selector) print(f"第一个仓库名称: {repo_name}") # 获取仓库描述(可能不存在,需要处理) # 描述通常在仓库名称后面的p标签里,但DOM结构可能变化,这里是一个示例选择器 desc_selector = f"xpath://a[contains(text(), '{repo_name}')]/../../p" if browser_lib.is_element_visible(desc_selector): repo_description = browser_lib.get_text(desc_selector) print(f"仓库描述: {repo_description}") else: print("未找到仓库描述。") else: print("未找到搜索结果。") # 5. 为了演示,等待几秒后关闭浏览器 browser_lib.sleep(3) browser_lib.close_all_browsers() if __name__ == "__main__": open_github_and_search()

这段代码充分体现了TagUI的核心思想:将自动化指令写得像简单的英语句子open_available_browser,input_text,press_keys,wait_until_element_is_visible,get_text,这些方法名几乎是不言自明的。即使你不懂复杂的CSS选择器,通过浏览器的开发者工具复制元素选择器,也能快速上手。

实操心得:网页元素定位是RPA中最不稳定的一环。网站前端稍作改动,你的脚本就可能失效。因此,不要使用绝对路径的XPath或依赖固定索引的CSS选择器。优先使用具有稳定idname属性的元素,其次是用包含关键字的placeholderaria-label或相对稳定的CSS类。在RPA.Browser.Selenium中,wait_until_element_is_visible是你的好朋友,它能有效避免因网络延迟或动态加载导致的元素找不到错误。

4. 将脚本升级为Robocorp机器人:结构化与健壮性

刚才的脚本是一个简单的原型,但它缺乏错误处理、配置管理和可调度性。现在,我们把它改造为一个真正的Robocorp机器人。这涉及到利用Robocorp框架的几个核心特性:任务定义、上下文管理、日志和错误处理

4.1 定义机器人任务 (tasks.py)

首先,我们修改项目根目录下的tasks.py,这是机器人的主入口。我们将定义一个名为GitHub Search Example的任务。

from robocorp.tasks import task from robocorp import log from my_first_rpa.github_search import open_github_and_search @task def github_search_example(): """ 一个示例任务:打开GitHub,搜索‘Robocorp’并打印第一个结果。 此任务展示了如何将简单的TagUI风格脚本集成到Robocorp框架中。 """ log.info("开始执行GitHub搜索示例任务。") try: # 调用我们之前写好的函数 open_github_and_search() log.info("任务执行成功!") except Exception as e: # Robocorp框架会自动捕获异常并记录,但我们可以进行自定义处理 log.error(f"任务执行失败: {e}") # 可以在这里添加截图、发送通知等清理或上报操作 raise # 重新抛出异常,让框架知道任务失败

@task装饰器是Robocorp框架的关键。它告诉框架,这是一个可执行的任务。框架会为这个任务的执行自动配置日志、管理生命周期(如开始、结束、清理)。robocorp.log模块提供了结构化的日志输出,比简单的print更强大,日志会自动输出到output/log.html等文件中,便于事后审计。

4.2 增强的脚本实现:集成配置与资源管理

现在,我们回头重构src/my_first_rpa/github_search.py,使其更健壮、更可配置。

from RPA.Browser.Selenium import Selenium from robocorp import log, workitems from robocorp.tasks import get_output_dir import os from datetime import datetime class GitHubSearchRobot: def __init__(self): self.browser_lib = Selenium() # 可以从环境变量、work item或配置文件中读取搜索关键词 self.search_keyword = "Robocorp" self.output_dir = get_output_dir() def open_github_and_search(self): """主业务流程""" log.info(f"使用关键词 '{self.search_keyword}' 开始搜索。") try: # 1. 打开浏览器 - 可配置headless模式 headless = os.getenv("RPA_HEADLESS_MODE", "False").lower() == "true" browser_args = ["--start-maximized"] if headless: browser_args.append("--headless=new") self.browser_lib.open_available_browser( "https://github.com", browser_selection="chrome", arguments=browser_args ) # 2. 执行搜索 search_selector = "input[placeholder='Search GitHub']" self.browser_lib.input_text(search_selector, self.search_keyword) self.browser_lib.press_keys(search_selector, "ENTER") # 3. 等待并验证结果页 results_locator = "css:div.codesearch-results" self.browser_lib.wait_until_element_is_visible(results_locator, timeout=15) # 4. 捕获结果并记录 self._capture_search_results() # 5. 在出错或成功时都截图(用于审计和调试) screenshot_path = os.path.join(self.output_dir, f"final_state_{datetime.now().strftime('%H%M%S')}.png") self.browser_lib.screenshot(filename=screenshot_path) log.info(f"最终状态截图已保存至: {screenshot_path}") except Exception as e: # 发生异常时截图 error_screenshot = os.path.join(self.output_dir, f"error_{datetime.now().strftime('%H%M%S')}.png") self.browser_lib.screenshot(filename=error_screenshot) log.error(f"流程执行出错,错误截图已保存。错误信息: {e}") raise # 将异常向上传递 finally: # 无论成功与否,确保关闭浏览器 self.browser_lib.close_all_browsers() log.info("浏览器已关闭。") def _capture_search_results(self): """提取并记录搜索结果""" first_repo_selector = "css:div.codesearch-results ul.repo-list li:first-child a.v-align-middle" if not self.browser_lib.is_element_visible(first_repo_selector, timeout=5): log.warn("未在页面上找到预期的第一个仓库元素。") # 可以尝试备用选择器或记录整个页面HTML用于分析 return repo_name = self.browser_lib.get_text(first_repo_selector) log.info(f"找到的第一个仓库: {repo_name}") # 尝试获取描述 - 使用更灵活的XPath,基于找到的仓库名元素 repo_link = self.browser_lib.get_element_attribute(first_repo_selector, "href") # 构建一个相对可靠的描述定位器:在同一列表项(li)内寻找p标签 desc_selector = f"xpath://a[@href='{repo_link}']/ancestor::li//p" if self.browser_lib.is_element_visible(desc_selector, timeout=2): repo_description = self.browser_lib.get_text(desc_selector) log.info(f"仓库描述: {repo_description}") # 可以将结果存储为结构化的数据,例如写入JSON文件或发送到下一个流程 result_data = { "keyword": self.search_keyword, "first_repo_name": repo_name, "first_repo_description": repo_description, "timestamp": datetime.now().isoformat() } # 这里可以调用 workitems.output.create() 将结果输出给下一个机器人 log.info(f"提取的结果数据: {result_data}") else: log.info(f"仓库 '{repo_name}' 没有描述文本。")

这个升级版的类GitHubSearchRobot体现了生产级机器人的几个关键点:

  1. 配置化:搜索关键词可以从环境变量 (RPA_HEADLESS_MODE) 或未来的work items中读取,而不是硬编码。
  2. 健壮的错误处理:使用try...except...finally结构确保浏览器资源在任何情况下都能被正确关闭。发生错误时自动截图,极大方便了远程调试。
  3. 结构化日志:使用log.info/log.warn/log.error替代print,日志会自动分级、着色,并输出到HTML报告。
  4. 资源管理:使用get_output_dir()获取框架管理的输出目录,用于存放截图和日志,避免路径混乱。
  5. 可扩展性:将核心业务逻辑_capture_search_results封装成独立方法,便于未来修改或复用。结果数据被组织成字典,为后续的数据持久化或流程衔接做好准备。

4.3 运行与调试你的机器人

现在,我们可以用Robocorp CLI来运行这个机器人了。在项目根目录下,执行:

robocorp run

默认情况下,CLI会查找tasks.py中的@task装饰函数并运行它们。你会看到控制台输出结构化的日志,同时所有日志和截图都会保存在output/目录下。运行结束后,打开output/log.html,你会得到一个格式美观、包含所有步骤详情和时间戳的HTML日志报告,这对于审计和复盘至关重要。

如果你想在无头模式下运行(不显示浏览器界面),以节省资源或在服务器上运行,可以设置环境变量:

# 在Linux/macOS上 export RPA_HEADLESS_MODE=True robocorp run # 在Windows PowerShell上 $env:RPA_HEADLESS_MODE="True" robocorp run

5. 进阶:处理更复杂的交互与数据

一个真实的RPA流程很少只是简单的搜索和抓取。通常涉及登录、表单填写、文件下载、数据处理等。让我们扩展一下,模拟一个需要登录(使用假想网站)并导出数据的场景,展示TagUI/Robocorp如何处理更复杂的交互。

假设我们需要登录一个内部仪表盘,筛选某些数据,然后点击“导出为CSV”按钮。这里的关键挑战是等待动态加载的元素处理文件下载

我们在src/my_first_rpa/下创建另一个文件dashboard_automation.py

from RPA.Browser.Selenium import Selenium from RPA.HTTP import HTTP from robocorp import log, workitems import os import time from pathlib import Path class DashboardAutomation: def __init__(self, username=None, password=None): self.browser = Selenium() self.http = HTTP() self.username = username or os.getenv("DASHBOARD_USERNAME") self.password = password or os.getenv("DASHBOARD_PASSWORD") if not self.username or not self.password: log.error("未找到仪表盘登录凭据。请设置DASHBOARD_USERNAME和DASHBOARD_PASSWORD环境变量。") raise ValueError("Missing credentials") self.download_dir = str(Path(os.getenv("ROBOT_ROOT", os.getcwd())) / "output" / "downloads") os.makedirs(self.download_dir, exist_ok=True) def automate_dashboard_export(self): """自动化登录仪表盘并导出数据""" log.info("开始仪表盘自动化流程。") try: # 1. 配置浏览器以下载文件到指定目录 prefs = { "download.default_directory": self.download_dir, "download.prompt_for_download": False, "plugins.always_open_pdf_externally": True } options = { "prefs": prefs, "arguments": ["--start-maximized"] } self.browser.open_available_browser( "https://example-internal-dashboard.com/login", # 示例URL browser_selection="chrome", options=options ) # 2. 登录 self._perform_login() # 3. 导航到报表页面并筛选 self._navigate_and_filter() # 4. 触发导出并等待下载完成 exported_file = self._trigger_export_and_wait() if exported_file: log.info(f"文件下载成功: {exported_file}") # 此处可以添加处理文件的逻辑,例如用RPA.FileSystem或pandas读取CSV # from RPA.FileSystem import FileSystem # lib = FileSystem() # content = lib.read_file(exported_file) else: log.warn("未检测到文件下载。") except Exception as e: log.error(f"仪表盘自动化失败: {e}") raise finally: self.browser.close_all_browsers() def _perform_login(self): """处理登录逻辑""" log.info("正在登录...") # 假设登录表单有id为‘username’和‘password’的字段 self.browser.input_text("id:username", self.username) self.browser.input_text("id:password", self.password) self.browser.click_button("css:button[type='submit']") # 等待登录成功后的页面元素出现,例如用户头像或仪表盘标题 self.browser.wait_until_element_is_visible("css:.user-avatar", timeout=10) log.info("登录成功。") def _navigate_and_filter(self): """导航到目标页面并应用筛选器""" log.info("导航至数据报表页面。") # 点击侧边栏菜单 self.browser.click_element("css:a[href='/reports']") # 等待报表页面加载 self.browser.wait_until_page_contains_element("h1:Reports", timeout=10) # 应用日期筛选器(示例:选择‘最近7天’) log.info("应用‘最近7天’筛选器。") self.browser.click_element("css:#date-filter-dropdown") self.browser.click_element("xpath://li[contains(text(), 'Last 7 days')]") # 等待数据刷新。这是一个关键点:需要等待加载动画消失或数据表格更新。 # 假设有一个加载指示器 loading_indicator = "css:.data-loading-spinner" self.browser.wait_until_element_is_not_visible(loading_indicator, timeout=30) log.info("数据筛选完成。") def _trigger_export_and_wait(self, timeout=60): """点击导出按钮并等待文件下载完成""" log.info("触发CSV导出。") export_button = "css:button.export-csv" self.browser.click_element(export_button) # 方法:轮询下载目录,检查是否有新的.csv文件出现 log.info(f"等待文件下载到: {self.download_dir}") initial_files = set(os.listdir(self.download_dir)) start_time = time.time() while (time.time() - start_time) < timeout: time.sleep(2) # 每2秒检查一次 current_files = set(os.listdir(self.download_dir)) new_files = current_files - initial_files csv_files = [f for f in new_files if f.lower().endswith('.csv')] if csv_files: # 通常最新的文件就是我们刚下载的 downloaded_file = sorted(csv_files, key=lambda f: os.path.getmtime(os.path.join(self.download_dir, f)))[-1] full_path = os.path.join(self.download_dir, downloaded_file) # 额外检查文件是否已下载完全(没有.crdownload临时文件) if not downloaded_file.endswith('.crdownload'): log.info(f"发现新CSV文件: {downloaded_file}") return full_path # 如果发现临时文件,说明下载还在进行中 crdownload_files = [f for f in new_files if f.endswith('.crdownload')] if crdownload_files: log.debug("下载仍在进行中...") log.error(f"在{timeout}秒内未检测到完成的CSV文件下载。") return None

这个进阶示例揭示了几个在真实RPA项目中必然会遇到的痛点及其解决方案:

  1. 凭据管理:绝对不要将用户名密码硬编码在脚本中。我们通过环境变量 (DASHBOARD_USERNAME,DASHBOARD_PASSWORD) 来注入敏感信息。Robocorp Cloud等平台提供了更安全的Vault来存储机密。
  2. 文件下载:处理浏览器下载是RPA的常见需求。我们通过Chrome选项预设下载目录并禁用下载提示。最关键的是_trigger_export_and_wait方法中的轮询逻辑,它主动检查目标文件夹,等待预期的文件出现并确认其下载完成(通过检查.crdownload临时文件是否消失)。这是一种可靠且跨平台的方法。
  3. 等待策略:对于动态加载的页面,简单的sleep是脆弱且低效的。我们组合使用了wait_until_element_is_visiblewait_until_element_is_not_visible。等待加载动画消失是判断页面“就绪”状态的黄金法则。
  4. 模块化设计:将登录、导航、导出等步骤拆分为私有方法 (_perform_login),使得主流程automate_dashboard_export清晰易懂,也便于单独测试和维护。

6. 项目打包、部署与流程编排初探

开发调试完成后,你可能需要将机器人部署到服务器或Robocorp Cloud上定期执行。Robocorp提供了完整的工具链。

6.1 本地打包与运行

首先,确保你的robot.yaml配置完整。一个更完整的示例如下:

tasks: GitHub 搜索示例: shell: python -m my_first_rpa.github_search 仪表盘自动化: shell: python -m my_first_rpa.dashboard_automation environmentConfigs: - environment_windows_amd64_freeze.yaml - environment_linux_amd64_freeze.yaml artifactsDir: output ignoreFiles: - .gitignore - .git/**

你可以通过CLI运行特定任务:

robocorp run -t “GitHub 搜索示例”

为了创建可以在其他没有Python环境的机器上运行的包,可以使用robocorp的打包功能(通常与Robocorp Cloud结合更紧密)。但本地也可以创建conda环境包。

6.2 使用Robocorp Work Items进行数据驱动

真正的生产流程往往是数据驱动的。例如,从一个队列中读取需要搜索的关键词列表,然后逐个执行。Robocorp的Work Items机制就是为此而生。

你可以修改tasks.py,使其从输入Work Item中读取数据:

from robocorp.tasks import task from robocorp import workitems from my_first_rpa.github_search import GitHubSearchRobot @task def process_search_keywords(): """从输入Work Item中读取关键词并执行搜索""" for item in workitems.inputs(): keyword = item.payload.get("search_keyword", "Robocorp") log.info(f"处理关键词: {keyword}") robot = GitHubSearchRobot() robot.search_keyword = keyword # 动态传入关键词 try: robot.open_github_and_search() # 可以将结果保存到输出Work Item,供下一个机器人使用 # result_payload = {"keyword": keyword, "status": "success"} # workitems.outputs.create(payload=result_payload) item.done() except Exception as e: log.error(f"处理关键词 '{keyword}' 时失败: {e}") item.fail()

然后,你可以通过CLI或Robocorp Control Room(云端管理平台)向这个机器人发送包含{"search_keyword": "某关键词"}的Work Item,实现流程的编排和复用。

6.3 部署到Robocorp Cloud(可选)

如果你需要调度、监控和集中管理机器人,可以将项目推送到Robocorp Cloud。

  1. 在 Robocorp Portal 上创建一个机器人工作区(Workspace)。
  2. 使用robocorpCLI 登录:robocorp cloud login
  3. 将你的机器人上传(Push)到云端:robocorp cloud push
  4. 在Portal的Control Room中,你可以配置机器人的运行环境(如Windows/Linux)、设置定时触发器(Cron表达式)、管理密钥、查看详细的运行日志和报告。

这种模式将本地开发的灵活性与云端的运维能力结合了起来,非常适合团队协作和商业项目。

从在本地用TagUI风格的语法快速验证一个想法,到用Robocorp框架将其重构为结构清晰、错误处理完善、配置灵活的模块化脚本,再到考虑数据驱动、打包部署和云端协作,这条路径清晰地展示了一个RPA爱好者如何一步步成长为能够交付生产级自动化解决方案的开发者。这个组合的魅力在于,它既照顾了入门时的简单直观,又为你铺好了通往复杂、健壮企业级应用的道路。

返回列表