给程序员用的RPA打包发布工具

为什么有这个

1. 现在市场上有很多的RPA工具,UIPATH, UIBOT. 阿里云,华为云也都提供了各自的RPA开发套件. 但是对程序员来说,这些层层包装过的RPA工具用起来并不利落,总不如直接写代码操作来的舒服. 最重要这些工具的bot执行环境都是收费产品.
2. 程序员可以通过使用python, selenium,pywin32等等这些库和框架轻松的实现各种RPA流程,但是如何发布给RPA工具的最终用户,并没有一个简单的方案

提供了什么

1. 一个简单的python脚本加载和运行和打包环境, 直接打包成exe发布
2. 一个简单通用的gui界面, 只需要在python脚本中声明

怎么用

1. 在scripts目录下新建脚本,比如

    hello_world.py

2. 实现脚本

   from runtime.flow import ScriptDef, FileItem
   from lib.log import logger
   from lib import excel
   from runtime.m import retry
   
   def reset():
       logger.info("RESET")
   
   @retry(num_times=3, sleep_time=1, reset_func=reset)
   def print_item(item):
       logger.info(item)
       raise Exception("111")
   
   def run(excel_file):
       data = excel.loads2dict(excel_file)
       for item in data:
           print_item(item)
           
   export = ScriptDef(
               func=run,
               group="其他",
               title="测试流程EXCEL",
               arguments=[
                   FileItem(title="excel文件", name="excel_file")
               ]   
           )

3. 基于class的用法:

   from lib.log import logger
   from runtime.flow import ScriptDef, StringItem
   
   class SayHello(object):
       def __init__(self, name):
           super().__init__()
           self.name = name
   
       def run(self):
           logger.info(f"Hello {self.name}!")
   
   export = ScriptDef(
               cls=SayHello,
               group="DEMO",
               title="Hello world!",
               arguments=[
                   StringItem(title="姓名", name="name"),
               ]   
           ) 

4. 启动测试

    python launch.py 

详细说明

GUI实现方案

GUI界面用VUE开实现,代码在 gui 目录下, 用yarn 打包管理

cd gui/rpa-bot
yarn dev

使用pywebview来加载展示

Flow类说明

流程脚本只需要有一个run的方法定义,就可以被加载执行,但是建议使用内置的Flow类,定义了一下几个方法

class Flow(object):

    def precheck(self):  
        # 流程实际执行前做一下参数和数据的检查工作,对定时执行的任务比较重要,可以提前发现输入数据中可能的错误
        pass

    def prefun(self):
        # 在执行run之前做的一些操作
        pass

    def run(self):
        # 流程的执行代码
        pass

    def afterfun(self):
        # 在执行run之前做的一些操作
        pass

在实际的使用中, 很多的流程是加载一张excel表格, 然后一行一行来执行操作. 针对这种类型的任务,提供了ListDataFlow和MultDataFlow这两个基类, 实现了基础的任务调度异常报错和重试机制,所有类似的流程可以直接继承ListDataFlow类, 例如下面这个批量发送邮件的流程

class Mail(object):
    mail: str # 邮箱
    name: str # 姓名
    text: str # 发送内容
    def __str__(self):
        return f"{self.mail}"
    
class SendMail(ListDataFlow):
    def __init__(self, excel_file):
        self.excel_file = excel_file
    def load_list(self):
        # 这里负责加载要处理的任务数据,一般是加载表格, 返回的Item类的定义注意一样要实现自己的__str__方法, 
        # 返回可以明确描述任务的信息
        # 上层的框架代码在记录出错记录时直接str(item)的
        return excel.load(self.excel_file, Item, {})
    def run_job(self, item):
        # 这里是具体的每一条任务的处理流程,有异常时直接抛出,上层的代码会捕获并记录出错记录
        # send_mail(item.mail, item.text)
    def reset(self):
        # 这里负责在异常的时候重置状态,比如重新启动浏览器,重新登录等, 确保下一条数据还可以正常执行
        pass

GUI表单

目前提供的表单项有以下几个:

1. StringItem

   基础的文本输入框

   StringItem(title="姓名", name="name")

2. NumberItem

   数字输入框

   NumberItem(title="数字", name="number")

3. FileItem

   文件选择框

   FileItem(title="excel表格", name="excel_file"),

4. DirectoryItem

   目录选择

   DirectoryItem(title="文件夹", name="image_dir"),

5. SelectItem

   下拉选择

   SelectItem(title="目标网址", name="url", options=[ "http://baidu.com", "http://google.com" ]),

6. UserSelectItem

   账号选择,注意目前账号密码文件默认读取的是C:\RPA\账号.xlsx中的配置

   UserSelectItem(title="账号", username_field="name", password_field="pwd"),

编译打包

1. 所有脚本中会加载的依赖包,需要在runtime\load.py文件中申明导入以下,这样pyinstaller打包的时候会打进去,已申明的常用依赖如下:

   import win32
   import jinja2
   import requests
   import openpyxl
   import xlrd
   import selenium
   import fitz
   import cv2
   import numpy
   import PIL
   import imutils
   import docxtpl
   import pdfplumber
   import Crypto
   import pyperclip
   import pynput

2. 要发布哪些脚本出去,可以在build.py文件中修改配置:

   ENV_CONFIG = {
       "default": ["hello_world.py"],
   }

   然后执行

   ​ python build.py default

3. 一般来说,引用的依赖没有变化,不需要重新打包发布exe,只需要更新个发布scripts目录下的脚本即可

几个基础库

LOG库使用

from lib.log import logger
logger.info("msg")
logger.error("msg", stack_info=True)

进度保存和错误保存

from lib import snapshot
snapshot.save_check_point("taskname", "progress")
snapshot.save_error("taskname", ["e1", "e2"])

excel快速读取

from lib import excel

class Item(object):
    idcard:str
    username:str
    time_number:str
    lian_approver:str

items = excel.loads("文件.xlsx", Item, {
    "idcard": {"title": "被申请人证件号", "required": True},  
    "username": {"title": "被申请人姓名", "required": True},
    "time_number": {"title": "缴费日期", "required": True},  
    "lian_approver": {"title": "立案审批人", "required": True}
})