不曉得大家有沒有遇到過定時打卡的需求，比如商品秒殺，火車票定時開售、每日健康打卡等，這時候我們往往可以通過一些技術手段，撰寫一些自動化操作的腳本，來實作定時自動打卡的操作，
本文來自 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 = "[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 = [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 = https://www.cnblogs.com/serverlesscloud/p/{ # 登陸資訊'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 = https://www.cnblogs.com/serverlesscloud/p/{ # 健康資訊
"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！
詳情可查閱：Serverless Framework 試用計劃
One More Thing
3 秒你能做什么？喝一口水，看一封郵件，還是 —— 部署一個完整的 Serverless 應用？
復制鏈接至 PC 瀏覽器訪問：https://serverless.cloud.tencent.com/deploy/express
3 秒極速部署，立即體驗史上最快的 Serverless HTTP 實戰開發！
傳送門：
- GitHub: github.com/serverless
- 官網：serverless.com
歡迎訪問：Serverless 中文網，您可以在 最佳實踐 里體驗更多關于 Serverless 應用的開發！
推薦閱讀：《Serverless 架構：從原理、設計到專案實戰》
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/6767.html
標籤：其他