【玩转腾讯云】万物皆可Serverless之使用云函数Timer触发器实现每天自动定时打卡

万物皆可Serverless系列文章

  1. 万物皆可Serverless之免费搭建自己的不限速大容量云盘(5TB)
  2. 万物皆可Serverless之使用云函数Timer触发器实现每天自动定时打卡
  3. 万物皆可Serverless之使用SCF+COS快速开发全栈应用
  4. 万物皆可Serverless之使用SCF+COS免费运营微信公众号
  5. 万物皆可Serverless之使用SCF快速部署验证码识别接口
  6. 万物皆可Serverless之Kaggle+SCF端到端验证码识别从训练到部署
  7. 万物皆可Serverless之借助微信公众号简单管理用户激活码
  8. 万物皆可Serverless之使用SCF+COS给未来写封信
  9. 万物皆可Serverless之在Flutter中快速接入腾讯云开发
  10. 万物皆可Serverless之在Flutter中写一个Dart原生腾讯云对象存储插件
  11. 万物皆可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 = "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 = []
    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触发器来触发云函数了

何不赶快自己动手尝试一下呢?

发挥你的想象力,试着做些有趣又有用的小东西吧

本文仅供学习交流之用途,不要学来干坏事哦~

本站文章资源均来源自网络,除非特别声明,否则均不代表站方观点,并仅供查阅,不作为任何参考依据!
如有侵权请及时跟我们联系,本站将及时删除!
如遇版权问题,请查看 本站版权声明
THE END
分享
二维码
海报
【玩转腾讯云】万物皆可Serverless之使用云函数Timer触发器实现每天自动定时打卡
模板函数的描述里写着“本示例代码的功能是定时拨测 URL 列表中的地址,并通过邮件发送告警”
<<上一篇
下一篇>>