Python爬虫实战5:渤大自动健康打卡-南国旧梦i

前言

渤大学子众所周知目前学校打卡通过企业微信：1-企业微信健康上报，2-企业微信学工系统。前者本人能力有限分析不透企业微信cookie，好处是它保留昨天的记录，每天只需要点击提交即可。后者则是需要每天定位、填写信息体温等，抓包发现学工系统对接的是易班APP，于是我参考CSDN- looyeagee 教程并升级改进出了一个适合我们学校的易班APP自动填写的脚本,为了避免一些不会抓包的同学，我写了一个GetForm.py来通过曾经表单直接返回表单的提交内容。

注意：这个脚本只适合渤海大学，如果是其他学校的同学，需要重新抓包修改。

言归正传，此项目包含三个文件：main.py(执行)，yiban.py(对象类)，util(时间工具)。

这里我给出的辅助文件GetForm.py用来获取打卡提交内容,好处是不用单独抓包。

使用教程： 打开企业微信-工作台-学工系统-任务反馈-已办-随便进入一次上报-我的反馈-转发审批表单-复制链接替换到GetForm.py的url变量

GetForm.py

# -*- coding: utf-8 -*-
#   @Time    : 2021/2/18 11:47
#   @Author  : 南国旧梦i
#   @FileName: GetForm.py
#   @Software: PyCharm
import json
import random
import requests
#FIXME 更改 base_url  CSRF当前cookie 固定值无需更改
base_url = 'https://app.uyiban.com/workflow/client/#/share?initiateId=ecea7f1f4db3f0a2a9f1657ac1c4efba'
def parse_data(url):
    """
    :param url
    将分享链接的数据解析出来生成表单数据字典
    """
    #TODO 进阶版本 写为分享表单自动获取
    initiateId = url.split('=')[-1]
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.100 Safari/537.36',
        'origin': 'https://app.uyiban.com',
        'referer': 'https://app.uyiban.com/',
        'cookie': 'PHPSESSID=7867f91581ade0a7e29c3805622d8055; csrf_token=7bed020ce1638b4cbaac0004f4366033'
    }
    form_url='https://api.uyiban.com/workFlow/c/share/index?InitiateId={}&amp;CSRF=7bed020ce1638b4cbaac0004f4366033'.format(initiateId)
    res_con = requests.get(form_url,headers=headers)
    save_data_url= res_con.json()['data']['uri']
    res_form = requests.get(save_data_url).json()
    FormDataJson = res_form['Initiate']['FormDataJson']
    dict_form = {i.get('id'): i.get("value") for i in FormDataJson}
    return dict_form
if __name__ == '__main__':
    # 最终提交的表单数据
    form_data = parse_data(base_url)
    print(form_data)
    print("请将上面输出结果复制替换到main.py的28行的dict_form的字典")

下列三个py文件放在同一目录，执行main.py即可。

util.py

# -*- coding: utf-8 -*-
#   @Time    : 2021/2/17 16:30
#   @Author  : 南国旧梦i
#   @FileName: util.py
#   @Software: PyCharm
import datetime
import time
def get_time():
    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
def get_time_no_second():
    return time.strftime("%Y-%m-%d %H:%M", time.localtime())
def get_7_day_ago():
    now = datetime.datetime.now()
    delta = datetime.timedelta(days=-7)
    n_days = now + delta
    return n_days.strftime('%Y-%m-%d')
def get_today():
    return time.strftime("%Y-%m-%d", time.localtime())
def desc_sort(array, key="FeedbackTime"):
    for i in range(len(array) - 1):
        for j in range(len(array) - 1 - i):
            if array[j][key] < array[j + 1][key]:
                array[j], array[j + 1] = array[j + 1], array[j]
    return array

yiban.py

