Python 搭配 Google Cloud Platform 与 Line Bot|打造每日自动签到脚本与排程
透过 Python 结合 Google Cloud Function、Cloud Scheduler 及 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
起源
一直以来都有使用 Python 做小工具的习惯;有做正经的,工作上自动爬数据、产报表,也有不正经的,排程自动查想要的资讯或是交给脚本完成本来要手动执行的动作。
一直以来「自动」这件事,我都很粗暴直接开一台电脑挂著 Python 脚本让他挂著跑;优点是简单方便,缺点是要有台设备接著网路接著电;就算是树莓派也是要消耗著微量的电费网路钱,还有也不能远端控制启动或关闭(其实可以,但很麻烦);这次趁著工作空挡,研究了一下免费&上云端的方法。
目标
将 Python 脚本搬到云端执行、定时自动执行、可透过网路开启/关闭。
本篇以我耍的小聪明,针对签到奖励型 APP 撰写的自动完成签到的脚本为例,能每日自动帮我签到,我不用在特别打开 APP 使用;并在执行完成后发通知给我。
完成通知!
本篇章节顺序
使用 Proxyman 进行 Man in the middle attack API 嗅探
撰写 Python 脚本,伪造 APP API 请求(模拟签到动作)
将 Python 脚本搬到 Google Cloud 上
在 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 通知:
这部分我做的很简单,仅共参考,仅通知自己。
建立一个 Provider
- 选择「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-WEST1
、US-CENTRAL1
、US-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 设定自动排程
有了脚本之后再来是要让他自动执行才能达到我们的最终目标。
前往 Google Cloud Scheduler 控制台首页
上方「建立工作」
- 输入工作基本资料
执行频率: 同 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;我们不希望在小东西上花太。
下一步。
动作这边可以设定当预算达到多少百分比时会触发通知。
勾选 「 透过电子邮件将快讯传送给帐单管理员和使用者 」,这样当条件处发时就能第一时间收到通知。
点击「完成」送出储存。
当预算超过时我们就能马上就能知道,避免产生更多费用。
总结
人的精力是有限的,现今科技资讯洪流,每个平台每个服务都想要榨取我们有限的精力;如果能透过一些自动化脚本分担我们的日常生活,聚沙成塔,让我们省下更多精力专心在重要的事情之上!
延伸阅读
有任何问题及指教欢迎 与我联络 。
有自动化相关优化需求也欢迎 发案给我 ,谢谢。
本文首次发表于 Medium (点击查看原始版本),由 ZMediumToMarkdown 提供自动转换与同步技术。