今日校園自動提交 AUST

說明

• 此為 AUST （安徽理工大學）專版。其實就是為了方便校友
• 基於騰訊雲函數
• 使用了自建api。大佬造好的輪子。
• 新增消息通知Server醬
• 新增消息通知Qmsg醬

所用代碼

config.yml

#登陸相關配置
login:
  #開放的模擬登陸api服務器地址
  api: "http://cpdaily.kangblogs.top:8080/wisedu-unified-login-api-v1.0/api/login"
#用戶組配置
users:
  #單個用戶配置
  - user:
      #username 學號或者工號
      username: '2019******'
      #password 密碼
      password: '*************'
      #address 地址，定位信息
      address: 中國**省**市
      #email 接受通知消息的郵箱
      email: ************@qq.com
      #lon 當前位置經度，可以訪問http://zuobiao.ay800.com/s/27/index.php獲取
      lon: '11*.******'
      #lat 當前位置緯度
      lat: '3*.******'
      #school 學校全稱
      school: 安徽理工大學
#今日校園相關配置
cpdaily:
  #表單組默認選項配置
  defaults:
    #表單默認選項配置，按順序，注意，只有標必填項的才處理，不會配置就執行generate.py
    - default:
        title: 今天你的所在地是？
        type: 1
        value: **省/**市/**市
    - default:
        title: 今天你的體溫是多少？
        type: 2
        value: 37.2℃及以下
    - default:
        title: 今天你的身體狀況是？
        type: 2
        value: 健康
    - default:
        title: 近14天你或你的共同居住人是否有疫情中、高風險區域人員接觸史？
        type: 2
        value: 否
    - default:
        title: 近14天你或你的共同居住人是否和確診、疑似病人接觸過？
        type: 2
        value: 否
    - default:
        title: 近14天你或你的共同居住人是否是確診、疑似病例？
        type: 2
        value: 否
    - default:
        title: 你或你的共同居住人目前是否被醫學隔離？
        type: 2
        value: 否
    - default:
        title: 今天你當地的健康碼顏色是？
        type: 2
        value: 綠色
    - default:
        title: 本人是否承諾以上所填報的全部內容均屬實、准確，不存在任何隱瞞與不實的情況，更無遺漏之處
        type: 2
        value: 是

generate.py

# -*- coding: utf-8 -*-
import index as app
import yaml


# 生成默認配置
def generate():
    config = app.config
    user = config['users'][0]
    apis = app.getCpdailyApis(user)
    session = app.getSession(user, apis['login-url'])
    form = dict(app.queryForm(session, apis))['form']
    # app.log(form)
    defaults = []
    sort = 1
    for formItem in form:
        if formItem['isRequired'] == 1:
            default = {}
            one = {}
            default['title'] = formItem['title']
            default['type'] = formItem['fieldType']
            print('問題%d：' % sort + default['title'])
            if default['type'] == 1 or default['type'] == 5:
                default['value'] = input("請輸入文本：")
            if default['type'] == 2:
                fieldItems = formItem['fieldItems']
                num = 1
                for fieldItem in fieldItems:
                    print('\t%d ' % num + fieldItem['content'])
                    num += 1
                choose = int(input("請輸入序號："))
                if choose < 1 or choose > num:
                    print('輸入錯誤，請重新執行此腳本')
                    exit(-1)
                default['value'] = fieldItems[choose - 1]['content']
            if default['type'] == 3:
                fieldItems = formItem['fieldItems']
                num = 1
                for fieldItem in fieldItems:
                    print('\t%d ' % num + fieldItem['content'])
                    num += 1
                chooses = list(map(int, input('請輸入序號（可輸入多個，請用空格隔開）：').split()))
                default['value'] = ''
                for i in range(0, len(chooses)):
                    choose = chooses[i]
                    if choose < 1 or choose > num:
                        print('輸入錯誤，請重新執行此腳本')
                        exit(-1)
                    if i != len(chooses) - 1:
                        default['value'] += fieldItems[choose - 1]['content'] + ','
                    else:
                        default['value'] += fieldItems[choose - 1]['content']
            if default['type'] == 4:
                default['value'] = input("請輸入圖片名稱：")
            one['default'] = default
            defaults.append(one)
            sort += 1
    print('======================分隔線======================')
    print(yaml.dump(defaults, allow_unicode=True))


if __name__ == "__main__":
    generate()

index.py

# -*- coding: utf-8 -*-
import sys
import requests
import json
import yaml
import oss2
from urllib.parse import urlparse
from datetime import datetime, timedelta, timezone
from urllib3.exceptions import InsecureRequestWarning
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr

# debug模式
debug = False
if debug:
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


# 讀取yml配置
def getYmlConfig(yaml_file='config.yml'):
    file = open(yaml_file, 'r', encoding="utf-8")
    file_data = file.read()
    file.close()
    config = yaml.load(file_data, Loader=yaml.FullLoader)
    return dict(config)


# 全局配置
config = getYmlConfig(yaml_file='config.yml')


# 獲取今日校園api
def getCpdailyApis(user):
    headers = { 'User-Agent': 'Mozilla/5.0 (iPad; U; CPU OS 3_2_2 like Mac OS X; en-us) AppleWebKit/531.21.10 (KHTML, like Gecko) Version/4.0.4 Mobile/7B500 Safari/531.21.10' }
    apis = {}
    user = user['user']
    schools = requests.get(
        url='https://mobile.campushoy.com/v6/config/guest/tenant/list', verify=not debug).json()['data']
    flag = True
    for one in schools:
        if one['name'] == user['school']:
            if one['joinType'] == 'NONE':
                log(user['school'] + ' 未加入今日校園')
                exit(-1)
            flag = False
            params = {
                'ids': one['id']
            }
            res = requests.get(url='https://mobile.campushoy.com/v6/config/guest/tenant/info', params=params,
                               verify=not debug)
            data = res.json()['data'][0]
            joinType = data['joinType']
            idsUrl = data['idsUrl']
            ampUrl = data['ampUrl']
            if 'campusphere' in ampUrl or 'cpdaily' in ampUrl:
                parse = urlparse(ampUrl)
                host = parse.netloc
                res = requests.get(parse.scheme + '://' + host)
                parse = urlparse(res.url)
                apis[
                    'login-url'] = idsUrl + '/login?service=' + parse.scheme + r"%3A%2F%2F" + host + r'%2Fportal%2Flogin'
                apis['host'] = host

            ampUrl2 = data['ampUrl2']
            if 'campusphere' in ampUrl2 or 'cpdaily' in ampUrl2:
                parse = urlparse(ampUrl2)
                host = parse.netloc
                res = requests.get(parse.scheme + '://' + host)
                parse = urlparse(res.url)
                apis[
                    'login-url'] = idsUrl + '/login?service=' + parse.scheme + r"%3A%2F%2F" + host + r'%2Fportal%2Flogin'
                apis['host'] = host
            break
    if flag:
        log(user['school'] + ' 未找到該院校信息，請檢查是否是學校全稱錯誤')
        exit(-1)
    log(apis)
    return apis


# 獲取當前utc時間，並格式化為北京時間
def getTimeStr():
    utc_dt = datetime.utcnow().replace(tzinfo=timezone.utc)
    bj_dt = utc_dt.astimezone(timezone(timedelta(hours=8)))
    return bj_dt.strftime("%Y-%m-%d %H:%M:%S")


# 輸出調試信息，並及時刷新緩沖區
def log(content):
    print(getTimeStr() + ' ' + str(content))
    sys.stdout.flush()


# 登陸並返回session
def getSession(user, loginUrl):
    user = user['user']
    params = {
        'login_url': loginUrl,
        # 保證學工號和密碼正確下面兩項就不需要配置
        'needcaptcha_url': '',
        'captcha_url': '',
        'username': user['username'],
        'password': user['password']
    }

    cookies = {}
    # 借助上一個項目開放出來的登陸API，模擬登陸
    res = requests.post(config['login']['api'], params, verify=not debug)
    cookieStr = str(res.json()['cookies'])
    log(cookieStr)
    if cookieStr == 'None':
        log(res.json())
        return None

    # 解析cookie
    for line in cookieStr.split(';'):
        name, value = line.strip().split('=', 1)
        cookies[name] = value
    session = requests.session()
    session.cookies = requests.utils.cookiejar_from_dict(cookies)
    return session


# 查詢表單
def queryForm(session, apis):
    host = apis['host']
    headers = {
        'Accept': 'application/json, text/plain, */*',
        'User-Agent': 'Mozilla/5.0 (Linux; Android 4.4.4; OPPO R11 Plus Build/KTU84P) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/33.0.0.0 Safari/537.36 yiban/8.1.11 cpdaily/8.1.11 wisedu/8.1.11',
        'content-type': 'application/json',
        'Accept-Encoding': 'gzip,deflate',
        'Accept-Language': 'zh-CN,en-US;q=0.8',
        'Content-Type': 'application/json;charset=UTF-8'
    }
    queryCollectWidUrl = 'https://{host}/wec-counselor-collector-apps/stu/collector/queryCollectorProcessingList'.format(
        host=host)
    params = {
        'pageSize': 6,
        'pageNumber': 1
    }
    res = session.post(queryCollectWidUrl, headers=headers,
                       data=json.dumps(params), verify=not debug)
    if len(res.json()['datas']['rows']) < 1:
        return None

    collectWid = res.json()['datas']['rows'][0]['wid']
    formWid = res.json()['datas']['rows'][0]['formWid']

    detailCollector = 'https://{host}/wec-counselor-collector-apps/stu/collector/detailCollector'.format(
        host=host)
    res = session.post(url=detailCollector, headers=headers,
                       data=json.dumps({"collectorWid": collectWid}), verify=not debug)
    schoolTaskWid = res.json()['datas']['collector']['schoolTaskWid']

    getFormFields = 'https://{host}/wec-counselor-collector-apps/stu/collector/getFormFields'.format(
        host=host)
    res = session.post(url=getFormFields, headers=headers, data=json.dumps(
        {"pageSize": 100, "pageNumber": 1, "formWid": formWid, "collectorWid": collectWid}), verify=not debug)

    form = res.json()['datas']['rows']
    return {'collectWid': collectWid, 'formWid': formWid, 'schoolTaskWid': schoolTaskWid, 'form': form}