# -*- coding: utf-8 -*-
#   @Time    : 2021/2/17 14:30
#   @Author  : 南国旧梦i
#   @FileName: yiban.py
#   @Software: PyCharm
import re
import requests
import util
class YiBan:
    CSRF = "a6b0515sf434b7d1b652fba42490297f"  # 随机值 随便填点东西
    COOKIES = {"csrf_token": CSRF}  # 固定cookie 无需更改
    HEADERS = {"Origin": "https://c.uyiban.com", "User-Agent": "yiban"}  # 固定头 无需更改

    def __init__(self, account, passwd):
        self.account = account
        self.passwd = passwd
        self.session = requests.session()
        self.name = ""
        self.iapp = ""

    def request(self, url, method="get", params=None, cookies=None):
        if method == "get":
            req = self.session.get(url, params=params, timeout=10, headers=self.HEADERS, cookies=cookies)
        else:
            req = self.session.post(url, data=params, timeout=10, headers=self.HEADERS, cookies=cookies)
        try:
            return req.json()
        except:
            return None

    def login(self):
        params = {
            "mobile": self.account,
            "imei": "0",
            "password": self.passwd
        }
        r = self.request(url="https://mobile.yiban.cn/api/v3/passport/login", params=params)
        if r is not None and str(r["response"]) == "100":
            self.access_token = r["data"]["user"]["access_token"]
            return r
        else:
            raise Exception("账号或密码错误")
    def getHome(self):
        params = {
            "access_token": self.access_token,
        }
        r = self.request(url="https://mobile.yiban.cn/api/v3/home", params=params)
        self.name = r["data"]["user"]["userName"]
        for i in r["data"]["hotApps"]: # 动态取得iapp号
            if i["name"] == "校本化":
                self.iapp = re.findall(r"(iapp.*)\?", i["url"])[0]
        return r
    def auth(self):
        params = {
            "act": self.iapp,
            "v": self.access_token
        }
        print()
        location = self.session.get("http://f.yiban.cn/iapp/index",params=params,allow_redirects=False).headers.get("Location")
        if location is None:
            raise Exception("该用户可能没进行校方认证，无此APP权限")
        verifyRequest = re.findall(r"verify_request=(.*?)&amp;", location)[0]
        result_auth = self.request(
            "https://api.uyiban.com/base/c/auth/yiban?verifyRequest=%s&amp;CSRF=%s" % (verifyRequest, self.CSRF),
            cookies=self.COOKIES)
        data_url = result_auth["data"].get("Data")
        if data_url is not None:  # 授权过期
            result_html = self.session.get(url=data_url, headers=self.HEADERS,cookies={"loginToken": self.access_token}).text
            re_result = re.findall(r'input type="hidden" id="(.*?)" value="(.*?)"', result_html)
            post_data = {"scope": "1,2,3,"}
            for re_i in re_result:
                post_data[re_i[0]] = re_i[1]
            usersure_result = self.session.post(url="https://oauth.yiban.cn/code/usersure",data=post_data,headers=self.HEADERS, cookies={"loginToken": self.access_token})
            if usersure_result.json()["code"] == "s200":
                return self.auth()
            else:
                return False
        else:
            return True

    def getUncompletedList(self):
        params = {
            "CSRF": self.CSRF,
            "StartTime": util.get_today(),
            "EndTime": util.get_time()
        }
        return self.request("https://api.uyiban.com/officeTask/client/index/uncompletedList", params=params,cookies=self.COOKIES)

    def getCompletedList(self):
        params = {
            "CSRF": self.CSRF,
            "StartTime": util.get_7_day_ago(),
            "EndTime": util.get_time()
        }
        return self.request("https://api.uyiban.com/officeTask/client/index/completedList", params=params,
                            cookies=self.COOKIES)

    def getJsonByInitiateId(self, initiate_id):
        params = {
            "CSRF": self.CSRF
        }
        return self.request("https://api.uyiban.com/workFlow/c/work/show/view/%s" % initiate_id, params=params,
                            cookies=self.COOKIES)

    def getTaskDetail(self, taskId):
        return self.request(
            "https://api.uyiban.com/officeTask/client/index/detail?TaskId=%s&amp;CSRF=%s" % (taskId, self.CSRF),
            cookies=self.COOKIES)

    def submit(self, data, extend, wfid):
        params = {
            "data": data,
            "extend": extend
        }
        return self.request(
            "https://api.uyiban.com/workFlow/c/my/apply/%s?CSRF=%s" % (wfid, self.CSRF), method="post",
            params=params,
            cookies=self.COOKIES)

    def getShareUrl(self, initiateId):
        return self.request(
            "https://api.uyiban.com/workFlow/c/work/share?InitiateId=%s&amp;CSRF=%s" % (initiateId, self.CSRF),
            cookies=self.COOKIES)

