「初学者向」使用Python处理重复工作,解放工作量【上】

最近需要在工作中重复劳动一些耗时且细碎的工作,写了个脚本解放工作时间,记录下整个流程。

# // 1. 起因

最近需要在公司内部系统上重复提交几百个任务,这项工作其实并不复杂,问题在于每执行一次操作需要耗时2分钟,几百个任务就是一下午的时间就没了,而且还容易出错,需要把这部分工作量解放出来,让python替我完成。

# // 2. 任务概述

我司内部系统是基于web的,我现在需要在gui上操作去添加任务,这个任务是从一个内部ftp上的,以日期命名的目录中取该目录下的所有文件,那么我在web gui上就需要填写:

  • 任务标题
  • 任务描述
  • ftp信息(地址,端口,类型,账号,密码)
  • 路径
    添加完成任务之后,我需要每2分钟启动一个任务,因为如果一次性开启所有任务,后端可能会崩。

所以我需要实现的功能如下:

  • 在进入程序后输入列表ID和token,可以自动获取所需要的json;
  • 之后循环输入任务名称和ftp路径,即可自动添加任务;
  • 所有任务添加完成之后,每2分钟自动启动一个任务。

# // 3. 观察

浏览器中打开控制台的网络监控,然后尝试添加一个任务,会看到以下行为:

  1. 打开添加任务页,会先看到向某个url发送了get请求,url的最后一个参数是该任务列表的列表ID,得到一个json,该json包含该任务列表的信息和列表下所有任务的信息;
  2. 填写完成信息之后点击“提交”,看到向上述url发送了个put请求,请求体是个json。

把上述的两个json放到json compare工具中看下,发现第2步中put过去的json,就是第1步的json的缩略版,但是包含了个新的object,这个object中的内容就是填写的任务标题、ID等等信息。

然后启动个任务看看:

  1. 点击“开始任务”,发现向某个url发送了post请求,url中参数包含该任务ID。

验证方式上,咨询我司工程师之后得知,只需要在请求头中包含token即可。
那流程和内容就很清晰了,接下来用代码实现就行了。

# // 4. 实现过程 - 任务批量添加器

由于是把所有任务添加完成之后再一个个启动任务,所以脚本分两个写。
本篇先写任务添加脚本。

# A. 规划

处理网络请求那肯定就是用requests库没跑了,而且需要处理json,那也先把Json库引入进来。先把程序结构写出来:

1
2
3
4
5
import requests
import json

if __name__ == "__main__":
# 主体结构在这里

先规划一下流程和函数:

  1. 进入程序之后先手动输入token和列表ID,因为这两个东西是在get列表的时候需要的内容,一次输入运行时可用,后续不用重复输入;
  2. 用上面的信息去get列表,如果成功的话:
  3. 删除json中不必要的信息;
  4. 手动输入任务名称和ftp路径;
  5. 把3和4的内容传给任务添加函数;
  6. 重复3 4 5。

# B. 列表获取函数

然后来实现第1步,先get任务列表,写一个函数:

1
2
3
4
5
def listgetter(auth, requrl):
headers = {"Content-Type":"application/json","Authorization":auth}
# 实例url为:https://example.com/api/data_sources/task_list_id
r = requests.get(requrl, headers=headers)
return r.text

这个函数有三个作用:

  1. 会接收两个参数,这两个参数需要用户(也就是我)在进入程序之后手动输入,一个是token,一个是请求地址,但是由于请求地址变化的只是最后一个url中的参数,所以其实只需要手动输入这个参数就行,然后在主流程中拼接一个url出来,作为参数给到这个函数;
  2. 然后用requests带headers请求一个get,具体文档可以看 这里
  3. 把get的结果返回出来。

# C. json处理函数

获取到json之后,也就是收到listgetter()函数的返回结果之后, 要对结果进行裁切,那再写一个函数进行json裁切:

1
2
3
4
5
6
7
8
9
10
11
def tailor(content):
gd = json.loads(content)
del gd['created_at']
del gd['user_id']
del gd['group']
del gd['group_id']
del gd['logo']
del gd['name']
del gd['updated_at']
del gd['desc']
return gd

这个函数有四个作用:

  1. 接受get到的json,此时为一个字符串;
  2. 用json.loads方法将json字符串解析为dict;
  3. 删除不需要的对象;
  4. 返回删除后的dict。

# D. 任务json生成函数

获取到处理过后的json dict之后,就需要在指定的对象中假如我们需要提交的任务的信息,这样添加完的最终结果就是需要在最后一步put的json。
裁剪过后的json大概是这样:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
"elementA":"elementa",
"elementB":"elementb",
"tasks":{
"task1ID":{
"id": "任务ID",
"desc": "",
"host": "ip地址",
"port": "1024",
"status": "",
"password": "密码",
"taskname": "任务名",
"username": "登陆用户名",
"createdAt": "日期",
"updatedAt": "日期",
"remotepath": "ftp路径",
"compressionmode": "targz"}
}
}

其中tasks中的每一个object即是一个任务,所以就需要在tasks中添加object,之后put完整的json即可添加一个任务。
那么函数如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
def idgen():
rdm = ''.join(random.sample(string.ascii_uppercase + string.digits, 24))
return "AA" + rdm

def taskadder(content, taskname): # 1
newid = idgen() # 2
created_at = time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.localtime())
remote_path = "/path/to/" + taskname + "/*.tar.gz"

newtask = {'id': newid, 'desc': '', 'host': 'ip地址', 'port': '1024', 'status': '', 'password': '密码', 'taskname': taskname, 'username': '登陆用户名', 'createdAt': created_at, 'updatedAt': created_at, 'remotepath': [remote_path], 'compressionmode': 'targz'} # 3

content['tasks'][newid] = newtask #4
return content #5

解释:

  1. 这个函数接收两个参数,
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    2. 上面的json中的“任务ID”,在咨询工程师之后告知其实是生成的随机字符串,所以就写个函数用来生成字符串。为了和其他任务区分,使用AA开头表示是我使用脚本生成的任务;
    3. 这一串就是实际新增任务,等会需要加到json里面的object;
    4. 往这个dict里面插入这个新任务;
    5. 返回这个dict。

    ## E. 任务提交函数
    最后一个函数就是提交了:
    ```python
    def taskputter(content, auth, requrl):
    headers = {"Content-Type":"application/json","Authorization":"token"}
    # 实例url为:https://example.com/api/data_sources/task_list_id
    p = requests.put(requrl,headers=headers, data=json.dumps(content) )
    return p

这个个B大同小异,就是多了个content,这个content就是D处理完成之后返回的那个dict。

# F. 主流程

函数都写好了之后把主流程写到主体结构中:

1
2
3
4
5
6
7
8
9
10
if __name__ == "__main__":
token = input("token?") #用户输入token
url = input("request url?") #用户输入请求地址

for i in range(0,1000): #因为要让下面的部分循环,所以就循环1000次
tname = input("task name / path?") #输入任务名,其实也是路径
g = listgetter(token, url)
t = tailor(g)
a = taskadder(t, tname)
r = taskputter(a, token, url)

完成。