# 填寫form
def fillForm(session, form, host):
    sort = 1
    for formItem in form[:]:
        # 只處理必填項
        if formItem['isRequired'] == 1:
            default = config['cpdaily']['defaults'][sort - 1]['default']
            if formItem['title'] != default['title']:
                log('第%d個默認配置不正確，請檢查' % sort)
                exit(-1)
            # 文本直接賦值
            if formItem['fieldType'] == 1 or formItem['fieldType'] == 5:
                formItem['value'] = default['value']
            # 單選框需要刪掉多余的選項
            if formItem['fieldType'] == 2:
                # 填充默認值
                formItem['value'] = default['value']
                fieldItems = formItem['fieldItems']
                for i in range(0, len(fieldItems))[::-1]:
                    if fieldItems[i]['content'] != default['value']:
                        del fieldItems[i]
            # 多選需要分割默認選項值，並且刪掉無用的其他選項
            if formItem['fieldType'] == 3:
                fieldItems = formItem['fieldItems']
                defaultValues = default['value'].split(',')
                for i in range(0, len(fieldItems))[::-1]:
                    flag = True
                    for j in range(0, len(defaultValues))[::-1]:
                        if fieldItems[i]['content'] == defaultValues[j]:
                            # 填充默認值
                            formItem['value'] += defaultValues[j] + ' '
                            flag = False
                    if flag:
                        del fieldItems[i]
            # 圖片需要上傳到阿里雲oss
            if formItem['fieldType'] == 4:
                fileName = uploadPicture(session, default['value'], host)
                formItem['value'] = getPictureUrl(session, fileName, host)
            log('必填問題%d：' % sort + formItem['title'])
            log('答案%d：' % sort + formItem['value'])
            sort += 1
        else:
            form.remove(formItem)
    # print(form)
    return form