main.py

import json
import random
from yiban import YiBan
import util
if __name__ == '__main__':
    try:
        yb = YiBan("易班账号", "密码") # TODO:账号密码
        yb.login()
        yb.getHome()
        print("登录成功 %s"%yb.name)
        yb.auth()
        all_task = yb.getUncompletedList()["data"]
        # print(all_task)
        all_task = list(filter(lambda x: "渤海大学" in x["Title"], all_task))  # FIXME: 根据标题找出任务列表
        if len(all_task) == 0:
            print("没找到今天渤大健康打卡的任务，可能是你已经上报，如果不是请手动上报。")
        else:
            all_task_sort = util.desc_sort(all_task, "StartTime")  # 按开始时间排序
            new_task = all_task_sort[0]  # 只取一个最新的
            print("找到未上报的任务：", new_task)
            task_detail = yb.getTaskDetail(new_task["TaskId"])["data"]
            ex = {"TaskId": task_detail["Id"],
                "title": "任务信息",
                "content": [{"label": "任务名称", "value": task_detail["Title"]},
                            {"label": "发布机构", "value": task_detail["PubOrgName"]},
                            {"label": "发布人", "value": task_detail["PubPersonName"]}]}
            # FIXME: 以下是【渤海大学】最新的表单信息，由于某些值（检测时间）必须是动态的，所以只能将form表单写死在这里 （可能会变）
            dict_form = {
                        '074dd5445cbb2f5f9ff46b3d1e59f43a': {
                            'time': util.get_time_no_second(),
                            'longitude': xxxxxxxx,
                            'latitude': xxxxx,
                            'address': 'xxxxxxx'
                        },
                        '6d128d8627101947365ebf305a2cfa98': '在家',
                        'c0f1ff887ba61230220dc6440ba78dcc': ['江苏省', 'xxx', 'xxx'],
                        'cfaeadae3b6395849a6109ff1d2a41c6': ["36.2", "36.3", "36.4", "36.5", "36.6", "36.7", "36.8"][random.randint(0, 6)],
                        'a70e21b065f12069260936ffd3ad6402': '健康',
                        'dd92f076c3c4931f03a9ffd99f3efbcc': None,
                        'd933705567588916c64c5ae20c91e1d7': '知道',
                        'a991a9051e0b2681094771e204847e48': '无',
                        '6dbe122c504151ce7f239cd7ffb985e3': '无',
                        '9380bbbb20ca38d48f0950f2854707b0': '无'
            }
            submit_result = yb.submit(json.dumps(dict_form, ensure_ascii=False), json.dumps(ex, ensure_ascii=False), task_detail["WFId"])
            if submit_result["code"] == 0:
                share_url = yb.getShareUrl(submit_result["data"])["data"]["uri"]
                print("已完成一次健康上报[%s]" % task_detail["Title"])
                print("访问此网址查看详情：%s" % share_url)
            else:
                print("[%s]遇到了一些错误:%s" % (task_detail["Title"], submit_result["msg"]))
    except Exception as e:
        print("出错啦")
        print(e)

✨ 使用说明：

根据GetForm获取提交内容替换main.py的28行
配置main.py内的账号和密码执行main.py即可

✨ 服务器每天定时打卡：

每天六点自动执行打卡,将结果保存到当前用户的home目录下的result.txt

00 06 * * * python3 mian.py >> ~/result.txt