不曉得大家有沒有遇到過定時打卡的需求,比如商品秒殺,火車票定時開售、每日健康打卡等。這時候我們往往可以通過一些技術手段,編寫一些自動化操作的腳本,來實現定時自動打卡的操作。
本文來自 Serverless 社區用戶「乂乂又又」供稿
當然本文並不探討如何編寫自動化的操作腳本,而是和大家介紹一下如何使用騰訊雲函數的 Timer 觸發器實現定時任務,來快速、穩定、低成本地實現一些 fancy 的操作(騷操作)
效果展示
- 每日健康信息自動更新

- 每日定時數據報告

可以看到,定時任務搭配郵箱發送雲函數運行結果,用起來還是蠻舒服的,還可以給自己做一個每日科技資訊推送、數據報告之類的小玩意,自娛自樂。其他的用途請大家大開腦洞,自行腦補吧~
實戰教程
1. 新建雲函數

運行環境我們選擇 python3,模板函數選擇定時撥測,然后點擊下一步

模板函數的描述里寫着「本示例代碼的功能是定時撥測 URL 列表中的地址,並通過郵件發送告警」
而這正是我們想要的實現的功能,不過這個模板函數的郵件發送有點問題,我們稍后會詳細說明
2. 模板函數分析
下面我們來分析一下這段示例代碼
# -*- 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 服務,用於發送告警郵件
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("", "")
這里要講一下雲函數的執行入口,
這個模板函數的默認入口是 main\_handler(event, context) 這個函數,
這個入口函數是可以自行配置的,具體配置方法可以翻看官方的文檔
def main_handler(event, context):
test_url(test_url_list)
另外這里的 py 文件的主函數入口,實際上是可以缺省的。這里加上應該是為了方便本地調試和運行函數。
if __name__ == '__main__':
main_handler("", "")
然后看一下依賴庫的導入部分
import requests
from email.mime.text import MIMEText
from email.header import Header
import smtplib
注意到有 import requests,但本地文件並沒有 requests 庫,說明騰訊雲函數的運行環境中已經安裝了 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 函數的思路非常清晰,首先請求 url\_list 內的目標網頁,如果請求超時或者出現錯誤碼就會記錄下 errorinfo。
當 errorinfo 列表非空時,也就是有鏈接的訪問出現問題時就會調用 sendEmail 函數
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 函數負責登錄郵箱並發送 errorinfo 郵件提醒
smtpObj = smtplib.SMTP_SSL(mail_host, mail_port)
smtpObj.login(mail_user, mail_pass)
smtpObj.sendmail(sender, receivers, message.as_string())
下面我們再看一下雲函數的配置文件

注意圖中畫紅圈的部分
"CronExpression": "* */1 * * * * *",
這是 Cron 表達式,用來描述定時任務開始執行時間用的,這里的 \* \*/1 \* \* \* \* \* 表示每分鍾執行一次雲函數,以達到網站監控撥測的功能。有關 Cron 表達式的具體用法可翻閱騰訊雲官方文檔。
以上就是整個撥測示例雲函數的工作流程。下面就讓我們來照葫蘆畫瓢編寫自己的雲函數吧。
3. 請求數據分析
喜聞樂見的抓包環節,看看打卡的時候時手機應用都和服務器交流了些啥

點進去看一下

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

這里就是向服務器發送我們填寫的健康信息,一會我們再把這些信息一股腦再重新拋給服務器就好了
4. 編寫雲函數
根據上面的分析,直接上代碼
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('提交健康信息失敗:'+r.json()['message'])
s.close() # 關閉連接
emailTask(errorinfo, email) # 發送郵件
嗯,替換一下模板函數里面的 test\_url 函數就 ok 了
不過前面我有提到郵件發送有問題,下面我們來看下 sendemai 函數里郵件內容編碼部分
message['From'] = Header(fromAddr, 'utf-8')
message['To'] = Header(toAddr, 'utf-8')
message['Subject'] = Header(subject, 'utf-8')
這里的收件人,發件人和主題信息都經過了 Header(string, 'utf-8') 來編碼。不過在我用 163 郵箱發信時,這種方法只能自己給自己的郵箱發郵件,給別人發會被郵件系統當成垃圾郵件發送失敗。所以如果你需要給其他郵箱發郵件的話,這里需要去掉編碼,改成
message['From'] = fromAddr
message['To'] = toAddr
message['Subject'] = subject
這樣就可以正常發送郵件了
5. 設置觸發器
OK,我們把修改好的雲函數保存一下

然后把內存改到 64mb,超時時間給個 3s 即可

最后添加定時觸發器,這里我們選擇自定義觸發周期。
Cron 表達式 0 0 6 \* \* \* \* 代表每天早上 6 點觸發一次,注意千萬不要寫成 \* \* 6 \* \* \* \* ,不然將會在每天的 6-7 點內每秒觸發一次。這樣的話就,畫面太美不敢想象,哈哈哈 ~
寫在最后
以上,想必現在你已經 get 了如何使用 Timer 觸發器來觸發雲函數了,何不趕快自己動手嘗試一下呢?
發揮你的想象力,試着做些有趣又有用的小東西吧!本文僅供學習交流之用途,不要學來干壞事哦~
Serverless Framework 30 天試用計划
我們誠邀您來體驗最便捷的 Serverless 開發和部署方式。在試用期內,相關聯的產品及服務均提供免費資源和專業的技術支持,幫助您的業務快速、便捷地實現 Serverless!
One More Thing
3 秒你能做什么?喝一口水,看一封郵件,還是 —— 部署一個完整的 Serverless 應用?
復制鏈接至 PC 瀏覽器訪問:https://serverless.cloud.tencent.com/deploy/express
3 秒極速部署,立即體驗史上最快的 Serverless HTTP 實戰開發!
傳送門:
- GitHub: github.com/serverless
- 官網:serverless.com
歡迎訪問:Serverless 中文網,您可以在 最佳實踐 里體驗更多關於 Serverless 應用的開發!
