Post

Python 搭配 Google Cloud Platform 与 Line Bot|打造每日自动签到脚本与排程

透过 Python 结合 Google Cloud Function、Cloud Scheduler 及 Line Bot,自动完成每日签到任务并即时通知,解决手动操作繁琐问题,提升工作效率并节省时间成本。

Python 搭配 Google Cloud Platform 与 Line Bot|打造每日自动签到脚本与排程

Click here to view the English version of this article.

點擊這裡查看本文章正體中文版本。

基于 SEO 考量,本文标题与描述经 AI 调整,原始版本请参考内文。


使用 Python+Google Cloud Platform+Line Bot 自动执行例行琐事

以签到奖励 APP 为例,打造每日自动签到脚本

Photo by [Paweł Czerwiński](https://unsplash.com/@pawel_czerwinski?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText){:target="_blank"}

Photo by Paweł Czerwiński

起源

一直以来都有使用 Python 做小工具的习惯;有做正经的,工作上自动爬数据、产报表,也有不正经的,排程自动查想要的资讯或是交给脚本完成本来要手动执行的动作。

一直以来「自动」这件事,我都很粗暴直接开一台电脑挂著 Python 脚本让他挂著跑;优点是简单方便,缺点是要有台设备接著网路接著电;就算是树莓派也是要消耗著微量的电费网路钱,还有也不能远端控制启动或关闭(其实可以,但很麻烦);这次趁著工作空挡,研究了一下免费&上云端的方法。

目标

将 Python 脚本搬到云端执行、定时自动执行、可透过网路开启/关闭。

本篇以我耍的小聪明,针对签到奖励型 APP 撰写的自动完成签到的脚本为例,能每日自动帮我签到,我不用在特别打开 APP 使用;并在执行完成后发通知给我。

完成通知!

完成通知!

本篇章节顺序

  1. 使用 Proxyman 进行 Man in the middle attack API 嗅探

  2. 撰写 Python 脚本,伪造 APP API 请求(模拟签到动作)

  3. 将 Python 脚本搬到 Google Cloud 上

  4. 在 Google Cloud 设定自动排程

  • 因涉及到敏感领域本篇不会告知是哪个签到奖励型 APP,大家可以延伸自行使用

  • 如果只想了解 Python 怎么串自动执行可跳过前半段 Man in the middle attack API 嗅探部分,从第 3 章看起

使用到的工具

  • Proxyman :Man in the middle attack API 嗅探

  • Python :撰写脚本

  • Linebot :发送脚本执行结果通知给自己

  • Google Cloud Function :Python 脚本寄存服务

  • Google Cloud Scheduler :自动排程服务

1.使用 Proxyman 进行 Man in the middle attack API 嗅探

之前有发过一篇「 APP有用HTTPS传输,但资料还是被偷了。 」的文章,道理类似,不过这次改用 Proxyman 取代 mitmproxy;同样免费,但更好用。

  • 到官网 https://proxyman.io/ 下载 Proxyman 工具

  • 下载完后启动 Proxyman,安装 Root 凭证(为了做 Man in the middle attack 解包 https 流量内容)

「Certificate 」->「 Install Certificate On this Mac」->「Installed & Trusted」

电脑的 Root 凭证装好后换手机的:

「Certificate 」->「 Install Certificate On iOS」->「Physical Devices…」

依照指示在手机上挂好 Proxy 并完成凭证安装及启用。

  • 在手机上打开想要嗅探 API 传输内容的 APP

这时候 Mac 上的 Proxyman 就会出现嗅探到的流量,点击装置 IP 下想要查看的 APP API 网域;第一次查看需要先点「Enable only this domain」之后的流量才能被解包出来。

「Enable only this domain」后就能看到新拦截的流量就会出现原始的 Request、Response 资讯:

我们使用此方法嗅探 APP 上操作签到时打了哪只 API EndPoint 及带了哪些资料,将这些资讯记录下来,等下使用 Python 直接模拟请求。

⚠️要注意有的 APP token 资讯可能会换,导致日后 Python 模拟请求失效,还要多了解 APP token 交换的方式。

⚠️如果确定 Proxyman 有正常运作,但在挂 Proxyman 的情况下 APP 无法发出请求,代表 APP 可能有做 SSL Pining;目前无解,只能放弃。

⚠️APP 开发者想知道怎么防范嗅探可参考 之前的文章

这边假设我们得到的资讯如下:

1
2
3
4
5
6
7
8
9
10
11
POST /usercenter HTTP/1.1
Host: zhgchg.li
Content-Type: application/x-www-form-urlencoded
Cookie: PHPSESSID=dafd27784f94904dd586d4ca19d8ae62
Connection: keep-alive
Accept: */*
User-Agent: (iPhone12,3;iOS 14.5)
Content-Length: 1076
Accept-Language: zh-tw
Accept-Encoding: gzip, deflate, br
AuthToken: 12345

2. 撰写 Python 脚本,伪造 APP API 请求(模拟签到动作)

在撰写 Python 脚本之前,我们可先使用 Postman 调试一下参数,观察看看哪个参数是必要的或是有时效会改变;但要直接照搬也可以。

checkIn.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import requests
import json

def main(args):
    results = {}
    try:
      data = { "action" : "checkIn" }
      headers = { "Cookie" : "PHPSESSID=dafd27784f94904dd586d4ca19d8ae62", 
      "AuthToken" : "12345",
      "User-Agent" : "(iPhone12,3;iOS 14.5)"
      }
      
      request = requests.post('https://zhgchg.li/usercenter', data = data, headers = headers)
      result = json.loads(request.content)
      if result['status_code'] == 200:
        return "CheckIn Success!"
      else:
        return result['message']
    except Exception as e:
      return str(e)

⚠️ main(args) 这边的 args 用途后面会讲,如果要在本地测试直接带 main(True) 就好。

使用 Requests 套件帮我们执行 HTTP Request,如果出现:

1
ImportError: No module named requests

请先使用 pip install requests 安装套件。

加上执行结果 Linebot 通知:

这部分我做的很简单,仅共参考,仅通知自己。

  • 选择「Create a Messaging API channel」

下一步填好基本讯息后按「Create」送出建立。

  • 建立好之后在第一个「Basic settings」Tab 下面找到「Your user ID」区块,这就是你的 User ID

  • 建立好之后,选择「Messaging API」Tab,扫描 QRCode 将机器人加入好友。

  • 继续往下滚找到「Channel access token」区块,点击「Issue」产生 token。

  • 复制下来产生出来的 Token,我们有这组 Token 就能发讯息给使用者。

有了 User Id 跟 Token 之后我们就能发讯息给自己了。

因没有要做其他功能所以连 python line sdk 都不用装,直接打 http 发。

串上之前的 Python 脚本后…

checkIn.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import requests
import json

def main(args):
    results = {}
    try:
      data = { "action" : "checkIn" }
      headers = { "Cookie" : "PHPSESSID=dafd27784f94904dd586d4ca19d8ae62", 
      "AuthToken" : "12345",
      "User-Agent" : "(iPhone12,3;iOS 14.5)"
      }
      
      request = requests.post('https://zhgchg.li/usercenter', data = data, headers = headers)
      result = json.loads(request.content)
      if result['status_code'] == 200:
        sendLineNotification("CheckIn Success!")
        return "CheckIn Success!"
      else:
        sendLineNotification(result['message'])
        return result['message']
    except Exception as e:
      sendLineNotification(str(e))
      return str(e)
      
def sendLineNotification(message):
    data = {
        "to" : "这边带你的 User ID",
        "messages" : [
            {
                "type" : "text",
                "text" : message
            }
        ]
    }
    headers = {
        "Content-Type" : "application/json",
        "Authorization" : "这边带channel access token"
    }
    request = requests.post('https://api.line.me/v2/bot/message/push',json = data, headers = headers)

测看看通知有没有发成功:

Success!

小插曲,通知部分我本来是想用 Gmail SMTP 用信件来发,结果上到 Google Cloud 后发现无法使用…

3. 将 Python 脚本搬到 Google Cloud 上

前面基本的讲完了,正式进入本篇重头戏;将 Python 脚本搬上云端。

这部分我一开始向中的是 Google Cloud Run 但用了下觉得太复杂,我实际懒得研究,因为我的需求太小用不到这么多功能;所以 我用的是 Google Cloud Function serverless 方案;实际上比较常用来做的是构建 serverless web 服务。

  • 如果没使用过 Google Cloud 的朋友,请先前往 主控台 新增好专案&设定好帐单资讯

  • 在专案主控台首页,资源的地方点击「Cloud Functions」

  • 上方选择「建立函式」

  • 输入基本资讯

⚠️记下「 触发网址」

区域可选:

  • US-WEST1US-CENTRAL1US-EAST1 可享 Cloud Storage 服务免费额度。

  • asia-east2 (Hong Kong) 靠我们比较近,但需要支付微微的 Cloud Storage 费用。

⚠️建立 Cloud Functions 时会需要 Cloud Storage 寄存程式码。

⚠️详细计价方式请参考文末。

触发条件选: HTTP

验证: 依需求,我希望我能从外部点连结执行脚本,所以选择「允许未经验证的叫用」;如果选择需要验证,后续 Scheduler 服务也要做相应设定。

变数、网路及进阶设定可在变数中设定变数给 Python 使用(这样参数有变动就不用改到 Python 程式码):

在 Python 中调用的方式:

1
2
3
4
import os

def main(request):
  return os.environ.get('test', 'DEFAULT VALUE')

其他设定都不需要动,直接「储存」->「下一步」。

  • 执行阶段选「Python 3.x」并将写好的 Python 脚本贴上,进入点改成「main」

补充 main(args) ,同前述,此项服务比较是用来做 serverless web;所以 args 实际是 Request 物件,你能从其中拿到 http get query 及 http post body 资料,具体方式如下:

1
2
取得 GET Query 资讯:
request_args = args.args

example: ?name=zhgchgli => request_args = [“name”:”zhgchgli”]

1
2
取得 POST Body 资料:
request_json = request.get_json(silent=True)

example: name=zhgchgli => request_json = [“name”:”zhgchgli”]

如果使用 Postman 测试 POST 记得使用「Raw+JSON」POST 资料,否则不会有东西:

  • 程式码部分 OK 之后,切换到「requirements.txt」输入有用到的套件依赖:

我们使用「request」这个套件帮我们打 API,此套件不在原生 Python 库里面;所以我们要在这里加上去:

1
requests>=2.25.1

这边指定版本 ≥ 2.25.1,也可不指定只输入 requests 安装最新版。

  • 都 OK 之后点击「部署」开始部署。

需要花约 1~3 分钟的时间等他部署完成。

  • 部署完成后可由前面记下的「 触发网址 」前去执行查看是否正确运行,或使用「动作」->「测试函式」进行测试

如果出现 500 Internal Server Error 则代表程式有错,可点击名称进入查看「纪录」,在其中找到原因:

1
UnboundLocalError: local variable 'db' referenced before assignment
  • 点击名称进入后也可按「编辑」修改脚本内容

测试没问题就完成了!我们已经顺利将 Python 脚本搬上云端。

补充关于变数部分

依照我们的需求,我们需要能有个地方存放、读取签到 APP 的 token;因为 token 可能会失效;需要重新要求并写入共下次执行时使用。

想要从外部动态传入变数到脚本中有以下方法:

  • [Read Only] 前述所提到的,执行阶段环境变数

  • [Temp] Cloud Functions 有提供一个 /tmp 目录共执行时写入、读取档案,但结束后就会删除,详情请参考 官方文件

  • [Read Only] GET/POST 传送资料

  • [Read Only] 放入附加档案

在程式中使用相对路径 ./ 就能读取到, 仅限读取无法动态修改 ;要修改只能在控制台这修改&重新部署。

想要可以读取、动态修改就需要串接其他 GCP 服务,例如:Cloud SQL、Google Storage、Firebase Cloud Firestore…

  • [Read & Write] 这边我选择的是 Firebase Cloud Firestore 因为目前只有此方案有免费额度使用。

按照 入门步骤 ,建立好 Firebase 专案后;进入 Firebase 后台:

在左方选单列找到「 Cloud Firestore 」->「 新增集合

输入集合 ID。

输入资料内容。

一个集合可以有多个文件,每个文件可以有各自的栏位内容;使用上非常弹性。

在 Python 中使用:

请先到 GCP控制台 -> IAM与管理 -> 服务帐户 ,按照以下步骤下载身份验证私钥文件:

首先选择帐号:

下方「新增金钥」->「建立新的金钥」

选择「JSON」下载档案。

将此 JSON 档案放到同 Python 的专案目录下。

本地开发环境下:

1
pip install --upgrade firebase-admin

安装 firebase-admin 套件。

在 Cloud Functions 上要在 requirements.txt 中多加入 firebase-admin

环境弄好后,可以来读取我们刚刚新增的数据了:

firebase_admin.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore

if not firebase_admin._apps:
  cred = credentials.Certificate('./身份验证.json')
  firebase_admin.initialize_app(cred)
# 因若重复 initialize_app 会报以下错误
# providing an app name as the second argument. In most cases you only need to call initialize_app() once. But if you do want to initialize multiple apps, pass a second argument to initialize_app() to give each app a unique name.
# 所以安全起见在 initialize_app 前先检查是否已 init

db = firestore.client()
ref = db.collection(u'example') //集合名称
stream = ref.stream()
for data in stream:
  print("id:"+data.id+","+data.to_dict())

如果是在 Cloud Functions 上除了可以把 身份验证 JSON 档一起上传上去,也可以在使用时将连接语法改成以下使用:

1
2
3
4
5
6
cred = credentials.ApplicationDefault()
firebase_admin.initialize_app(cred, {
  'projectId': project_id,
})

db = firestore.client()

如果出现 Failed to initialize a certificate credential. ,请检查身份验证 JSON 是否正确。

新增、删除更多操作请参考 官方文件

4. 在 Google Cloud 设定自动排程

有了脚本之后再来是要让他自动执行才能达到我们的最终目标。

  • 输入工作基本资料

执行频率: 同 crontab 输入方式,如果你对 crontab 语法不熟,可以直接使用 crontab.guru 这个神器网站

他能很直白的翻译给你所设定的语法实际意思。(点 next 可查看下次执行时间)

这边我设定 15 1 * * * ,因为签到每天只需要执行一次,设在每日凌晨 1:15 执行。

网址部分: 输入前面记下的「 触发网址

时区: 输入「台湾」,选择台北标准时间

HTTP 方法: 照前面 Python 程式码我们用 Get 就好

如果前面有设「验证」 记得展开「SHOW MORE」进行验证设定。

都填好后 ,按下「 建立 」。

  • 建立成功后可选择「立即执行」测试一下正不正常。

  • 可查看执行结果、上次执行日期

⚠️ 请注意,执行结果「失败」仅针对 web status code 是 400~500 或 python 程式有错误。

大功告成!

我们已达成将例行任务 Python 脚本上传到云端&设定自动排成自动执行的目标。

计价方式

还有一部分很重要,就是计价方式;Google Cloud、Linebot 都不是全免费服务,所以了解收费方式很重要;不然为了一个小小的脚本,付出太多的金钱那不如电脑开著挂著跑哩。

Linebot

参考 官方定价 资讯,一个月 500 则内免费。

Google Cloud Functions

参考 官方定价 资讯,每月有 200 万次叫用、400,000 GB/秒和 200,000 GHz/秒的运算时间、 5 GB 的网际网路输出流量。

Google Firebase Cloud Firestore

参考 官方定价 资讯,有 1 GB 大小容量、每月 10 GB 流量、每天 50,000 次读取、20,000 次写入/删除;轻量使用很够用了!

Google Cloud Scheduler

参考 官方定价 资讯,每个帐号有 3 项免费工作可设定。

对脚本来说以上免费用量就绰绰有余啦!

Google Cloud Storage 有条件免费

东躲西躲,还是躲不掉可能被收费的服务。

Cloud Functions 建立好之后会自动建立两个 Cloud Storage 实体:

如果刚刚 Cloud Functions 选择的是 US-WEST1、US-CENTRAL1 或 US-EAST1 这三个地区则可享有免费使用额度:

我是选择 US-CENTRAL1 没错,可以看到第一个 Cloud Storage 实体的地区是 US-CENTRAL1 没错,但第二个是写 美国多个地区我自已估计这项是会被收费的

参考 官方定价 资讯,依照主机地区不同有不同的价格。

程式码没多大,估计应该就是每个月最低收费 0.0X0 元(?

⚠️以上资讯均为 2021/02/21 时撰写时纪录,实际以当前价格为主,仅共参考。

计价预算控制通知

just in case…假设真的有状况超出免费用量开始计价,我希望能收到通知;避免可能程式错误暴冲造成帐单金额报表却浑然不知。。。

  • 前往 主控台

  • 找到「 计费功能 」Card:

点击「 查看详细扣款纪录 」进入。

  • 展开左边选单,进入「 预算与快讯 」功能

  • 点击上方「 设定预算

  • 输入自订名称

下一步。

  • 金额,输入「 目标金额 」,可输入 $1、$10;我们不希望在小东西上花太。

下一步。

动作这边可以设定当预算达到多少百分比时会触发通知。

勾选透过电子邮件将快讯传送给帐单管理员和使用者 」,这样当条件处发时就能第一时间收到通知。

点击「完成」送出储存。

当预算超过时我们就能马上就能知道,避免产生更多费用。

总结

人的精力是有限的,现今科技资讯洪流,每个平台每个服务都想要榨取我们有限的精力;如果能透过一些自动化脚本分担我们的日常生活,聚沙成塔,让我们省下更多精力专心在重要的事情之上!

延伸阅读

有任何问题及指教欢迎 与我联络

有自动化相关优化需求也欢迎 发案给我 ,谢谢。


Buy me a beer

本文首次发表于 Medium (点击查看原始版本),由 ZMediumToMarkdown 提供自动转换与同步技术。

Improve this page on Github.

This post is licensed under CC BY 4.0 by the author.