騰訊云函式實作今日校園自動填寫

說明

參考github上auto-submit專案
學習交流使用

工具

騰訊云函式

注意

1 附件里面的config.yml只有HFUT可以用，其他學校需要用generate.py來進行配置
2 少部分學校不支持云函式

使用步驟

配置config.yml中對應的學號（username）和密碼（password）還有地址（address）等等資訊，詳情請看config.yml中的注釋說明，注意這里的學號和密碼都是今日校園的學號和密碼
打開百度搜索騰訊云函式，注冊認證后
1. 進入控制臺
  ，
2. 進入云函式
3. 點擊左邊的層
4. 點新建，名稱隨意
5. 然后點擊上傳zip，選擇dependency.zip上傳
6. 選擇運行環境python3.6
7. 然后點擊確定，耐心等待一下，上傳依賴包需要花費的時間比較長, 完成后界面如下
8. 點左邊的函式服務，新建云函式
  
  名稱隨意，運行環境選擇python3.6，創建方式選擇空白函式，然后點擊下一步
提交方法選擇在線編輯，把附件里面的index.py直接全文復制粘貼到云函式的index.py，
然后點擊檔案->新建，檔案名命名為config.yml，然后把下面附件里面config.yml的內容直接全文復制粘貼到云函式的config.yml檔案（別的學校需要自己用generate.py配置好），點擊下面的高級設定，設定超時時間為60秒，添加層為剛剛新建的函式依賴層，然后點擊完成
進入新建好的云函式，左邊點擊觸發管理，點擊創建觸發器，名稱隨意，觸發周期選擇自定義，然后配置cron運算式，下面的運算式表示每天晚上20：20執行
```
0 20 20 * * * *
```

附件

dependency.zip

dependency

config.yml

#登陸相關配置
login:
  #開放的模擬登陸api服務器地址
  api: "http://www.zimo.wiki:8080/wisedu-unified-login-api-v1.0/api/login"
#用戶組配置
users:
  #單個用戶配置
  - user:
      #username 學號或者工號
      username: '1111111111'
      #password 密碼
      password: '111111111'
      #address 地址，定位資訊
      address: 中國四川省成都市金牛區一環路北1段-129號-附9號
      #email 接受通知訊息的郵箱
      email: 1111111@qq.com
      #school 學校全稱
      school: 合肥工業大學
  #多用戶配置，將下面的注釋去掉即可，如果有表單內容有圖片，不建議使用多用戶配置
#  - user:
#      #username 學號或者工號
#      username: 161105024
#      #password 密碼
#      password: 161105024
#      #address 地址
#      address: 中國四川省成都市金牛區一環路北1段-129號-附9號
#      #email 接受通知訊息的郵箱
#      email: 461009747@qq.com
#      school: 宜賓學院
#今日校園相關配置
cpdaily:
  #表單組默認選項配置
  defaults:
    #表單默認選項配置，按順序，注意，只有標必填項的才處理
    - default:
        title: 今日在校情況
        type: 2
        value: 在校，校內住宿
    - default:
        title: 今日所在地
        type: 1
        value: 安徽省/合肥市/蜀山區
    - default:
        title: 今日晨檢體溫
        type: 2
        value: 正常，不發熱
    - default:
        title: 今日午檢體溫
        type: 2
        value: 正常，不發熱
    - default:
        title: 今日晚檢體溫
        type: 2
        value: 正常，不發熱
    - default:
        title: 今日是否有咳嗽、呼吸困難、腹瀉癥狀
        type: 2
        value: 否，無上述癥狀
    - default:
        title: 今日是否被醫學隔離觀察
        type: 2
        value: 否，未隔離
    - default:
        title: 簡述今日主要行程
        type: 1
        value: 校內
    - default:
        title: 今日晚間回傳宿舍（家）時間
        type: 1
        value: 2020-12-19/22:00

Info:
  ServerChan: # 填寫Server醬的SCKEY
  Qsmg: # 填寫Qsmg醬的SCKEY
  Email:
    enable: true
    server: smtp.exmail.qq.com # 填寫郵件的smtp服務器
    port: 465 # 填寫郵件服務器的埠號
    account: '' # 郵件服務器登錄用戶名
    password: '' # 郵件服務器登錄密碼

generate.py

功能：生成config.yml，今日校園問題的配置
如學校為hfut，不用該步驟
如其他學校，需要輸入自己學校今日校園用戶名和密碼資訊

# -*- 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

功能：自動打卡，需要配置好config.yml

