jd/smzdm_script
1
0
Fork 0
mirror of https://github.com/hex-ci/smzdm_script.git synced 2026-02-03 02:24:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Enable setting telegram bot api

Browse Source
This commit is contained in:
LuckyHunter
2022-10-30 13:45:48 +08:00
parent 423399340a
commit 23c8f3b8f8
4 changed files with 64 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.github/workflows/docker-run.yml vendored
View File

@@ -8,8 +8,8 @@ on:
- completed
# Uncomment below to schedule your job
schedule:
- cron: "00 01 * * *"
# schedule:
# - cron: "00 01 * * *"
jobs:
container-test-job:

4
README.md
View File

@@ -12,9 +12,10 @@
- `什么值得买`每日签到
- Github Action(两种配置方式，直接运行或者调用 Docker 运行)
- 本地 Docker 定时运行
- 通过`pushplus`推送运行结果到微信
- 通过`pushplus`推送运行结果到微信(不推荐)
- 通过`server酱`推送运行结果到微信
- 通过`telegram bot`推送
- 自定义反代`Telegram Bot API`, [搭建教程](https://anerg.com/2022/07/25/reverse-proxy-telegram-bot-api-using-cloudflare-worker.html)
## 2. 使用方法
@@ -37,6 +38,7 @@ on:
4. (可选) Secret 新增`PUSH_PLUS_TOKEN`用于推送通知, [详见](https://www.pushplus.plus/)
5. (可选) Secret 新增`SC_KEY`用于推送通知, [详见](https://sct.ftqq.com/)
6. (可选) Secret 新增`TG_BOT_TOKEN` 和`TG_USER_ID`用于推送通知
7. (可选) Secret 新增`TG_BOT_API`用于自定义反代的`Telegram Bot API`
### 2.2 本地运行

1
main.py
View File

@@ -88,6 +88,7 @@ def main():
"SC_KEY": os.environ.get("SC_KEY", None),
"TG_BOT_TOKEN": os.environ.get("TG_BOT_TOKEN", None),
"TG_USER_ID": os.environ.get("TG_USER_ID", None),
"TG_BOT_API": os.environ.get("TG_BOT_API", None),
}
SMZDM_COOKIE = conf_kwargs.get(
"SMZDM_COOKIE").encode('UTF-8').decode('latin-1')

104
notify/notify.py
View File

@@ -1,7 +1,7 @@
import json
import os
from pprint import pp, pprint
from pprint import pprint
from typing import Dict
from urllib.parse import urljoin
import requests
@@ -18,55 +18,67 @@ class NotifyBot(object):
self.tg_bot()
def push_plus(self, template='html'):
if self.kwargs.get("PUSH_PLUS_TOKEN", None):
PUSH_PLUS_TOKEN = self.kwargs.get("PUSH_PLUS_TOKEN")
else:
pprint("⚠️ PUSH_PLUS_TOKEN not set, skip PushPlus nofitication")
return
try:
if self.kwargs.get("PUSH_PLUS_TOKEN", None):
PUSH_PLUS_TOKEN = self.kwargs.get("PUSH_PLUS_TOKEN")
else:
pprint("⚠️ PUSH_PLUS_TOKEN not set, skip PushPlus nofitication")
return
url = 'https://www.pushplus.plus/send'
body = {
'token': PUSH_PLUS_TOKEN,
'title': self.title,
'content': self.content,
'template': template
}
data = json.dumps(body).encode(encoding='utf-8')
headers = {'Content-Type': 'application/json'}
resp = requests.post(url, data=data, headers=headers)
if resp.status_code == 200:
pprint("✅ Push Plus notified")
return resp.json()
url = 'https://www.pushplus.plus/send'
body = {
'token': PUSH_PLUS_TOKEN,
'title': self.title,
'content': self.content,
'template': template
}
data = json.dumps(body).encode(encoding='utf-8')
headers = {'Content-Type': 'application/json'}
resp = requests.post(url, data=data, headers=headers)
if resp.status_code == 200:
pprint("✅ Push Plus notified")
return resp.json()
except Exception as e:
pprint(e)
def server_chain(self):
if self.kwargs.get("SC_KEY", None):
SC_KEY = self.kwargs.get("SC_KEY")
else:
pprint("⚠️ SC_KEY not set, skip ServerChain notification")
return
try:
if self.kwargs.get("SC_KEY", None):
SC_KEY = self.kwargs.get("SC_KEY")
else:
pprint("⚠️ SC_KEY not set, skip ServerChain notification")
return
url = f'http://sc.ftqq.com/{SC_KEY}.send'
url = f'http://sc.ftqq.com/{SC_KEY}.send'
data = {'text': self.title, 'desp': self.content}
resp = requests.post(url, data=data)
if resp.status_code == 200:
pprint("✅ Server Chain notified")
return resp.json()
data = {'text': self.title, 'desp': self.content}
resp = requests.post(url, data=data)
if resp.status_code == 200:
pprint("✅ Server Chain notified")
return resp.json()
except Exception as e:
pprint(e)
def tg_bot(self):
if self.kwargs.get("TG_BOT_TOKEN", None) and self.kwargs.get("TG_USER_ID", None):
TG_BOT_TOKEN = self.kwargs.get("TG_BOT_TOKEN")
TG_USER_ID = self.kwargs.get("TG_USER_ID")
else:
pprint("⚠️ TG_BOT_TOKEN & TG_USER_ID not set, skip TelegramBot notification")
return
try:
if self.kwargs.get("TG_BOT_TOKEN", None) and self.kwargs.get("TG_USER_ID", None):
TG_BOT_TOKEN = self.kwargs.get("TG_BOT_TOKEN")
TG_USER_ID = self.kwargs.get("TG_USER_ID")
else:
pprint(
"⚠️ TG_BOT_TOKEN & TG_USER_ID not set, skip TelegramBot notification")
return
url = f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage"
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
payload = {'chat_id': str(TG_USER_ID),
'text': f'{self.title}\n\n{self.content}',
'disable_web_page_preview': 'true'}
resp = requests.post(url=url, headers=headers, params=payload)
if resp.status_code == 200:
pprint("✅ Telegram Bot notified")
return resp.json()
TG_BOT_API = self.kwargs.get(
"TG_BOT_API", "https://api.telegram.org")
url = urljoin(TG_BOT_API, f"/bot{TG_BOT_TOKEN}/sendMessage")
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
payload = {'chat_id': str(TG_USER_ID),
'text': f'{self.title}\n\n{self.content}',
'disable_web_page_preview': 'true'}
resp = requests.post(url=url, headers=headers, params=payload)
if resp.status_code == 200:
pprint("✅ Telegram Bot notified")
return resp.json()
except Exception as e:
pprint(e)