如何基于Serverless使用云函數(shù)Timer觸發(fā)器實現(xiàn)每天自動定時打卡

這篇文章給大家介紹如何基于Serverless使用云函數(shù)Timer觸發(fā)器實現(xiàn)每天自動定時打卡，內(nèi)容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

不曉得大家有沒有遇到過定時打卡的需求，比如商品秒殺，火車票定時開售、每日健康打卡等。這時候我們往往可以通過一些技術手段，編寫一些自動化操作的腳本，來實現(xiàn)定時自動打卡的操作。

當然并不探討如何編寫自動化的操作腳本，而是和大家介紹一下如何使用騰訊云函數(shù)的 Timer 觸發(fā)器實現(xiàn)定時任務，來快速、穩(wěn)定、低成本地實現(xiàn)一些 fancy 的操作（騷操作）

效果展示

• 每日健康信息自動更新

• 每日定時數(shù)據(jù)報告

可以看到，定時任務搭配郵箱發(fā)送云函數(shù)運行結果，用起來還是蠻舒服的，還可以給自己做一個每日科技資訊推送、數(shù)據(jù)報告之類的小玩意，自娛自樂。其他的用途請大家大開腦洞，自行腦補吧~

實戰(zhàn)教程

1. 新建云函數(shù)

運行環(huán)境我們選擇 python3，模板函數(shù)選擇定時撥測，然后點擊下一步

模板函數(shù)的描述里寫著「本示例代碼的功能是定時撥測 URL 列表中的地址，并通過郵件發(fā)送告警」

而這正是我們想要的實現(xiàn)的功能，不過這個模板函數(shù)的郵件發(fā)送有點問題，我們稍后會詳細說明

2. 模板函數(shù)分析

下面我們來分析一下這段示例代碼

# -*- coding: utf8 -*-
import sys
import os

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)) + "/..")

import logging
import json
import requests
from email.mime.text import MIMEText
from email.header import Header
import smtplib

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

# Third-party SMTP service for sending alert emails. 第三方 SMTP 服務，用于發(fā)送告警郵件
mail_host = "smtp.qq.com"       # SMTP server, such as QQ mailbox, need to open SMTP service in the account. SMTP服務器,如QQ郵箱，需要在賬戶里開啟SMTP服務
mail_user = "XXXXXXXXX@qq.com"  # Username 用戶名
mail_pass = "****************"  # Password, SMTP service password. 口令，SMTP服務密碼
mail_port = 465  # SMTP service port. SMTP服務端口

# The URL address need to dial test. 需要撥測的URL地址
test_url_list = [
    "http://www.baidu.com",
    "http://www.qq.com",
    "http://wrong.tencent.com",
    "http://unkownurl.com"
]

# The notification list of alert emails. 告警郵件通知列表
email_notify_list = {
    "XXXXXXXXX@qq.com",
    "XXXXXXXXX@qq.com"
}


def sendEmail(fromAddr, toAddr, subject, content):
    sender = fromAddr
    receivers = [toAddr]
    message = MIMEText(content, 'plain', 'utf-8')
    message['From'] = Header(fromAddr, 'utf-8')
    message['To'] = Header(toAddr, 'utf-8')
    message['Subject'] = Header(subject, 'utf-8')
    try:
        smtpObj = smtplib.SMTP_SSL(mail_host, mail_port)
        smtpObj.login(mail_user, mail_pass)
        smtpObj.sendmail(sender, receivers, message.as_string())
        print("send email success")
        return True
    except smtplib.SMTPException as e:
        print(e)
        print("Error: send email fail")
        return False


