同步操作将从 thor.lee/rpa 强制同步,此操作会覆盖自 Fork 仓库以来所做的任何修改,且无法恢复!!!
确定后同步将在后台操作,完成时将刷新页面,请耐心等待。
在scripts目录下新建脚本,比如
hello_world.py
实现脚本
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")
]
)
基于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"),
]
)
启动测试
python launch.py
GUI界面用VUE开实现,代码在 gui 目录下, 用yarn 打包管理
cd gui/rpa-bot
yarn dev
使用pywebview来加载展示
流程脚本只需要有一个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
目前提供的表单项有以下几个:
StringItem
基础的文本输入框
StringItem(title="姓名", name="name")
NumberItem
数字输入框
NumberItem(title="数字", name="number")
FileItem
文件选择框
FileItem(title="excel表格", name="excel_file"),
DirectoryItem
目录选择
DirectoryItem(title="文件夹", name="image_dir"),
SelectItem
下拉选择
SelectItem(title="目标网址", name="url", options=[ "http://baidu.com", "http://google.com" ]),
UserSelectItem
账号选择,注意目前账号密码文件默认读取的是C:\RPA\账号.xlsx中的配置
UserSelectItem(title="账号", username_field="name", password_field="pwd"),
所有脚本中会加载的依赖包,需要在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
要发布哪些脚本出去,可以在build.py文件中修改配置:
ENV_CONFIG = {
"default": ["hello_world.py"],
}
然后执行
python build.py default
一般来说,引用的依赖没有变化,不需要重新打包发布exe,只需要更新个发布scripts目录下的脚本即可
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"])
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}
})
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。