萬物皆可Serverless系列文章
- 萬物皆可Serverless之免費搭建自己的不限速大容量雲盤（5TB）
- 萬物皆可Serverless之使用雲函數Timer觸發器實作每天自動定時打卡
- 萬物皆可Serverless之使用SCF+COS快速開發全棧應用
- 萬物皆可Serverless之使用SCF+COS免費營運微信公衆号
- 萬物皆可Serverless之使用SCF快速部署驗證碼識别接口
- 萬物皆可Serverless之Kaggle+SCF端到端驗證碼識别從訓練到部署
- 萬物皆可Serverless之借助微信公衆号簡單管理使用者激活碼
- 萬物皆可Serverless之使用SCF+COS給未來寫封信
- 萬物皆可Serverless之在Flutter中快速接入騰訊雲開發
- 萬物皆可Serverless之在Flutter中寫一個Dart原生騰訊雲對象存儲插件
- 萬物皆可Serverless之我的Serverless之路
一、本文介紹
不曉得大家有沒有遇到過定時打卡的需求，
比如商品秒殺，火車票定時開售、每日健康打卡等
這時候我們往往可以通過一些技術手段，
編寫一些自動化操作的腳本，
來實作定時自動打卡的操作。
當然本文并不探讨如何編寫自動化的操作腳本，
而是和大家介紹一下如何使用騰訊雲函數的Timer觸發器實作定時任務，
來快速、穩定、低成本地實作一些 fancy 的操作（騷操作）
廢話少說，上圖?
每日健康資訊自動更新
每日定時資料報告
可以看到，定時任務搭配郵箱發送雲函數運作結果，用起來還是蠻舒服的，
還可以給自己做一個每日科技資訊推送、資料報告之類的小玩意，自娛自樂
其他的用途請大家大開腦洞，自行腦補吧~
OK，話不多說，上教程
二、操作步驟
第一步：建立雲函數
建立函數
運作環境我們選擇python3，模闆函數選擇定時撥測，然後點選下一步
定時撥測模闆函數
模闆函數的描述裡寫着“本示例代碼的功能是定時撥測 URL 清單中的位址，并通過郵件發送告警”
而這正是我們想要的實作的功能，不過這個模闆函數的郵件發送有點問題，我們稍後會詳細說明
第二步：模闆函數分析
下面我們來分析一下這段示例代碼
# -*- 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 = "[email protected]" # 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 = {
"[email protected]",
"[email protected]"
}
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 = []
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 = []
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表達式的具體用法可翻閱騰訊雲官方文檔或自行百度。
那麼以上就是整個撥測示例雲函數的工作流程
有了這個示例，下面就讓我們來照葫蘆畫瓢編寫自己的雲函數吧
第三步：請求資料分析
喜聞樂見的抓包環節，看看打卡的時候時手機應用都和伺服器交流了些啥
應用資料首頁
點進去看一下
登陸應用
OK，這裡我們已經看到了應用的登入過程，
這裡送出了username，password和type三個參數
分别對應我們的使用者名，登陸密碼和使用者類型，
後面我們隻需要把這些資料重新發送給伺服器就可以模拟登陸APP了
送出資訊
這裡就是向伺服器發送我們填寫的健康資訊
一會我們再把這些資訊一股腦再重新抛給伺服器就好了
第四步：編寫雲函數
根據上面的分析，直接上代碼
def myHealth(user, pwd, email):
errorinfo = []
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 複制
這樣就可以正常發送郵件了
第四步：設定觸發器
OK，我們把修改好的雲函數儲存一下
配置函數
然後把記憶體改到64mb，逾時時間給個3s即可
添加觸發器
最後添加定時觸發器，這裡我們選擇自定義觸發周期
Cron表達式 “0 0 6 * * * * ” 代表每天早上6點觸發一次
注意千萬不要寫成 “* * 6 * * * * ”，
不然将會在每天的6-7點内每秒觸發一次。。。
這樣的話就，，，畫面太美不敢想象，哈哈哈
三、文章最後
以上，想必現在你已經get了如何使用Timer觸發器來觸發雲函數了
何不趕快自己動手嘗試一下呢？
發揮你的想象力，試着做些有趣又有用的小東西吧
本文僅供學習交流之用途，不要學來幹壞事哦~