# 上傳圖片到阿里雲oss
def uploadPicture(session, image, host):
    url = 'https://{host}/wec-counselor-collector-apps/stu/collector/getStsAccess'.format(
        host=host)
    res = session.post(url=url, headers={
        'content-type': 'application/json'}, data=json.dumps({}), verify=not debug)
    datas = res.json().get('datas')
    fileName = datas.get('fileName')
    accessKeyId = datas.get('accessKeyId')
    accessSecret = datas.get('accessKeySecret')
    securityToken = datas.get('securityToken')
    endPoint = datas.get('endPoint')
    bucket = datas.get('bucket')
    bucket = oss2.Bucket(oss2.Auth(access_key_id=accessKeyId,
                                   access_key_secret=accessSecret), endPoint, bucket)
    with open(image, "rb") as f:
        data = f.read()
    bucket.put_object(key=fileName, headers={
        'x-oss-security-token': securityToken}, data=data)
    res = bucket.sign_url('PUT', fileName, 60)
    # log(res)
    return fileName


# 獲取圖片上傳位置
def getPictureUrl(session, fileName, host):
    url = 'https://{host}/wec-counselor-collector-apps/stu/collector/previewAttachment'.format(
        host=host)
    data = {
        'ossKey': fileName
    }
    res = session.post(url=url, headers={
        'content-type': 'application/json'}, data=json.dumps(data), verify=not debug)
    photoUrl = res.json().get('datas')
    return photoUrl