def test_url(url_list):
    errorinfo = [serverless]
    for url in url_list:
        resp = None
        try:
            resp = requests.get(url, timeout=3)
            print (resp)
        except (
        requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.ConnectTimeout) as e:
            logger.warn("request exceptions:" + str(e))
            errorinfo.append("Access " + url + " timeout")
        else:
            if resp.status_code >= 400:
                logger.warn("response status code fail:" + str(resp.status_code))
                errorinfo.append("Access " + url + " fail, status code:" + str(resp.status_code))
    if len(errorinfo) != 0:
        body = "\r\n".join(errorinfo)
        subject = "Please note: PlayCheck Error"
        for toAddr in email_notify_list:
            print ("send message [%s] to [%s]" % (body, toAddr))
            sendEmail(mail_user, toAddr, subject, body)


def main_handler(event, context):
    test_url(test_url_list)


if __name__ == '__main__':
    main_handler("", "")

這里要講一下云函數(shù)的執(zhí)行入口，

這個模板函數(shù)的默認入口是 main\_handler(event, context) 這個函數(shù)，

這個入口函數(shù)是可以自行配置的，具體配置方法可以翻看官方的文檔

def main_handler(event, context):
    test_url(test_url_list)

另外這里的 py 文件的主函數(shù)入口，實際上是可以缺省的。這里加上應該是為了方便本地調(diào)試和運行函數(shù)。

if __name__ == '__main__':
    main_handler("", "")

然后看一下依賴庫的導入部分

import requests
from email.mime.text import MIMEText
from email.header import Header
import smtplib

注意到有 import requests，但本地文件并沒有 requests 庫，說明騰訊云函數(shù)的運行環(huán)境中已經(jīng)安裝了 requests 庫，并不需要我們再手動上傳添加 requests 依賴。

def test_url(url_list):
    errorinfo = [serverless]
    for url in url_list:
        resp = None
        try:
            resp = requests.get(url, timeout=3)
            print (resp)
        except (
        requests.exceptions.Timeout, requests.exceptions.ConnectionError, requests.exceptions.ConnectTimeout) as e:
            logger.warn("request exceptions:" + str(e))
            errorinfo.append("Access " + url + " timeout")
        else:
            if resp.status_code >= 400:
                logger.warn("response status code fail:" + str(resp.status_code))
                errorinfo.append("Access " + url + " fail, status code:" + str(resp.status_code))
    if len(errorinfo) != 0:
        body = "\r\n".join(errorinfo)
        subject = "Please note: PlayCheck Error"
        for toAddr in email_notify_list:
            print ("send message [%s] to [%s]" % (body, toAddr))
            sendEmail(mail_user, toAddr, subject, body)

這里的 test\_url 函數(shù)的思路非常清晰，首先請求 url\_list 內(nèi)的目標網(wǎng)頁，如果請求超時或者出現(xiàn)錯誤碼就會記錄下 errorinfo。

當 errorinfo 列表非空時，也就是有鏈接的訪問出現(xiàn)問題時就會調(diào)用 sendEmail 函數(shù)

def sendEmail(fromAddr, toAddr, subject, content):
    sender = fromAddr
    receivers = [toAddr]
    message = MIMEText(content, 'plain', 'utf-8')
    message['From'] = Header(fromAddr, 'utf-8')
    message['To'] = Header(toAddr, 'utf-8')
    message['Subject'] = Header(subject, 'utf-8')
    try:
        smtpObj = smtplib.SMTP_SSL(mail_host, mail_port)
        smtpObj.login(mail_user, mail_pass)
        smtpObj.sendmail(sender, receivers, message.as_string())
        print("send email success")
        return True
    except smtplib.SMTPException as e:
        print(e)
        print("Error: send email fail")
        return False

sendEmail 函數(shù)負責登錄郵箱并發(fā)送 errorinfo 郵件提醒

smtpObj = smtplib.SMTP_SSL(mail_host, mail_port)
smtpObj.login(mail_user, mail_pass)
smtpObj.sendmail(sender, receivers, message.as_string())

下面我們再看一下云函數(shù)的配置文件

注意圖中畫紅圈的部分

"CronExpression": "* */1 * * * * *",

這是 Cron 表達式，用來描述定時任務開始執(zhí)行時間用的，這里的 \* \*/1 \* \* \* \* \* 表示每分鐘執(zhí)行一次云函數(shù)，以達到網(wǎng)站監(jiān)控撥測的功能。有關 Cron 表達式的具體用法可翻閱騰訊云官方文檔。

