mirror of
https://github.com/wq-h/qinglong.git
synced 2025-12-16 23:10:13 +08:00
自用脚本
This commit is contained in:
weiqun
26
README.md
26
README.md
@@ -1 +1,25 @@
|
||||
# qinglong
|
||||
# Tieba
|
||||
一个python脚本，可以实现百度贴吧签到，支持青龙面板运行只需要百度账号的BDUSS即可
|
||||
引用项目: https://github.com/fanfan142/Tieba_Sgin.git
|
||||
## 修改后的功能：
|
||||
1. 修改成变量方式
|
||||
export Tieba_BDUSS=""
|
||||
2. 修改通知方式，去除只Server酱通知
|
||||
## 青龙面板使用
|
||||
1. 青龙面板使用需要再python3中添加 pretty_errors 和 requests 依赖
|
||||
2. 抓取百度贴吧Application中的Cookies BDUSS的value值
|
||||
|
||||
# glados 机场
|
||||
引用项目: https://github.com/komori-flag/glados_automation.git
|
||||
我的邀请码:DU5MD-3NWHA-GP3DK-IESMU 注册地址: https://glados.space/landing/DU5MD-3NWHA-GP3DK-IESMU
|
||||
## 修改后的功能：
|
||||
1. 修改通知方式，去除只PUSHPLUS通知
|
||||
## 青龙面板使用
|
||||
1. 抓取glados 控制台中的Cookies 值
|
||||
2. 配置变量 GLADOS_COOKIES 如果有多个账号需使用 '&' 隔开 示例：cookie&cookie&cookie
|
||||
|
||||
|
||||
# 拉取命令
|
||||
```shell
|
||||
ql repo https://github.com/wq-h/qinglong.git "tieba.py|glados.py" "README.md" "sendNotify.py" "main"
|
||||
```
|
||||
|
||||
46
glados.py
Normal file
46
glados.py
Normal file
@@ -0,0 +1,46 @@
|
||||
import datetime
|
||||
import requests
|
||||
import os
|
||||
import json
|
||||
|
||||
try:
|
||||
from sendNotify import send
|
||||
except ImportError:
|
||||
print("加载通知服务失败")
|
||||
send = lambda title, message: print(f"Title: {title}\nMessage: {message}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
# GLaDOS cookie
|
||||
cookies = os.environ.get('GLADOS_COOKIES', []).split('&')
|
||||
if cookies[0] == '':
|
||||
print('未获取到GLADOS_COOKIES环境变量')
|
||||
cookies = []
|
||||
exit(0)
|
||||
checkin_url = 'https://glados.rocks/api/user/checkin'
|
||||
status_url = 'https://glados.rocks/api/user/status'
|
||||
referrer = 'https://glados.rocks/console/checkin'
|
||||
origin = 'https://glados.rocks'
|
||||
useragent = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||||
payload = {
|
||||
'token': 'glados.one'
|
||||
}
|
||||
|
||||
msg = "glados 机场签到结果：\n"
|
||||
for cookie in cookies:
|
||||
checkin = requests.post(checkin_url,
|
||||
headers={'cookie': cookie, 'referrer': referrer, 'origin': origin, 'user-agent': useragent, 'content-type': 'application/json; charset=utf-8'},
|
||||
data=json.dumps(payload))
|
||||
status = requests.get(status_url, headers={'cookie': cookie, 'referrer': referrer, 'origin': origin, 'user-agent': useragent})
|
||||
|
||||
days = str(status.json()['data']['leftDays']).split('.')[0]
|
||||
email = status.json()['data']['email']
|
||||
|
||||
balance = str(checkin.json()['list'][0]['balance']).split('.')[0]
|
||||
change = str(checkin.json()['list'][0]['change']).split('.')[0]
|
||||
|
||||
if 'message' in checkin.text:
|
||||
msg += f"{email}|剩余：{days}天|积分：{balance}|变化：{change}\n"
|
||||
else:
|
||||
msg += f"{email} 签到失败，请更新 cookies\n"
|
||||
|
||||
send('Glados_Sign', msg.strip())
|
||||
471
sendNotify.py
Normal file
471
sendNotify.py
Normal file
@@ -0,0 +1,471 @@
|
||||
'''
|
||||
Author : BNDou
|
||||
Date : 2022-11-01 00:18:08
|
||||
LastEditTime : 2022-11-01 00:18:51
|
||||
FilePath : \DailyCheckin_EnShan\sendNotify.py
|
||||
Description :
|
||||
'''
|
||||
#!/usr/bin/env python3
|
||||
# _*_ coding:utf-8 _*_
|
||||
|
||||
#Modify: Kirin
|
||||
|
||||
from curses.ascii import FS
|
||||
import sys
|
||||
import os, re
|
||||
import requests
|
||||
import json
|
||||
import time
|
||||
import hmac
|
||||
import hashlib
|
||||
import base64
|
||||
import urllib.parse
|
||||
from requests.adapters import HTTPAdapter
|
||||
from urllib3.util import Retry
|
||||
|
||||
cur_path = os.path.abspath(os.path.dirname(__file__))
|
||||
root_path = os.path.split(cur_path)[0]
|
||||
sys.path.append(root_path)
|
||||
|
||||
# 通知服务
|
||||
BARK = '' # bark服务,自行搜索; secrets可填;
|
||||
BARK_PUSH='' # bark自建服务器，要填完整链接，结尾的/不要
|
||||
PUSH_KEY = '' # Server酱的PUSH_KEY; secrets可填
|
||||
TG_BOT_TOKEN = '' # tg机器人的TG_BOT_TOKEN; secrets可填1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ
|
||||
TG_USER_ID = '' # tg机器人的TG_USER_ID; secrets可填 1434078534
|
||||
TG_API_HOST='' # tg 代理api
|
||||
TG_PROXY_IP = '' # tg机器人的TG_PROXY_IP; secrets可填
|
||||
TG_PROXY_PORT = '' # tg机器人的TG_PROXY_PORT; secrets可填
|
||||
DD_BOT_TOKEN = '' # 钉钉机器人的DD_BOT_TOKEN; secrets可填
|
||||
DD_BOT_SECRET = '' # 钉钉机器人的DD_BOT_SECRET; secrets可填
|
||||
QQ_SKEY = '' # qq机器人的QQ_SKEY; secrets可填
|
||||
QQ_MODE = '' # qq机器人的QQ_MODE; secrets可填
|
||||
QYWX_AM = '' # 企业微信
|
||||
QYWX_KEY = '' # 企业微信BOT
|
||||
PUSH_PLUS_TOKEN = '' # 微信推送Plus+
|
||||
FS_KEY = '' #飞书群BOT
|
||||
|
||||
notify_mode = []
|
||||
|
||||
message_info = ''''''
|
||||
|
||||
# GitHub action运行需要填写对应的secrets
|
||||
if "BARK" in os.environ and os.environ["BARK"]:
|
||||
BARK = os.environ["BARK"]
|
||||
if "BARK_PUSH" in os.environ and os.environ["BARK_PUSH"]:
|
||||
BARK_PUSH = os.environ["BARK_PUSH"]
|
||||
if "PUSH_KEY" in os.environ and os.environ["PUSH_KEY"]:
|
||||
PUSH_KEY = os.environ["PUSH_KEY"]
|
||||
if "TG_BOT_TOKEN" in os.environ and os.environ["TG_BOT_TOKEN"] and "TG_USER_ID" in os.environ and os.environ["TG_USER_ID"]:
|
||||
TG_BOT_TOKEN = os.environ["TG_BOT_TOKEN"]
|
||||
TG_USER_ID = os.environ["TG_USER_ID"]
|
||||
if "TG_API_HOST" in os.environ and os.environ["TG_API_HOST"]:
|
||||
TG_API_HOST = os.environ["TG_API_HOST"]
|
||||
if "DD_BOT_TOKEN" in os.environ and os.environ["DD_BOT_TOKEN"] and "DD_BOT_SECRET" in os.environ and os.environ["DD_BOT_SECRET"]:
|
||||
DD_BOT_TOKEN = os.environ["DD_BOT_TOKEN"]
|
||||
DD_BOT_SECRET = os.environ["DD_BOT_SECRET"]
|
||||
if "QQ_SKEY" in os.environ and os.environ["QQ_SKEY"] and "QQ_MODE" in os.environ and os.environ["QQ_MODE"]:
|
||||
QQ_SKEY = os.environ["QQ_SKEY"]
|
||||
QQ_MODE = os.environ["QQ_MODE"]
|
||||
# 获取pushplus+ PUSH_PLUS_TOKEN
|
||||
if "PUSH_PLUS_TOKEN" in os.environ:
|
||||
if len(os.environ["PUSH_PLUS_TOKEN"]) > 1:
|
||||
PUSH_PLUS_TOKEN = os.environ["PUSH_PLUS_TOKEN"]
|
||||
# print("已获取并使用Env环境 PUSH_PLUS_TOKEN")
|
||||
# 获取企业微信应用推送 QYWX_AM
|
||||
if "QYWX_AM" in os.environ:
|
||||
if len(os.environ["QYWX_AM"]) > 1:
|
||||
QYWX_AM = os.environ["QYWX_AM"]
|
||||
|
||||
|
||||
if "QYWX_KEY" in os.environ:
|
||||
if len(os.environ["QYWX_KEY"]) > 1:
|
||||
QYWX_KEY = os.environ["QYWX_KEY"]
|
||||
# print("已获取并使用Env环境 QYWX_AM")
|
||||
|
||||
#接入飞书webhook推送
|
||||
if "FS_KEY" in os.environ:
|
||||
if len(os.environ["FS_KEY"]) > 1:
|
||||
FS_KEY = os.environ["FS_KEY"]
|
||||
|
||||
|
||||
if BARK:
|
||||
notify_mode.append('bark')
|
||||
# print("BARK 推送打开")
|
||||
if BARK_PUSH:
|
||||
notify_mode.append('bark')
|
||||
# print("BARK 推送打开")
|
||||
if PUSH_KEY:
|
||||
notify_mode.append('sc_key')
|
||||
# print("Server酱 推送打开")
|
||||
if TG_BOT_TOKEN and TG_USER_ID:
|
||||
notify_mode.append('telegram_bot')
|
||||
# print("Telegram 推送打开")
|
||||
if DD_BOT_TOKEN and DD_BOT_SECRET:
|
||||
notify_mode.append('dingding_bot')
|
||||
# print("钉钉机器人 推送打开")
|
||||
if QQ_SKEY and QQ_MODE:
|
||||
notify_mode.append('coolpush_bot')
|
||||
# print("QQ机器人 推送打开")
|
||||
|
||||
if PUSH_PLUS_TOKEN:
|
||||
notify_mode.append('pushplus_bot')
|
||||
# print("微信推送Plus机器人 推送打开")
|
||||
if QYWX_AM:
|
||||
notify_mode.append('wecom_app')
|
||||
# print("企业微信机器人 推送打开")
|
||||
|
||||
if QYWX_KEY:
|
||||
notify_mode.append('wecom_key')
|
||||
# print("企业微信机器人 推送打开")
|
||||
|
||||
if FS_KEY:
|
||||
notify_mode.append('fs_key')
|
||||
# print("飞书机器人 推送打开")
|
||||
|
||||
def message(str_msg):
|
||||
global message_info
|
||||
print(str_msg)
|
||||
message_info = "{}\n{}".format(message_info, str_msg)
|
||||
sys.stdout.flush()
|
||||
|
||||
def bark(title, content):
|
||||
print("\n")
|
||||
if BARK:
|
||||
try:
|
||||
response = requests.get(
|
||||
f"""https://api.day.app/{BARK}/{title}/{urllib.parse.quote_plus(content)}""").json()
|
||||
if response['code'] == 200:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
except:
|
||||
print('推送失败！')
|
||||
if BARK_PUSH:
|
||||
try:
|
||||
response = requests.get(
|
||||
f"""{BARK_PUSH}/{title}/{urllib.parse.quote_plus(content)}""").json()
|
||||
if response['code'] == 200:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
except:
|
||||
print('推送失败！')
|
||||
print("bark服务启动")
|
||||
if BARK=='' and BARK_PUSH=='':
|
||||
print("bark服务的bark_token未设置!!\n取消推送")
|
||||
return
|
||||
|
||||
def serverJ(title, content):
|
||||
print("\n")
|
||||
if not PUSH_KEY:
|
||||
print("server酱服务的PUSH_KEY未设置!!\n取消推送")
|
||||
return
|
||||
print("serverJ服务启动")
|
||||
data = {
|
||||
"text": title,
|
||||
"desp": content.replace("\n", "\n\n")
|
||||
}
|
||||
response = requests.post(f"https://sc.ftqq.com/{PUSH_KEY}.send", data=data).json()
|
||||
if response['errno'] == 0:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
|
||||
# tg通知
|
||||
def telegram_bot(title, content):
|
||||
try:
|
||||
print("\n")
|
||||
bot_token = TG_BOT_TOKEN
|
||||
user_id = TG_USER_ID
|
||||
if not bot_token or not user_id:
|
||||
print("tg服务的bot_token或者user_id未设置!!\n取消推送")
|
||||
return
|
||||
print("tg服务启动")
|
||||
if TG_API_HOST:
|
||||
if 'http' in TG_API_HOST:
|
||||
url = f"{TG_API_HOST}/bot{TG_BOT_TOKEN}/sendMessage"
|
||||
else:
|
||||
url = f"https://{TG_API_HOST}/bot{TG_BOT_TOKEN}/sendMessage"
|
||||
else:
|
||||
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'{title}\n\n{content}', 'disable_web_page_preview': 'true'}
|
||||
proxies = None
|
||||
if TG_PROXY_IP and TG_PROXY_PORT:
|
||||
proxyStr = "http://{}:{}".format(TG_PROXY_IP, TG_PROXY_PORT)
|
||||
proxies = {"http": proxyStr, "https": proxyStr}
|
||||
try:
|
||||
response = requests.post(url=url, headers=headers, params=payload, proxies=proxies).json()
|
||||
except:
|
||||
print('推送失败！')
|
||||
if response['ok']:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
def dingding_bot(title, content):
|
||||
timestamp = str(round(time.time() * 1000)) # 时间戳
|
||||
secret_enc = DD_BOT_SECRET.encode('utf-8')
|
||||
string_to_sign = '{}\n{}'.format(timestamp, DD_BOT_SECRET)
|
||||
string_to_sign_enc = string_to_sign.encode('utf-8')
|
||||
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
|
||||
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) # 签名
|
||||
print('开始使用 钉钉机器人 推送消息...', end='')
|
||||
url = f'https://oapi.dingtalk.com/robot/send?access_token={DD_BOT_TOKEN}×tamp={timestamp}&sign={sign}'
|
||||
headers = {'Content-Type': 'application/json;charset=utf-8'}
|
||||
data = {
|
||||
'msgtype': 'text',
|
||||
'text': {'content': f'{title}\n\n{content}'}
|
||||
}
|
||||
response = requests.post(url=url, data=json.dumps(data), headers=headers, timeout=15).json()
|
||||
if not response['errcode']:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
|
||||
def coolpush_bot(title, content):
|
||||
print("\n")
|
||||
if not QQ_SKEY or not QQ_MODE:
|
||||
print("qq服务的QQ_SKEY或者QQ_MODE未设置!!\n取消推送")
|
||||
return
|
||||
print("qq服务启动")
|
||||
url=f"https://qmsg.zendee.cn/{QQ_MODE}/{QQ_SKEY}"
|
||||
payload = {'msg': f"{title}\n\n{content}".encode('utf-8')}
|
||||
response = requests.post(url=url, params=payload).json()
|
||||
if response['code'] == 0:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
# push推送
|
||||
def pushplus_bot(title, content):
|
||||
try:
|
||||
print("\n")
|
||||
if not PUSH_PLUS_TOKEN:
|
||||
print("PUSHPLUS服务的token未设置!!\n取消推送")
|
||||
return
|
||||
print("PUSHPLUS服务启动")
|
||||
url = 'http://www.pushplus.plus/send'
|
||||
data = {
|
||||
"token": PUSH_PLUS_TOKEN,
|
||||
"title": title,
|
||||
"content": content
|
||||
}
|
||||
body = json.dumps(data).encode(encoding='utf-8')
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
response = requests.post(url=url, data=body, headers=headers).json()
|
||||
if response['code'] == 200:
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！')
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
|
||||
|
||||
print("xxxxxxxxxxxx")
|
||||
def wecom_key(title, content):
|
||||
print("\n")
|
||||
if not QYWX_KEY:
|
||||
print("QYWX_KEY未设置!!\n取消推送")
|
||||
return
|
||||
print("QYWX_KEY服务启动")
|
||||
print("content"+content)
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
data = {
|
||||
"msgtype":"text",
|
||||
"text":{
|
||||
"content":title+"\n"+content.replace("\n", "\n\n")
|
||||
}
|
||||
}
|
||||
|
||||
print(f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={QYWX_KEY}")
|
||||
response = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={QYWX_KEY}", json=data,headers=headers).json()
|
||||
print(response)
|
||||
|
||||
# 飞书机器人推送
|
||||
def fs_key(title, content):
|
||||
print("\n")
|
||||
if not FS_KEY:
|
||||
print("FS_KEY未设置!!\n取消推送")
|
||||
return
|
||||
print("FS_KEY服务启动")
|
||||
print("content"+content)
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
data = {
|
||||
"msg_type":"text",
|
||||
"content":{
|
||||
"text":title+"\n"+content.replace("\n", "\n\n")
|
||||
}
|
||||
}
|
||||
|
||||
print(f"https://open.feishu.cn/open-apis/bot/v2/hook/{FS_KEY}")
|
||||
response = requests.post(f"https://open.feishu.cn/open-apis/bot/v2/hook/{FS_KEY}", json=data,headers=headers).json()
|
||||
print(response)
|
||||
|
||||
|
||||
# 企业微信 APP 推送
|
||||
def wecom_app(title, content):
|
||||
try:
|
||||
if not QYWX_AM:
|
||||
print("QYWX_AM 并未设置！！\n取消推送")
|
||||
return
|
||||
QYWX_AM_AY = re.split(',', QYWX_AM)
|
||||
if 4 < len(QYWX_AM_AY) > 5:
|
||||
print("QYWX_AM 设置错误！！\n取消推送")
|
||||
return
|
||||
corpid = QYWX_AM_AY[0]
|
||||
corpsecret = QYWX_AM_AY[1]
|
||||
touser = QYWX_AM_AY[2]
|
||||
agentid = QYWX_AM_AY[3]
|
||||
try:
|
||||
media_id = QYWX_AM_AY[4]
|
||||
except:
|
||||
media_id = ''
|
||||
wx = WeCom(corpid, corpsecret, agentid)
|
||||
# 如果没有配置 media_id 默认就以 text 方式发送
|
||||
if not media_id:
|
||||
message = title + '\n\n' + content
|
||||
response = wx.send_text(message, touser)
|
||||
else:
|
||||
response = wx.send_mpnews(title, content, media_id, touser)
|
||||
if response == 'ok':
|
||||
print('推送成功！')
|
||||
else:
|
||||
print('推送失败！错误信息如下：\n', response)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
|
||||
class WeCom:
|
||||
def __init__(self, corpid, corpsecret, agentid):
|
||||
self.CORPID = corpid
|
||||
self.CORPSECRET = corpsecret
|
||||
self.AGENTID = agentid
|
||||
|
||||
def get_access_token(self):
|
||||
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
|
||||
values = {'corpid': self.CORPID,
|
||||
'corpsecret': self.CORPSECRET,
|
||||
}
|
||||
req = requests.post(url, params=values)
|
||||
data = json.loads(req.text)
|
||||
return data["access_token"]
|
||||
|
||||
def send_text(self, message, touser="@all"):
|
||||
send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + self.get_access_token()
|
||||
send_values = {
|
||||
"touser": touser,
|
||||
"msgtype": "text",
|
||||
"agentid": self.AGENTID,
|
||||
"text": {
|
||||
"content": message
|
||||
},
|
||||
"safe": "0"
|
||||
}
|
||||
send_msges = (bytes(json.dumps(send_values), 'utf-8'))
|
||||
respone = requests.post(send_url, send_msges)
|
||||
respone = respone.json()
|
||||
return respone["errmsg"]
|
||||
|
||||
def send_mpnews(self, title, message, media_id, touser="@all"):
|
||||
send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=' + self.get_access_token()
|
||||
send_values = {
|
||||
"touser": touser,
|
||||
"msgtype": "mpnews",
|
||||
"agentid": self.AGENTID,
|
||||
"mpnews": {
|
||||
"articles": [
|
||||
{
|
||||
"title": title,
|
||||
"thumb_media_id": media_id,
|
||||
"author": "Author",
|
||||
"content_source_url": "",
|
||||
"content": message.replace('\n', '<br/>'),
|
||||
"digest": message
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
send_msges = (bytes(json.dumps(send_values), 'utf-8'))
|
||||
respone = requests.post(send_url, send_msges)
|
||||
respone = respone.json()
|
||||
return respone["errmsg"]
|
||||
|
||||
def send(title, content):
|
||||
"""
|
||||
使用 bark, telegram bot, dingding bot, server, feishuJ 发送手机推送
|
||||
:param title:
|
||||
:param content:
|
||||
:return:
|
||||
"""
|
||||
|
||||
for i in notify_mode:
|
||||
if i == 'bark':
|
||||
if BARK or BARK_PUSH:
|
||||
bark(title=title, content=content)
|
||||
else:
|
||||
print('未启用 bark')
|
||||
continue
|
||||
if i == 'sc_key':
|
||||
if PUSH_KEY:
|
||||
serverJ(title=title, content=content)
|
||||
else:
|
||||
print('未启用 Server酱')
|
||||
continue
|
||||
elif i == 'dingding_bot':
|
||||
if DD_BOT_TOKEN and DD_BOT_SECRET:
|
||||
dingding_bot(title=title, content=content)
|
||||
else:
|
||||
print('未启用 钉钉机器人')
|
||||
continue
|
||||
elif i == 'telegram_bot':
|
||||
if TG_BOT_TOKEN and TG_USER_ID:
|
||||
telegram_bot(title=title, content=content)
|
||||
else:
|
||||
print('未启用 telegram机器人')
|
||||
continue
|
||||
elif i == 'coolpush_bot':
|
||||
if QQ_SKEY and QQ_MODE:
|
||||
coolpush_bot(title=title, content=content)
|
||||
else:
|
||||
print('未启用 QQ机器人')
|
||||
continue
|
||||
elif i == 'pushplus_bot':
|
||||
if PUSH_PLUS_TOKEN:
|
||||
pushplus_bot(title=title, content=content)
|
||||
else:
|
||||
print('未启用 PUSHPLUS机器人')
|
||||
continue
|
||||
elif i == 'wecom_app':
|
||||
if QYWX_AM:
|
||||
wecom_app(title=title, content=content)
|
||||
else:
|
||||
print('未启用企业微信应用消息推送')
|
||||
continue
|
||||
elif i == 'wecom_key':
|
||||
if QYWX_KEY:
|
||||
|
||||
for i in range(int(len(content)/2000)+1):
|
||||
wecom_key(title=title, content=content[i*2000:(i+1)*2000])
|
||||
else:
|
||||
print('未启用企业微信应用消息推送')
|
||||
continue
|
||||
elif i == 'fs_key':
|
||||
if FS_KEY:
|
||||
fs_key(title=title, content=content)
|
||||
else:
|
||||
print('未启用飞书机器人消息推送')
|
||||
continue
|
||||
else:
|
||||
print('此类推送方式不存在')
|
||||
|
||||
|
||||
def main():
|
||||
send('title', 'content')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
142
tieba.py
Normal file
142
tieba.py
Normal file
@@ -0,0 +1,142 @@
|
||||
# -*- coding: utf8 -*-
|
||||
import os
|
||||
import time
|
||||
from requests import session
|
||||
from hashlib import md5
|
||||
|
||||
try:
|
||||
from sendNotify import send
|
||||
except ImportError:
|
||||
print("加载通知服务失败")
|
||||
|
||||
class Tieba():
|
||||
Tieba_BDUSS = os.getenv("Tieba_BDUSS")
|
||||
|
||||
def __init__(self, STOKEN):
|
||||
self.BDUSS = Tieba.Tieba_BDUSS
|
||||
self.STOKEN = STOKEN
|
||||
self.success_list = []
|
||||
self.result = {}
|
||||
self.sign_list = []
|
||||
self.fail_list = []
|
||||
self.session = session()
|
||||
self.session.headers.update(
|
||||
{'Accept': 'text/html, */*; q=0.01',
|
||||
'Accept-Encoding': 'gzip, deflate',
|
||||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||||
'Connection': 'keep-alive',
|
||||
'Host': 'tieba.baidu.com',
|
||||
'Referer': 'http://tieba.baidu.com/i/i/forum',
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
|
||||
'Chrome/71.0.3578.98 Safari/537.36',
|
||||
'X-Requested-With': 'XMLHttpRequest'}
|
||||
)
|
||||
|
||||
def set_cookie(self):
|
||||
self.session.cookies.update({'BDUSS': self.BDUSS, 'STOKEN': self.STOKEN})
|
||||
|
||||
def fetch_tbs(self):
|
||||
r = self.session.get('http://tieba.baidu.com/dc/common/tbs').json()
|
||||
if r['is_login'] == 1:
|
||||
self.tbs = r['tbs']
|
||||
else:
|
||||
raise Exception('获取tbs错误！以下为返回数据：' + str(r))
|
||||
|
||||
def fetch_likes(self):
|
||||
self.rest = set()
|
||||
self.already = set()
|
||||
r = self.session.get('https://tieba.baidu.com/mo/q/newmoindex?').json()
|
||||
if r['no'] == 0:
|
||||
for forum in r['data']['like_forum']:
|
||||
if forum['is_sign'] == 1:
|
||||
self.already.add(forum['forum_name'])
|
||||
else:
|
||||
self.rest.add(forum['forum_name'])
|
||||
else:
|
||||
raise Exception('获取关注贴吧错误！以下为返回数据：' + str(r))
|
||||
|
||||
def sign(self, forum_name):
|
||||
data = {
|
||||
'kw': forum_name,
|
||||
'tbs': self.tbs,
|
||||
'sign': md5(f'kw={forum_name}tbs={self.tbs}tiebaclient!!!'.encode('utf8')).hexdigest()
|
||||
}
|
||||
r = self.session.post('http://c.tieba.baidu.com/c/c/forum/sign', data).json()
|
||||
if r['error_code'] == '160002':
|
||||
print(f'"{forum_name}"已签到')
|
||||
self.sign_list.append(forum_name)
|
||||
return True
|
||||
elif r['error_code'] == '0':
|
||||
print(f'"{forum_name}">>>>>>>签到成功，您是第{r["user_info"]["user_sign_rank"]}个签到的用户！')
|
||||
self.result[forum_name] = r
|
||||
self.success_list.append(forum_name)
|
||||
return True
|
||||
else:
|
||||
print(f'"{forum_name}"签到失败！以下为返回数据：{str(r)}')
|
||||
self.fail_list.append(forum_name)
|
||||
return False
|
||||
|
||||
def loop(self, n):
|
||||
print(f'* 开始第{n}轮签到 *')
|
||||
rest = set()
|
||||
self.fetch_tbs()
|
||||
for forum_name in self.rest:
|
||||
flag = self.sign(forum_name)
|
||||
if not flag:
|
||||
rest.add(forum_name)
|
||||
self.rest = rest
|
||||
if n >= 10: # 最大重试次数
|
||||
self.rest = set()
|
||||
|
||||
def main(self, max):
|
||||
self.set_cookie()
|
||||
self.fetch_likes()
|
||||
n = 0
|
||||
if self.already:
|
||||
print('---------- 已经签到的贴吧 ---------')
|
||||
for forum_name in self.already:
|
||||
print(f'"{forum_name}"已签到')
|
||||
self.sign_list.append(forum_name)
|
||||
while n < max and self.rest:
|
||||
n += 1
|
||||
self.loop(n)
|
||||
|
||||
if self.rest:
|
||||
print('--------- 签到失败列表 ----------')
|
||||
for forum_name in self.rest:
|
||||
print(f'"{forum_name}"签到失败！')
|
||||
|
||||
# 在签到结束后调用 send 函数发送通知
|
||||
self.send_notification_message()
|
||||
|
||||
def send_notification_message(self):
|
||||
msg = "贴吧签到结果：\n"
|
||||
|
||||
if self.success_list:
|
||||
msg += "- **签到成功贴吧**：\n"
|
||||
for forum in self.success_list:
|
||||
sign_rank = self.result[forum]['user_info']['user_sign_rank']
|
||||
msg += f" {forum} （签到成功，第{sign_rank}个签到）\n"
|
||||
|
||||
if self.sign_list:
|
||||
msg += "- **已经签到的贴吧**：\n"
|
||||
msg += " " + "\n ".join(self.sign_list) + "\n"
|
||||
|
||||
msg += f"\n共关注了{len(self.already) + len(self.success_list)}个贴吧，"
|
||||
msg += f"本次成功签到了{len(self.success_list)}个，"
|
||||
msg += f"失败了{len(self.fail_list)}个，"
|
||||
msg += f"有{len(self.sign_list)}个贴吧已经签到。"
|
||||
|
||||
# 发送通知
|
||||
send('Tieba_Sign', msg)
|
||||
|
||||
# Add a 10-second delay before exiting
|
||||
time.sleep(10)
|
||||
|
||||
if __name__ == "__main__":
|
||||
BDUSS_values = os.getenv("Tieba_BDUSS")
|
||||
STOKEN = ''
|
||||
task = Tieba(STOKEN)
|
||||
print("\n========================\n")
|
||||
task.main(3)
|
||||
print("----------贴吧签到执行完毕----------")
|
||||
Reference in New Issue
Block a user