自动打卡脚本

发布于 2020-03-31  271 次阅读


前言 学程每日打卡,研究了一下,顺手写了一个自动打卡的脚本。

模块引用:

requests 核心模块,用于网络请求

json Json解析库

datetime 时间库

sys 系统库

yagmail 用于邮件通知

apscheduler 定时任务模块

代码:

# coding=utf-8
# Code By Rade
import requests
import json
import datetime
import time
import sys
import yagmail
from apscheduler.schedulers.blocking import BlockingScheduler

reload(sys)
sys.setdefaultencoding('utf8')

# 常量
user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:74.0) Gecko/20100101 Firefox/74.0'
# 登录地址
login_url = 'https://pa.pkqa.com.cn/dapi/v2/account/account_service/login'
# 打卡请求
clock_req_url = 'https://pa.pkqa.com.cn/dapi/v2/autoform/autoform_service/save_form_input'
# 查询当日组信息
find_item_url = 'https://pa.pkqa.com.cn/dapi/v2/form/daily_check_in_service/find_item_by_theme_id_and_date_with_self'
# 邮件通知地址
notice_email = '邮件通知地址'
# 错误次数预设阈值
clock_error_num_max = 10
# 用户名
loginName = "学号"
# 密码
password = "密码"


def clock_in(clock_error_num=0):
    # time.sleep(3)
    if clock_error_num > clock_error_num_max:
        print("异常次数超过预设阈值,结束打卡!")
    else:
        headers = {
            'User-agent': user_agent,
            'App-Code': '3',
            'Content-Type': 'application/json;charset=UTF-8'
        }
        login_data = {
            "loginName": loginName,
            "password": password,
            "type": "account"
        }

        resp = requests.post(login_url, headers=headers, data=json.dumps(login_data))
        res_json = json.loads(resp.content)

        # 异常判断
        if res_json['code'] != 200:
            print("[" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "] 登录异常,失败原因:" + res_json['msg'])
            clock_error_num += 1
            print("正在第" + str(clock_error_num) + "次尝试重试...")
            if clock_error_num == clock_error_num_max + 1:
                send_email(0, res_json['msg'], loginName)
            clock_in(clock_error_num)
        else:
            token = "Bearer " + res_json['data']['token']  # 获取Token

            user_real_name = res_json['data']['user']['user']['name']

            print("[" + datetime.datetime.now().strftime(
                '%Y-%m-%d %H:%M:%S') + "] 用户:" + user_real_name + " 登录成功,携带Token值为:" + token)

            # 发送打卡请求
            headers = {
                'User-agent': user_agent,
                'authorization': token,
                'app-code': "3",
                'Content-Type': 'application/json'
            }

            # 查询groupId信息
            item_data = {
                "themeId": "5e49692499f12000011ceef5",
                "date": int(round(time.time() * 1000))
            }

            resp = requests.post(find_item_url, headers=headers, data=json.dumps(item_data))
            res_json = json.loads(resp.content)

            # 异常判断
            if res_json['code'] != 200:
                print("[" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "] 查询表单信息异常,失败原因:" + res_json[
                    'msg'])
                clock_error_num += 1
                print("正在第" + str(clock_error_num) + "次尝试重试...")
                if clock_error_num == clock_error_num_max + 1:
                    send_email(0, res_json['msg'], loginName)
                clock_in(clock_error_num)
            else:
                group_id = res_json['data']['group']['id']  # 获取groupId
                group_title = res_json['data']['group']['title']  # 获取groupTitle
                print("[" + datetime.datetime.now().strftime(
                    '%Y-%m-%d %H:%M:%S') + "] 用户:" + user_real_name + " 取" + group_title + "信息成功,携带group_id值为:" + group_id)
                clock_req_data = {
                    "bizType": "D00",
                    "groupid": group_id,
                    "value": [
                        {
                            "location": [
                                "北京市",
                                "北京城区",
                                "朝阳区"
                            ],
                            "whatColorIsYourHangzhouHealthCode": "greenCode",
                            "currentHealthCondition": "beInGoodHealth",
                            "currentLifeSituation": "normalHome",
                            "closeContactWithConfirmedOrSuspectedCases": "no",
                            "inHubeiOrPassOrComeIntoContactWithPeopleFromHubei": "no",
                            "inWenzhouHuangyanWenlingOrPassOrContactPersonsFromTheAboveAreas": "no"
                        }
                    ]
                }

                resp = requests.post(clock_req_url, headers=headers, data=json.dumps(clock_req_data))
                res_json = json.loads(resp.content)

                # 异常判断
                if res_json['code'] != 200:
                    print("[" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "] 打卡异常,失败原因:" + res_json[
                        'msg'])
                    clock_error_num += 1
                    print("正在第" + str(clock_error_num) + "次尝试重试...")
                    if clock_error_num == clock_error_num_max + 1:
                        send_email(0, res_json['msg'], user_real_name)
                    clock_in(clock_error_num)
                else:
                    clock_error_num = 0
                    print("[" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "] 打卡成功!")
                    send_email(1, "", user_real_name)


def send_email(state, msg, name):
    yag = yagmail.SMTP(user="xxxxx", password="xxxxx", host="xxxxx", port=465)
    if state:
        contents = "用户" + name + "在" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "学程打卡成功"
        yag.send(notice_email, '学程自动打卡成功', contents)
    else:
        contents = "用户" + name + "在" + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "学程打卡失败,原因是因为" + msg
        yag.send(notice_email, '学程自动打卡失败', contents)


# 定时执行
scheduler = BlockingScheduler()
scheduler.add_job(clock_in, 'cron', hour=5, minute=01)
scheduler.start()

预览效果

回显预览
邮件通知

Nothing is impossible!