以上就是整個撥測示例云函數(shù)的工作流程。下面就讓我們來照葫蘆畫瓢編寫自己的云函數(shù)吧。

3. 請求數(shù)據(jù)分析

喜聞樂見的抓包環(huán)節(jié)，看看打卡的時候時手機應用都和服務器交流了些啥

點進去看一下

OK，這里我們已經(jīng)看到了應用的登錄過程，這里提交了 username，password 和 type 三個參數(shù)，分別對應我們的用戶名，登陸密碼和用戶類型，后面我們只需要把這些數(shù)據(jù)重新發(fā)送給服務器就可以模擬登陸 App 了

這里就是向服務器發(fā)送我們填寫的健康信息，一會我們再把這些信息一股腦再重新拋給服務器就好了

4. 編寫云函數(shù)

根據(jù)上面的分析，直接上代碼

def myHealth(user, pwd, email):
    errorinfo = [serverless]
    s = requests.Session()  # 新建一個request對象
    data = {  # 登陸信息
        'username': user,
        'password': pwd,
        'type': 'student',
    }
    r = s.post(server+'/authentication/login', json=data)  # 登錄
    if r.json()['ok']:
        errorinfo.append('登陸成功')
    else:
        s.close()
        errorinfo.append('登陸失敗')
        return
    data = {  # 健康信息
        "home": "在家",
        "address": "",
        "keepInHome": "否",
        "keepInHomeDate": 'null',
        "contact": "否",
        "health": "否",
        "familyNCP": "否",
        "fever": "否",
        "feverValue": "",
        "cough": "否",
        "diarrhea": "否",
        "homeInHubei": "否",
        "arriveHubei": "無",
        "travel": "無",
        "remark": "無",
        "submitCount": '0'
    }
    r = s.post(server+'/student/healthInfo/save', json=data)  # 提交健康信息
    if r.json()['ok']:
        errorinfo.append('提交健康信息成功')
    else:
        errorinfo.append('提交健康信息失?。?#39;+r.json()['message'])
    s.close()  # 關閉連接
    emailTask(errorinfo, email)  # 發(fā)送郵件

嗯，替換一下模板函數(shù)里面的 test\_url 函數(shù)就 ok 了

不過前面我有提到郵件發(fā)送有問題，下面我們來看下 sendemai 函數(shù)里郵件內(nèi)容編碼部分

    message['From'] = Header(fromAddr, 'utf-8')
    message['To'] = Header(toAddr, 'utf-8')
    message['Subject'] = Header(subject, 'utf-8')

這里的收件人，發(fā)件人和主題信息都經(jīng)過了 Header(string, 'utf-8') 來編碼。不過在我用 163 郵箱發(fā)信時，這種方法只能自己給自己的郵箱發(fā)郵件，給別人發(fā)會被郵件系統(tǒng)當成垃圾郵件發(fā)送失敗。所以如果你需要給其他郵箱發(fā)郵件的話，這里需要去掉編碼，改成

    message['From'] = fromAddr
    message['To'] = toAddr
    message['Subject'] = subject

這樣就可以正常發(fā)送郵件了

5. 設置觸發(fā)器

OK，我們把修改好的云函數(shù)保存一下

然后把內(nèi)存改到 64mb，超時時間給個 3s 即可

最后添加定時觸發(fā)器，這里我們選擇自定義觸發(fā)周期。

Cron 表達式 0 0 6 \* \* \* \* 代表每天早上 6 點觸發(fā)一次，注意千萬不要寫成 \* \* 6 \* \* \* \* ，不然將會在每天的 6-7 點內(nèi)每秒觸發(fā)一次。這樣的話就，畫面太美不敢想象，哈哈哈 ~

關于如何基于Serverless使用云函數(shù)Timer觸發(fā)器實現(xiàn)每天自動定時打卡就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。