# -*- 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):
    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': '1wAXD2TvR72sQ8u+0Dw8Dr1Qo1jhbem8Nr+LOE6xdiqxKKuj5sXbDTrOWcaf v1X35UtZdUfxokyuIKD4mPPw5LwwsQXbVZ0Q+sXnuKEpPOtk2KDzQoQ89KVs gslxPICKmyfvEpl58eloAZSZpaLc3ifgciGw+PIdB6vOsm2H6KSbwD8FpjY3 3Tprn2s5jeHOp/3GcSdmiFLYwYXjBt7pwgd/ERR3HiBfCgGGTclquQz+tgjJ PdnDjA==',
        '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}
    # 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 = '今日校園疫結果通知'

# 發送郵件通知
def sendMessage(send, msg):
    if send != '':
        log('正在發送郵件通知，，，')
        res = requests.post(url='http://www.zimo.wiki:8080/mail-sender/sendMail',
                            data={'title': title_text, 'content': getTimeStr() + str(msg), 'to': send})

        code = res.json()['code']
        if code == 0:
            log('發送郵件通知成功，，，')
        else:
            log('發送郵件通知失敗，，，')
            log(res.json())

def sendEmail(send,msg):
    my_sender= config['Info']['Email']['account']   # 發件人郵箱賬號
    my_pass = config['Info']['Email']['password']         # 發件人郵箱密碼
    my_user = send      # 收件人郵箱賬號，我這邊發送給自己
    try:
        msg=MIMEText(getTimeStr() + str(msg),'plain','utf-8')
        msg['From']=formataddr(["FromRunoob",my_sender])  # 括號里的對應發件人郵箱昵稱、發件人郵箱賬號
        msg['To']=formataddr(["FK",my_user])              # 括號里的對應收件人郵箱昵稱、收件人郵箱賬號
        msg['Subject']=title_text               # 郵件的主題，也可以說是標題

        server=smtplib.SMTP_SSL(config['Info']['Email']['server'], config['Info']['Email']['port'])  # 發件人郵箱中的SMTP服務器，埠是25
        server.login(my_sender, my_pass)  # 括號中對應的是發件人郵箱賬號、郵箱密碼
        server.sendmail(my_sender,[my_user,],msg.as_string())  # 括號中對應的是發件人郵箱賬號、收件人郵箱賬號、發送郵件
        server.quit()  # 關閉連接
    except Exception:  # 如果 try 中的陳述句沒有執行，則會執行下面的 ret=False
        log("郵件發送失敗")
    else: print("郵件發送成功")

# server醬通知
def sendServerChan(msg):
    log('正在發送Server醬，，，')
    res = requests.post(url='https://sc.ftqq.com/{0}.send'.format(config['Info']['ServerChan']),
                            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醬，，，')
    res = requests.post(url='https://qmsg.zendee.cn:443/send/{0}'.format(config['Info']['Qsmg']),
                            data={'msg': title_text + '\n時間：' + getTimeStr() + "\n 回傳結果：" + str(msg)})
    code = res.json()['success']
    if code:
        log('發送Qmsg醬通知成功，，，')
    else:
        log('發送Qmsg醬通知失敗，，，')
        log('Qmsg醬回傳結果'+code)

# 綜合提交
def InfoSubmit(msg, send=None):
    if(None != send):
        if(config['Info']['Email']['enable']): sendEmail(send,msg)
        else: sendMessage(send, msg)
    if(config['Info']['ServerChan']): sendServerChan(msg)
    if(config['Info']['Qsmg']): sendQmsgChan(msg)


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('獲取最新待填寫問卷失敗，可能是輔導員還沒有發布，，，')
                    InfoSubmit('沒有新問卷')
                    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('自動提交成功！')
                    InfoSubmit('自動提交成功！', user['user']['email'])
                elif msg == '該收集已填寫無需再次填寫':
                    log('今日已提交！')
                    # InfoSubmit('今日已提交！', user['user']['email'])
                    InfoSubmit('今日已提交！')
                else:
                    log('自動提交失敗，，，')
                    log('錯誤是' + msg)
                    InfoSubmit('自動提交失敗！錯誤是' + msg, user['user']['email'])
                    exit(-1)
            else:
                log('模擬登陸失敗，，，')
                log('原因可能是學號或密碼錯誤，請檢查配置后，重啟腳本，，，')
                exit(-1)
    except Exception as e:
        InfoSubmit("出現問題了！"+str(e))
        raise e
    else:
        return 'success'


# 配合Windows計劃任務等使用
if __name__ == '__main__':
    print(main_handler({}, {}))
    # for user in config['users']:
    #     log(getCpdailyApis(user))

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/238556.html

標籤：python

上一篇：python學習總結part2

下一篇：用Python做一個超簡單的小游戲（一聽就懂）

自動填寫今日校園

騰訊云函式實作今日校園自動填寫

說明

工具

注意

使用步驟

附件

dependency.zip

config.yml

generate.py

index.py