适配腾讯云函数
This commit is contained in:
commit
d07098a767
8
.idea/.gitignore
vendored
Normal file
8
.idea/.gitignore
vendored
Normal file
@ -0,0 +1,8 @@
|
||||
# Default ignored files
|
||||
/shelf/
|
||||
/workspace.xml
|
||||
# Datasource local storage ignored files
|
||||
/../../../../../:\Dev\PycharmProjects\Yuanshen\.idea/dataSources/
|
||||
/dataSources.local.xml
|
||||
# Editor-based HTTP Client requests
|
||||
/httpRequests/
|
||||
10
.idea/Yuanshen.iml
Normal file
10
.idea/Yuanshen.iml
Normal file
@ -0,0 +1,10 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<module type="PYTHON_MODULE" version="4">
|
||||
<component name="NewModuleRootManager">
|
||||
<content url="file://$MODULE_DIR$">
|
||||
<excludeFolder url="file://$MODULE_DIR$/venv" />
|
||||
</content>
|
||||
<orderEntry type="inheritedJdk" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
</module>
|
||||
6
.idea/inspectionProfiles/profiles_settings.xml
Normal file
6
.idea/inspectionProfiles/profiles_settings.xml
Normal file
@ -0,0 +1,6 @@
|
||||
<component name="InspectionProjectProfileManager">
|
||||
<settings>
|
||||
<option name="USE_PROJECT_PROFILE" value="false" />
|
||||
<version value="1.0" />
|
||||
</settings>
|
||||
</component>
|
||||
4
.idea/misc.xml
Normal file
4
.idea/misc.xml
Normal file
@ -0,0 +1,4 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9 (Yuanshen)" project-jdk-type="Python SDK" />
|
||||
</project>
|
||||
8
.idea/modules.xml
Normal file
8
.idea/modules.xml
Normal file
@ -0,0 +1,8 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ProjectModuleManager">
|
||||
<modules>
|
||||
<module fileurl="file://$PROJECT_DIR$/.idea/Yuanshen.iml" filepath="$PROJECT_DIR$/.idea/Yuanshen.iml" />
|
||||
</modules>
|
||||
</component>
|
||||
</project>
|
||||
6
.idea/vcs.xml
Normal file
6
.idea/vcs.xml
Normal file
@ -0,0 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="VcsDirectoryMappings">
|
||||
<mapping directory="$PROJECT_DIR$" vcs="Git" />
|
||||
</component>
|
||||
</project>
|
||||
264
index.py
Normal file
264
index.py
Normal file
@ -0,0 +1,264 @@
|
||||
'''
|
||||
@File : genshin.py
|
||||
@Github : https://github.com/y1ndan/genshin-impact-helper
|
||||
@Last modified by : y1ndan
|
||||
@Last modified time : 2021-01-13 11:10:30
|
||||
'''
|
||||
import hashlib
|
||||
import json
|
||||
import random
|
||||
import string
|
||||
import time
|
||||
import uuid
|
||||
import os
|
||||
|
||||
import requests
|
||||
from requests.exceptions import HTTPError
|
||||
|
||||
from settings import log, CONFIG
|
||||
from notify import Notify
|
||||
|
||||
|
||||
def hexdigest(text):
|
||||
md5 = hashlib.md5()
|
||||
md5.update(text.encode())
|
||||
return md5.hexdigest()
|
||||
|
||||
|
||||
class Base(object):
|
||||
def __init__(self, cookies: str = None):
|
||||
if not isinstance(cookies, str):
|
||||
raise TypeError('%s want a %s but got %s' %
|
||||
(self.__class__, type(__name__), type(cookies)))
|
||||
self._cookie = cookies
|
||||
|
||||
def get_header(self):
|
||||
header = {
|
||||
'User-Agent': CONFIG.USER_AGENT,
|
||||
'Referer': CONFIG.REFERER_URL,
|
||||
'Accept-Encoding': 'gzip, deflate, br',
|
||||
'Cookie': self._cookie
|
||||
}
|
||||
return header
|
||||
|
||||
@staticmethod
|
||||
def to_python(json_str: str):
|
||||
return json.loads(json_str)
|
||||
|
||||
@staticmethod
|
||||
def to_json(obj):
|
||||
return json.dumps(obj, indent=4, ensure_ascii=False)
|
||||
|
||||
|
||||
class Roles(Base):
|
||||
def get_awards(self):
|
||||
response = dict()
|
||||
try:
|
||||
content = requests.Session().get(
|
||||
CONFIG.AWARD_URL, headers=self.get_header()).text
|
||||
response = self.to_python(content)
|
||||
except json.JSONDecodeError as e:
|
||||
log.error(e)
|
||||
|
||||
return response
|
||||
|
||||
def get_roles(self, max_attempt_number: int = 4):
|
||||
log.info('准备获取账号信息...')
|
||||
error = None
|
||||
response = dict()
|
||||
|
||||
for i in range(1, max_attempt_number):
|
||||
try:
|
||||
content = requests.Session().get(
|
||||
CONFIG.ROLE_URL, headers=self.get_header()).text
|
||||
response = self.to_python(content)
|
||||
except HTTPError as error:
|
||||
log.error(
|
||||
'HTTP error when get game roles, retry %s time(s)...' % i)
|
||||
log.error('error is %s' % error)
|
||||
continue
|
||||
except KeyError as error:
|
||||
log.error(
|
||||
'Wrong response to get game roles, retry %s time(s)...'% i)
|
||||
log.error('response is %s' % error)
|
||||
continue
|
||||
except Exception as error:
|
||||
log.error('Unknown error %s, die' % error)
|
||||
raise Exception(error)
|
||||
error = None
|
||||
break
|
||||
|
||||
if error:
|
||||
log.error(
|
||||
'Maximum retry times have been reached, error is %s ' % error)
|
||||
raise Exception(error)
|
||||
if response.get(
|
||||
'retcode', 1) != 0 or response.get('data', None) is None:
|
||||
raise Exception(response['message'])
|
||||
|
||||
log.info('账号信息获取完毕')
|
||||
return response
|
||||
|
||||
|
||||
class Sign(Base):
|
||||
def __init__(self, cookies: str = None):
|
||||
super(Sign, self).__init__(cookies)
|
||||
self._region_list = []
|
||||
self._region_name_list = []
|
||||
self._uid_list = []
|
||||
|
||||
@staticmethod
|
||||
def get_ds():
|
||||
# v2.3.0-web @povsister & @journey-ad
|
||||
n = 'h8w582wxwgqvahcdkpvdhbh2w9casgfl'
|
||||
i = str(int(time.time()))
|
||||
r = ''.join(random.sample(string.ascii_lowercase + string.digits, 6))
|
||||
c = hexdigest('salt=' + n + '&t=' + i + '&r=' + r)
|
||||
return '{},{},{}'.format(i, r, c)
|
||||
|
||||
def get_header(self):
|
||||
header = super(Sign, self).get_header()
|
||||
header.update({
|
||||
'x-rpc-device_id':str(uuid.uuid3(
|
||||
uuid.NAMESPACE_URL, self._cookie)).replace('-', '').upper(),
|
||||
# 1: ios
|
||||
# 2: android
|
||||
# 4: pc web
|
||||
# 5: mobile web
|
||||
'x-rpc-client_type': '5',
|
||||
'x-rpc-app_version': CONFIG.APP_VERSION,
|
||||
'DS': self.get_ds(),
|
||||
})
|
||||
return header
|
||||
|
||||
def get_info(self):
|
||||
user_game_roles = Roles(self._cookie).get_roles()
|
||||
role_list = user_game_roles.get('data', {}).get('list', [])
|
||||
|
||||
# role list empty
|
||||
if not role_list:
|
||||
raise Exception(user_game_roles.get('message', 'Role list empty'))
|
||||
|
||||
log.info(f'当前账号绑定了 {len(role_list)} 个角色')
|
||||
info_list = []
|
||||
# cn_gf01: 天空岛
|
||||
# cn_qd01: 世界树
|
||||
self._region_list = [(i.get('region', 'NA')) for i in role_list]
|
||||
self._region_name_list = [(i.get('region_name', 'NA'))
|
||||
for i in role_list]
|
||||
self._uid_list = [(i.get('game_uid', 'NA')) for i in role_list]
|
||||
|
||||
log.info('准备获取签到信息...')
|
||||
for i in range(len(self._uid_list)):
|
||||
info_url = CONFIG.INFO_URL.format(
|
||||
self._region_list[i], CONFIG.ACT_ID, self._uid_list[i])
|
||||
try:
|
||||
content = requests.Session().get(
|
||||
info_url, headers=self.get_header()).text
|
||||
info_list.append(self.to_python(content))
|
||||
except Exception as e:
|
||||
raise Exception(e)
|
||||
|
||||
if not info_list:
|
||||
raise Exception('User sign info list is empty')
|
||||
log.info('签到信息获取完毕')
|
||||
return info_list
|
||||
|
||||
def run(self):
|
||||
info_list = self.get_info()
|
||||
message_list = []
|
||||
for i in range(len(info_list)):
|
||||
today = info_list[i]['data']['today']
|
||||
total_sign_day = info_list[i]['data']['total_sign_day']
|
||||
awards = Roles(self._cookie).get_awards()['data']['awards']
|
||||
uid = str(self._uid_list[i]).replace(
|
||||
str(self._uid_list[i])[1:8], '******', 1)
|
||||
|
||||
log.info(f'准备为旅行者 {i + 1} 号签到...')
|
||||
time.sleep(10)
|
||||
messgae = {
|
||||
'today': today,
|
||||
'region_name': self._region_name_list[i],
|
||||
'uid': uid,
|
||||
'award_name': awards[total_sign_day]['name'],
|
||||
'award_cnt': awards[total_sign_day]['cnt'],
|
||||
'total_sign_day': total_sign_day,
|
||||
'end': '',
|
||||
}
|
||||
if info_list[i]['data']['is_sign'] is True:
|
||||
messgae['award_name'] = awards[total_sign_day - 1]['name']
|
||||
messgae['award_cnt'] = awards[total_sign_day - 1]['cnt']
|
||||
messgae['status'] = f'👀 旅行者 {i + 1} 号, 你已经签到过了哦'
|
||||
message_list.append(self.message.format(**messgae))
|
||||
continue
|
||||
if info_list[i]['data']['first_bind'] is True:
|
||||
messgae['status'] = f'💪 旅行者 {i + 1} 号, 请先前往米游社App手动签到一次'
|
||||
message_list.append(self.message.format(**messgae))
|
||||
continue
|
||||
|
||||
data = {
|
||||
'act_id': CONFIG.ACT_ID,
|
||||
'region': self._region_list[i],
|
||||
'uid': self._uid_list[i]
|
||||
}
|
||||
|
||||
try:
|
||||
content = requests.Session().post(
|
||||
CONFIG.SIGN_URL,
|
||||
headers=self.get_header(),
|
||||
data=json.dumps(data, ensure_ascii=False)).text
|
||||
response = self.to_python(content)
|
||||
except Exception as e:
|
||||
raise Exception(e)
|
||||
code = response.get('retcode', 99999)
|
||||
# 0: success
|
||||
# -5003: already signed in
|
||||
if code != 0:
|
||||
message_list.append(response)
|
||||
continue
|
||||
messgae['total_sign_day'] = total_sign_day + 1
|
||||
messgae['status'] = response['message']
|
||||
message_list.append(self.message.format(**messgae))
|
||||
log.info('签到完毕')
|
||||
|
||||
return ''.join(message_list)
|
||||
|
||||
@property
|
||||
def message(self):
|
||||
return CONFIG.MESSGAE_TEMPLATE
|
||||
|
||||
|
||||
def main_handler(event, context):
|
||||
log.info('任务开始')
|
||||
notify = Notify()
|
||||
msg_list = []
|
||||
ret = success_num = fail_num = 0
|
||||
# ============= miHoYo BBS COOKIE ============
|
||||
# 此处填米游社的COOKIE
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=COOKIE,Value=<获取的值>
|
||||
# 多个账号的COOKIE值之间用 # 号隔开,例如: 1#2#3#4
|
||||
COOKIE = ''
|
||||
|
||||
if os.environ.get('COOKIE', '') != '':
|
||||
COOKIE = os.environ['COOKIE']
|
||||
|
||||
cookie_list = COOKIE.split('#')
|
||||
log.info(f'检测到共配置了 {len(cookie_list)} 个帐号')
|
||||
for i in range(len(cookie_list)):
|
||||
log.info(f'准备为 NO.{i + 1} 账号签到...')
|
||||
try:
|
||||
msg = f' NO.{i + 1} 账号:{Sign(cookie_list[i]).run()}'
|
||||
msg_list.append(msg)
|
||||
success_num = success_num + 1
|
||||
except Exception as e:
|
||||
msg = f' NO.{i + 1} 账号:\n {e}'
|
||||
msg_list.append(msg)
|
||||
fail_num = fail_num + 1
|
||||
log.error(msg)
|
||||
ret = -1
|
||||
continue
|
||||
notify.send(status=f'成功: {success_num} | 失败: {fail_num}', msg=msg_list)
|
||||
if ret != 0:
|
||||
log.error('异常退出')
|
||||
exit(ret)
|
||||
log.info('任务结束')
|
||||
339
notify.py
Normal file
339
notify.py
Normal file
@ -0,0 +1,339 @@
|
||||
'''
|
||||
@File : notify.py
|
||||
@Github : https://github.com/y1ndan/genshin-impact-helper
|
||||
@Last modified by : y1ndan
|
||||
@Last modified time : 2021-01-13 11:01:10
|
||||
'''
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import hmac
|
||||
import hashlib
|
||||
import base64
|
||||
|
||||
import requests
|
||||
from requests.exceptions import HTTPError
|
||||
from urllib import parse
|
||||
|
||||
from settings import log
|
||||
|
||||
|
||||
class Notify(object):
|
||||
@staticmethod
|
||||
def to_python(json_str: str):
|
||||
return json.loads(json_str)
|
||||
|
||||
@staticmethod
|
||||
def to_json(obj):
|
||||
return json.dumps(obj, indent=4, ensure_ascii=False)
|
||||
|
||||
# ============================== Server Chan ==============================
|
||||
# 此处填你申请的SCKEY
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=SCKEY,Value=<获取的值>
|
||||
SCKEY = ''
|
||||
|
||||
if os.environ.get('SCKEY', '') != '':
|
||||
SCKEY = os.environ['SCKEY']
|
||||
|
||||
# ============================== Cool Push ================================
|
||||
# 此处填你申请的SKEY(详见文档: https://cp.xuthus.cc/)
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=COOL_PUSH_SKEY,Value=<获取的值>
|
||||
COOL_PUSH_SKEY = ''
|
||||
# 此处填写私聊(send)或群组(group)或者微信(wx)推送方式，默认私聊推送
|
||||
# 注: Github Actions用户若要更改,请到Settings->Secrets里设置,Name=COOL_PUSH_MODE,Value=<group或wx>
|
||||
COOL_PUSH_MODE = 'send'
|
||||
|
||||
if os.environ.get('COOL_PUSH_SKEY', '') != '':
|
||||
COOL_PUSH_SKEY = os.environ['COOL_PUSH_SKEY']
|
||||
if os.environ.get('COOL_PUSH_MODE', '') != '':
|
||||
COOL_PUSH_MODE = os.environ['COOL_PUSH_MODE']
|
||||
|
||||
# ============================== iOS Bark App =============================
|
||||
# 此处填你Bark App的信息(IP/设备码,例如: https://api.day.app/XXXXXXXX)
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=BARK_KEY,Value=<获取的值>
|
||||
BARK_KEY = ''
|
||||
# BARK App推送铃声,铃声列表去App内查看
|
||||
# 注: Github Actions用户若要更改,请到Settings->Secrets里设置,Name=BARK_SOUND,Value=<铃声名称>
|
||||
BARK_SOUND = 'healthnotification'
|
||||
|
||||
if os.environ.get('BARK_KEY', '') != '':
|
||||
if os.environ['BARK_KEY'].find(
|
||||
'https') != -1 or os.environ['BARK_KEY'].find('http') != -1:
|
||||
# 兼容BARK自建服务端用户
|
||||
BARK_KEY = os.environ['BARK_KEY']
|
||||
else:
|
||||
BARK_KEY = 'https://api.day.app/' + os.environ['BARK_KEY']
|
||||
elif os.environ.get('BARK_SOUND', '') != '':
|
||||
BARK_SOUND = os.environ['BARK_SOUND']
|
||||
elif BARK_KEY != '' or BARK_KEY.find('https') != -1 or BARK_KEY.find(
|
||||
'http') != -1:
|
||||
# 兼容BARK本地用户只填写设备码的情况
|
||||
BARK_KEY = 'https://api.day.app/' + BARK_KEY
|
||||
|
||||
# ============================== Telegram Bot =============================
|
||||
# 此处填你telegram bot的Token,例如: 1077xxx4424:AAFjv0FcqxxxxxxgEMGfi22B4yh15R5uw
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=TG_BOT_TOKEN,Value=<获取的值>
|
||||
TG_BOT_TOKEN = ''
|
||||
# 此处填你接收通知消息的telegram用户的id,例如: 129xxx206
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=TG_USER_ID,Value=<获取的值>
|
||||
TG_USER_ID = ''
|
||||
|
||||
if os.environ.get('TG_BOT_TOKEN', '') != '':
|
||||
TG_BOT_TOKEN = os.environ['TG_BOT_TOKEN']
|
||||
if os.environ.get('TG_USER_ID', '') != '':
|
||||
TG_USER_ID = os.environ['TG_USER_ID']
|
||||
|
||||
# ============================== DingTalk Bot =============================
|
||||
# 此处填你钉钉机器人的webhook,例如: 5a544165465465645d0f31dca676e7bd07415asdasd
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=DD_BOT_TOKEN,Value=<获取的值>
|
||||
DD_BOT_TOKEN = ''
|
||||
# 加签密钥,机器人安全设置页面,加签一栏下面显示的SEC开头的字符串
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=DD_BOT_SECRET,Value=<获取的值>
|
||||
DD_BOT_SECRET = ''
|
||||
|
||||
if os.environ.get('DD_BOT_TOKEN', '') != '':
|
||||
DD_BOT_TOKEN = os.environ['DD_BOT_TOKEN']
|
||||
if os.environ.get('DD_BOT_SECRET', '') != '':
|
||||
DD_BOT_SECRET = os.environ['DD_BOT_SECRET']
|
||||
|
||||
# ============================== WeChat Work Bot ==========================
|
||||
# 此处填你企业微信机器人的webhook(详见文档 https://work.weixin.qq.com/api/doc/90000/90136/91770) 例如: 693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=WW_BOT_KEY,Value=<获取的值>
|
||||
WW_BOT_KEY = ''
|
||||
|
||||
if os.environ.get('WW_BOT_KEY', '') != '':
|
||||
WW_BOT_KEY = os.environ['WW_BOT_KEY']
|
||||
|
||||
# ============================== iGot聚合推送 =================================
|
||||
# 此处填你iGot的信息(推送key,例如: https://push.hellyw.com/XXXXXXXX)
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=IGOT_KEY,Value=<获取的值>
|
||||
IGOT_KEY = ''
|
||||
|
||||
if os.environ.get('IGOT_KEY', '') != '':
|
||||
IGOT_KEY = os.environ['IGOT_KEY']
|
||||
|
||||
# ============================== push+ ====================================
|
||||
# 官方文档: https://pushplus.hxtrip.com/
|
||||
# PUSH_PLUS_TOKEN: 微信扫码登录后一对一推送或一对多推送下面的token(您的Token)，不配置PUSH_PLUS_USER则默认为一对一推送
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=PUSH_PLUS_TOKEN,Value=<获取的值>
|
||||
PUSH_PLUS_TOKEN = ''
|
||||
# PUSH_PLUS_USER: 一对多推送的“群组编码”（一对多推送下面->您的群组(如无则新建)->群组编码，如果您是创建群组人。也需点击“查看二维码”扫描绑定，否则不能接受群组消息推送）
|
||||
# 注: Github Actions用户请到Settings->Secrets里设置,Name=PUSH_PLUS_USER,Value=<获取的值>
|
||||
PUSH_PLUS_USER = ''
|
||||
|
||||
if os.environ.get('PUSH_PLUS_TOKEN', '') != '':
|
||||
PUSH_PLUS_TOKEN = os.environ['PUSH_PLUS_TOKEN']
|
||||
if os.environ.get('PUSH_PLUS_USER', '') != '':
|
||||
PUSH_PLUS_USER = os.environ['PUSH_PLUS_USER']
|
||||
|
||||
def serverChan(self, text, status, desp):
|
||||
if Notify.SCKEY != '':
|
||||
url = 'https://sc.ftqq.com/{}.send'.format(Notify.SCKEY)
|
||||
data = {'text': '{} {}'.format(text, status), 'desp': desp}
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['errno'] == 0:
|
||||
log.info('Server酱推送成功')
|
||||
elif response['errno'] == 1024:
|
||||
# SCKEY错误或一分钟内发送相同内容
|
||||
log.error('Server酱推送失败:\n{}'.format(response['errmsg']))
|
||||
else:
|
||||
log.error('Server酱推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置Server酱推送所需的SCKEY,取消Server酱推送')
|
||||
pass
|
||||
|
||||
def coolPush(self, text, status, desp):
|
||||
if Notify.COOL_PUSH_SKEY != '':
|
||||
url = 'https://push.xuthus.cc/{}/{}'.format(
|
||||
Notify.COOL_PUSH_MODE, Notify.COOL_PUSH_SKEY)
|
||||
data = '{} {}\n\n{}'.format(text, status, desp).encode('utf-8')
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['code'] == 200:
|
||||
log.info('Cool Push推送成功')
|
||||
else:
|
||||
log.error('Cool Push推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置Cool Push推送所需的COOL_PUSH_SKEY,取消Cool Push推送')
|
||||
pass
|
||||
|
||||
def bark(self, text, status, desp):
|
||||
if Notify.BARK_KEY != '':
|
||||
url = '{}/{} {}/{}?sound={}'.format(Notify.BARK_KEY,
|
||||
text, status, parse.quote(desp), Notify.BARK_SOUND)
|
||||
try:
|
||||
response = self.to_python(requests.get(url).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['code'] == 200:
|
||||
log.info('Bark推送成功')
|
||||
elif response['code'] == 400:
|
||||
log.error('Bark推送失败:\n{}'.format(response['message']))
|
||||
else:
|
||||
log.error('Bark推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置Bark推送所需的BARK_KEY,取消Bark推送')
|
||||
pass
|
||||
|
||||
def tgBot(self, text, status, desp):
|
||||
if Notify.TG_BOT_TOKEN != '' or Notify.TG_USER_ID != '':
|
||||
url = 'https://api.telegram.org/bot{}/sendMessage'.format(
|
||||
Notify.TG_BOT_TOKEN)
|
||||
data = {
|
||||
'chat_id': Notify.TG_USER_ID,
|
||||
'text': '{} {}\n\n{}'.format(text, status, desp),
|
||||
'disable_web_page_preview': True
|
||||
}
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['ok']:
|
||||
log.info('Telegram推送成功')
|
||||
elif response['error_code'] == 400:
|
||||
log.error('请主动给bot发送一条消息并检查接收用户ID是否正确')
|
||||
elif response['error_code'] == 401:
|
||||
log.error('TG_BOT_TOKEN错误')
|
||||
else:
|
||||
log.error('Telegram推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置Telegram推送所需的TG_BOT_TOKEN和TG_USER_ID,取消Telegram推送')
|
||||
pass
|
||||
|
||||
def ddBot(self, text, status, desp):
|
||||
if Notify.DD_BOT_TOKEN != '':
|
||||
url = 'https://oapi.dingtalk.com/robot/send?access_token={}'.format(
|
||||
Notify.DD_BOT_TOKEN)
|
||||
data = {
|
||||
'msgtype': 'text',
|
||||
'text': {
|
||||
'content': '{} {}\n\n{}'.format(text, status, desp)
|
||||
}
|
||||
}
|
||||
if Notify.DD_BOT_SECRET != '':
|
||||
secret = Notify.DD_BOT_SECRET
|
||||
timestamp = int(round(time.time() * 1000))
|
||||
secret_enc = bytes(secret).encode('utf-8')
|
||||
string_to_sign = '{}\n{}'.format(timestamp, secret)
|
||||
string_to_sign_enc = bytes(string_to_sign).encode('utf-8')
|
||||
hmac_code = hmac.new(
|
||||
secret_enc, string_to_sign_enc,
|
||||
digestmod=hashlib.sha256).digest()
|
||||
sign = parse.quote_plus(base64.b64encode(hmac_code))
|
||||
url = 'https://oapi.dingtalk.com/robot/send?access_token={}×tamp={}&sign={}'.format(
|
||||
Notify.DD_BOT_TOKEN, timestamp, sign)
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['errcode'] == 0:
|
||||
log.info('钉钉推送成功')
|
||||
else:
|
||||
log.error('钉钉推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置钉钉推送所需的DD_BOT_TOKEN或DD_BOT_SECRET,取消钉钉推送')
|
||||
pass
|
||||
|
||||
def wwBot(self, text, status, desp):
|
||||
if Notify.WW_BOT_KEY != '':
|
||||
url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={}'.format(
|
||||
Notify.WW_BOT_KEY)
|
||||
data = {
|
||||
'msgtype': 'text',
|
||||
'text': {
|
||||
'content': '{} {}\n\n{}'.format(text, status, desp)
|
||||
}
|
||||
}
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['errcode'] == 0:
|
||||
log.info('企业微信推送成功')
|
||||
else:
|
||||
log.error('企业微信推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置企业微信推送所需的WW_BOT_KEY,取消企业微信推送')
|
||||
pass
|
||||
|
||||
def iGot(self, text, status, desp):
|
||||
if Notify.IGOT_KEY != '':
|
||||
url = 'https://push.hellyw.com/{}'.format(Notify.IGOT_KEY)
|
||||
data = {'title': '{} {}'.format(text, status), 'content': desp}
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['ret'] == 0:
|
||||
log.info('iGot推送成功')
|
||||
else:
|
||||
log.error('iGot推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置iGot推送所需的IGOT_KEY,取消iGot推送')
|
||||
pass
|
||||
|
||||
def pushPlus(self, text, status, desp):
|
||||
if Notify.PUSH_PLUS_TOKEN != '':
|
||||
url = 'https://pushplus.hxtrip.com/send'
|
||||
data = {
|
||||
'token': Notify.PUSH_PLUS_TOKEN,
|
||||
'title': '{} {}'.format(text, status),
|
||||
'content': desp,
|
||||
'topic': Notify.PUSH_PLUS_USER
|
||||
}
|
||||
try:
|
||||
response = self.to_python(requests.post(url, data=data).text)
|
||||
except Exception as e:
|
||||
log.error(e)
|
||||
raise HTTPError
|
||||
else:
|
||||
if response['code'] == 200:
|
||||
log.info('pushplus推送成功')
|
||||
else:
|
||||
log.error('pushplus推送失败:\n{}'.format(response))
|
||||
else:
|
||||
log.info('您未配置pushplus推送所需的PUSH_PLUS_TOKEN,取消pushplus推送')
|
||||
pass
|
||||
|
||||
def send(self, **kwargs):
|
||||
app = '原神签到小助手'
|
||||
status = kwargs.get('status', '')
|
||||
msg = kwargs.get('msg', '')
|
||||
if isinstance(msg, list) or isinstance(msg, dict):
|
||||
# msg = self.to_json(msg)
|
||||
msg = '\n\n'.join(msg)
|
||||
log.info(f'签到结果: {status}\n\n{msg}')
|
||||
log.info('准备推送通知...')
|
||||
|
||||
self.serverChan(app, status, msg)
|
||||
self.coolPush(app, status, msg)
|
||||
self.bark(app, status, msg)
|
||||
self.tgBot(app, status, msg)
|
||||
self.ddBot(app, status, msg)
|
||||
self.wwBot(app, status, msg)
|
||||
self.iGot(app, status, msg)
|
||||
self.pushPlus(app, status, msg)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
Notify().send(app='原神签到小助手', status='签到状态', msg='内容详情')
|
||||
|
||||
5
requirements.txt
Normal file
5
requirements.txt
Normal file
@ -0,0 +1,5 @@
|
||||
certifi==2020.12.5
|
||||
chardet==3.0.4
|
||||
idna==2.7
|
||||
requests==2.20.0
|
||||
urllib3==1.24.3
|
||||
57
settings.py
Normal file
57
settings.py
Normal file
@ -0,0 +1,57 @@
|
||||
# settings
|
||||
import logging
|
||||
|
||||
import os
|
||||
|
||||
__all__ = ['log', 'CONFIG']
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s %(levelname)s %(message)s',
|
||||
datefmt='%Y-%m-%dT%H:%M:%S')
|
||||
|
||||
|
||||
log = logger = logging
|
||||
|
||||
|
||||
class _Config:
|
||||
ACT_ID = 'e202009291139501'
|
||||
APP_VERSION = '2.3.0'
|
||||
REFERER_URL = 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?' \
|
||||
'bbs_auth_required={}&act_id={}&utm_source={}&utm_medium={}&' \
|
||||
'utm_campaign={}'.format('true', ACT_ID, 'bbs', 'mys', 'icon')
|
||||
AWARD_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home?act_id={}'.format(ACT_ID)
|
||||
ROLE_URL = 'https://api-takumi.mihoyo.com/binding/api/getUserGameRolesByCookie?game_biz={}'.format('hk4e_cn')
|
||||
INFO_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/info?region={}&act_id={}&uid={}'
|
||||
SIGN_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
|
||||
USER_AGENT = 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) ' \
|
||||
'miHoYoBBS/{}'.format(APP_VERSION)
|
||||
|
||||
|
||||
class ProductionConfig(_Config):
|
||||
LOG_LEVEL = logging.INFO
|
||||
|
||||
|
||||
class DevelopmentConfig(_Config):
|
||||
LOG_LEVEL = logging.DEBUG
|
||||
|
||||
|
||||
RUN_ENV = os.environ.get('RUN_ENV', 'dev')
|
||||
if RUN_ENV == 'dev':
|
||||
CONFIG = DevelopmentConfig()
|
||||
else:
|
||||
CONFIG = ProductionConfig()
|
||||
|
||||
log.basicConfig(level=CONFIG.LOG_LEVEL)
|
||||
|
||||
|
||||
MESSGAE_TEMPLATE = '''
|
||||
{today:#^28}
|
||||
🔅[{region_name}]{uid}
|
||||
今日奖励: {award_name} × {award_cnt}
|
||||
本月累签: {total_sign_day} 天
|
||||
签到结果: {status}
|
||||
{end:#^28}'''
|
||||
|
||||
CONFIG.MESSGAE_TEMPLATE = MESSGAE_TEMPLATE
|
||||
|
||||
Loading…
Reference in New Issue
Block a user