# 提交表單
def submitForm(formWid, address, collectWid, schoolTaskWid, form, session, host):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Linux; Android 4.4.4; OPPO R11 Plus Build/KTU84P) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/33.0.0.0 Safari/537.36 okhttp/3.12.4',
        'CpdailyStandAlone': '0',
        'extension': '1',
        'Cpdaily-Extension': 'eZbW2qLZT0G0VbYqnj5mz5UCyZiuS+Mht0ro4VCSTgTancCpi4ru3IpfZibLN2Q4JR3dl7wYTXnTi5dzfAwbYcs5FB4VPqOTrcYNVjoRY9h9J7sxA1MWIWZxiEC7iuzXwAeEjrGmnHnX3P7mprZW66fbhNsIrM938cVo6aK7fgdQx6vGY7OVJBS+kqwk/xE2ipLqV0ro4QNZ9u/6G9MUbyd7QghLIM9PIRJTrd6TzoYPFBHqDHIY57dHHUBUC8RzfvreU/2o5sY=',
        'Content-Type': 'application/json; charset=utf-8',
        # 請注意這個應該和配置文件中的host保持一致
        'Host': host,
        'Connection': 'Keep-Alive',
        'Accept-Encoding': 'gzip'
    }

    # 默認正常的提交參數json
    params = {"formWid": formWid, "address": address, "collectWid": collectWid, "schoolTaskWid": schoolTaskWid,
              "form": form, "uaIsCpadaily": True}
    # print(params)
    submitForm = 'https://{host}/wec-counselor-collector-apps/stu/collector/submitForm'.format(
        host=host)
    r = session.post(url=submitForm, headers=headers,
                     data=json.dumps(params), verify=not debug)
    msg = r.json()['message']

    return msg


title_text = '今日校園疫結果通知'


# server醬通知
# def sendServerChan(msg):
    # log('正在發送Server醬。。。')
    # SCKEY = "SCU103057*****************************"  # 這里填Server醬官方提供的SCKEY
    # url = "https://sc.ftqq.com/%s.send" % (SCKEY)
    # res = requests.post(url, data={'text': title_text, 'desp': getTimeStr() + "\n" + str(msg)})
    # code = res.json()['errmsg']
    # if code == 'success':
        # log('發送Server醬通知成功。。。')
    # else:
        # log('發送Server醬通知失敗。。。')
        # log('Server醬返回結果' + code)


# Qmsg醬通知
def sendQmsgChan(msg):
    log('正在發送Qmsg醬。。。')
    QmsgKey = "1aacce**********************"  # 這里填Qmsg醬官方提供的KEY
    url_send = "https://qmsg.zendee.cn/send/%s" % (QmsgKey)
    url_group = "https://qmsg.zendee.cn/group/%s" % (QmsgKey)
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'
    }
    payload = {'msg': '%s今日校園疫情上報自動填表結果通知' % (getTimeStr() + '\n') + '\n' + str(msg)}
    res = requests.post(url_send, params=payload, headers=headers)  # 私聊消息推送
    res_group = requests.post(url_group, params=payload, headers=headers)  # 群消息推送
    sucmsg = res.json()['success']
    if sucmsg == True:
        log('Qmsg醬通知成功')
    else:
        log('推送失敗')
        log(res.json())


def main_handler(event, context):
    try:
        for user in config['users']:
            log('當前用戶：' + str(user['user']['username']))
            apis = getCpdailyApis(user)
            log('腳本開始執行。。。')
            log('開始模擬登陸。。。')
            session = getSession(user, apis['login-url'])
            if session != None:
                log('模擬登陸成功。。。')
                log('正在查詢最新待填寫問卷。。。')
                params = queryForm(session, apis)
                if str(params) == 'None':
                    log('獲取最新待填寫問卷失敗，可能是輔導員還沒有發布。。。')
                    # sendServerChan('沒有新問卷')
                    sendQmsgChan('沒有新問卷')
                    exit(-1)
                log('查詢最新待填寫問卷成功。。。')
                log('正在自動填寫問卷。。。')
                form = fillForm(session, params['form'], apis['host'])
                log('填寫問卷成功。。。')
                log('正在自動提交。。。')
                msg = submitForm(params['formWid'], user['user']['address'], params['collectWid'],
                                 params['schoolTaskWid'], form, session, apis['host'])
                if msg == 'SUCCESS':
                    log('自動提交成功！')
                    # sendServerChan('🎉自動提交成功！o(*≧▽≦)ツ')
                    sendQmsgChan('🎉自動提交成功！o(*≧▽≦)ツ')
                elif msg == '該收集已填寫無需再次填寫':
                    log('今日已提交！')
                    # sendServerChan('🤔今日已提交！')
                    sendQmsgChan('🤔今日已提交！')
                else:
                    log('自動提交失敗。。。')
                    log('錯誤是' + msg)
                    # sendServerChan('自動提交失敗！錯誤是' + str(msg))
                    sendQmsgChan('自動提交失敗！錯誤是' + str(msg))
                    exit(-1)
            else:
                log('模擬登陸失敗。。。')
                log('原因可能是學號或密碼錯誤，請檢查配置后，重啟腳本。。。')
                exit(-1)
    except Exception as e:
        # sendServerChan("出現問題了！" + str(e))
        sendQmsgChan("出現問題了！" + str(e))
        raise e
    else:
        return 'success'


# 配合Windows計划任務等使用
if __name__ == '__main__':
    print(main_handler({}, {}))

• 層依賴 dependency.zip，點擊下載

注意

• 在config.yml中添加今日校園的賬戶信息😀、郵箱📪、address、經緯度以及第一個問題的所在地
• 在index.py中添加自己的SCKEY（281行左右）
• 在index.py中添加自己的QmsgKEY（295行左右）
• 有條件的可以自建api

Congratulations!