Compare commits

..

No commits in common. "master" and "v1.3.21" have entirely different histories.

7 changed files with 295 additions and 872 deletions

6
.github/pull.yml vendored
View File

@ -1,6 +0,0 @@
version: "1"
rules: # Array of rules
- base: master # Required. Target branch
upstream: y1ndan:master # Required. Must be in the same fork network.
mergeMethod: hardreset # Optional, one of [none, merge, squash, rebase, hardreset], Default: none.
mergeUnstable: true # Optional, merge pull request even when the mergeable_state is not clean. Default: false

View File

@ -6,8 +6,8 @@ on:
workflow_dispatch:
env:
RUN_ENV: 'prod'
TZ: 'Asia/Shanghai'
# auto merge from y1ndan/genshin-impact-helper, default: false
ALLOW_MERGE: 'false'
jobs:
build:
@ -19,7 +19,16 @@ jobs:
uses: actions/checkout@v2
with:
fetch-depth: 0
# ref: master
ref: master
- name: Auto merge
if: ${{ env.ALLOW_MERGE != 'false' }}
run: |
git config --global user.name github-actions
git config --global user.email github-actions@github.com
git remote add upstream https://github.com/y1ndan/genshin-impact-helper
git pull upstream master --allow-unrelated-histories
git push origin master
- name: Set up python
uses: actions/setup-python@v2
@ -31,23 +40,8 @@ jobs:
run: sleep $(shuf -i 10-300 -n 1)
- name: Run sign
env:
COOKIE: ${{ secrets.COOKIE }}
SCKEY: ${{ secrets.SCKEY }}
COOL_PUSH_SKEY: ${{ secrets.COOL_PUSH_SKEY }}
COOL_PUSH_MODE: ${{ secrets.COOL_PUSH_MODE }}
BARK_KEY: ${{ secrets.BARK_KEY }}
BARK_SOUND: ${{ secrets.BARK_SOUND }}
TG_BOT_TOKEN: ${{ secrets.TG_BOT_TOKEN }}
TG_USER_ID: ${{ secrets.TG_USER_ID }}
DD_BOT_TOKEN: ${{ secrets.DD_BOT_TOKEN }}
DD_BOT_SECRET: ${{ secrets.DD_BOT_SECRET }}
WW_BOT_KEY: ${{ secrets.WW_BOT_KEY }}
IGOT_KEY: ${{ secrets.IGOT_KEY }}
PUSH_PLUS_TOKEN: ${{ secrets.PUSH_PLUS_TOKEN }}
PUSH_PLUS_USER: ${{ secrets.PUSH_PLUS_USER }}
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
python3 ./genshin.py
# echo '${{ secrets.COOKIE }}' | tr '#' '\n' | sed 's/$/&#${{ secrets.SCKEY }}/g' | xargs -I {} sh -c 'echo "{}" | python3 ./genshin.py'
echo '${{ secrets.COOKIE }}' | tr '#' '\n' | sed 's/$/&#${{ secrets.SCKEY }}/g' | xargs -I {} sh -c 'echo "{}" | python3 ./genshin.py'

57
.gitignore vendored
View File

@ -1,57 +0,0 @@
*.py[cod]
# C extensions
*.so
# Packages
*.egg
*.egg-info
dist
build
eggs
parts
bin
var
sdist
develop-eggs
.installed.cfg
lib
lib64
__pycache__
# Installer logs
pip-log.txt
# Unit tmp / coverage reports
.coverage
.tox
nosetests.xml
.pytest_cache
.python-version
# Translations
*.mo
# Mr Developer
.mr.developer.cfg
.project
.pydevproject
# temp file
.DS_Store
*.pkl
# venv
.venv/
# Cookiecutter
output/
# vscode
.vscode
# notebooks
notebooks/
# idea
.idea

175
README.md
View File

@ -13,7 +13,7 @@ Genshin Impact Helper
</div>
## 💭前言[20210407]
## 💭前言
> 吹水交流:[130516740](https://qm.qq.com/cgi-bin/qm/qr?k=_M9lYFxkYD7yQQR2btyG3pkZWFys_I-l&authKey=evGDzE2eFVBm46jsHpgcWrokveg70Z9GKl3H45o0oJuia620UGeO27lDPG9gKb/2&noverify=0)
@ -32,18 +32,12 @@ Genshin Impact Helper 可以自动化为你获取原神每日福利。
## 💡特性
- [x] **自动签到** 程序会在每天早上自动执行签到流程,也可以随时通过部署教程的`步骤4`手动触发,具体时间参照[此处](.github/workflows/main.yml)
- [x] **支持同步** 自动同步上游仓库,默认关闭
- [x] **支持订阅** 可选多种订阅方式,通过配置不同参数开启,每天将签到结果推送给订阅用户
- [x] **支持多账号** 不同账号的`Cookie`值之间用`#`分隔,如:`Cookie1#Cookie2#Cookie3`
- [x] **支持订阅** 通过配置`SCKEY`开启订阅,每天将签到结果推送到微信上
- [x] **支持多账号** 不同账号的`Cookie`之间用`#`分隔,如:`myCookie1#myCookie2`
- [x] **支持多角色** 支持绑定官服和B站渠道服角色的米游社账号
## 📐部署
1. Fork 仓库
2. 获取 Cookie
3. 添加 Cookie 至 Secrets
4. 启用 Actions
<details>
<summary>查看教程</summary>
@ -116,177 +110,50 @@ if (ask == true) {
## 🔍结果
当你完成上述流程,可以在`Actions`页面点击`Genshin Impact Helper`-->`build`-->`Run sign`查看运行日志,注意`签到结果`的提示
当你完成上述流程,可以在`Actions`页面点击`Genshin Impact Helper`-->`build`-->`Run sign`查看结果。
<details>
<summary>查看结果</summary>
### 签到成功
如果成功,会输出类似`签到结果: 成功: 1 | 失败: 0 `的信息:
如果成功,会输出类似`"result": "Success"`的信息:
```
签到结果: 成功: 1 | 失败: 0
NO.1 账号:
#########2021-01-13#########
🔅[天空岛]1******9
今日奖励: 摩拉 × 8000
本月累签: 13 天
签到结果: OK
############################
#########2021-01-13#########
🔅[世界树]5******1
今日奖励: 精锻用良矿 × 3
本月累签: 2 天
签到结果: OK
############################
2020-11-18T22:11:45 INFO Sleep for 100 seconds ...
2020-11-18T22:13:26 INFO UID is 100***000
2020-11-18T22:13:27 INFO {
"result": "Success",
"message": "{\"retcode\": 0, \"message\": \"OK\", \"data\": {\"code\": \"ok\"}}"
}
```
### 签到失败
如果失败,会输出类似`签到结果: 成功: 0 | 失败: 1`的信息:
如果失败,会输出类似`"result": "Failed"`的信息:
```
签到结果: 成功: 0 | 失败: 1
NO.1 账号:
登录失效,请重新登录
2020-11-17T22:11:33 INFO Sleep for 54 seconds ...
2020-11-17T22:12:28 INFO UID is 100***000
2020-11-17T22:12:29 INFO {
"result": "Failed",
"message": "{\"data\": null, \"message\": \"请求异常\", \"retcode\": -401}"
}
Error: Process completed with exit code 255.
```
同时你会收到一封来自GitHub、标题为`Run failed: Genshin Impact Helper - master`的邮件。
</details>
注:若开启订阅推送,无论成功与否,都会收到推送通知。
## 🔄同步
因为接口请求上可能发生一些变化,所以上游源代码需要作出更改来适配这些变化,如果你没有及时同步项目源代码,可能会导致签到失败。
**如果你不熟悉 Github 如何同步上游仓库,建议删除你 Fork 的仓库(仓库的`Settings - Options - Danger Zone - Delete this repository`),以重新 Fork 的方式来同步更新,不要再乱点 Pull Request了**
⚠️开启自动同步后[存在的风险](https://github.com/y1ndan/genshin-impact-helper/pull/47#issuecomment-751869761)
> 这导致了开发者账号泄露后用户被供应链攻击的隐患,而主页的协议中没有明确指出这一点。协议中同时包含了“除此之外,开发者无权获取您的 Cookie”这一陈述而事实上开发者在此次PR后可以通过更改源代码来在用户未经授权的情况下收集用户Cookie。此前用户在使用本软件时应该默认进行代码审查然后手动在自己的Repo里PR进行更新。现在的则跳过了这一用户授权更新的动作。
若你了解并接受自动同步带来的可能的风险,请继续往下阅读:
<details>
<summary>开启同步</summary>
项目重新启用自动同步功能,默认关闭。
同步默认使用远程仓库覆盖复刻仓库的方式,如果想保留自己的修改,可以编辑`pull.yml`文件,将`mergeMethod: hardreset`修改为`mergeMethod: merge`。
### 激活安装
1. 前往 `https://pull.git.ci/check/${owner}/genshin-impact-helper` 激活配置文件,其中`${owner}`修改为你的 Github 用户名
2. 安装 [![<img src="https://prod.download/pull-18h-svg" valign="bottom"/> Pull](https://prod.download/pull-18h-svg) Pull app](https://github.com/apps/pull),在安装向导页选择`Only select repositories`,下拉列表选择`genshin-impact-helper`,点击`Install`完成安装
3. 程序会在上游仓库有更新时 3 小时内自动同步
### 手动触发
完成激活安装后,你可以随时前往 `https://pull.git.ci/process/${owner}/genshin-impact-helper` 手动触发同步,其中`${owner}`修改为你的 Github 用户名,网页显示`Success`则触发成功。
如果没有自动同步,应检查你的仓库是否已经是最新的;或者检查仓库的`Pull requests
`里是否有以`[pull]`开头的合并请求,若有则需要点进去找到`Merge pull request`按钮,点击确认合并。
</details>
## 🔔订阅
若开启订阅推送,无论成功与否,都会收到推送通知
### Push All In One
支持Server酱、酷推、Bark App、Telegram Bot、钉钉机器人、企业微信机器人、iGot聚合推送和pushplus 单个或多个推送,配置对应参数就会开启对应的推送方式,参数列表详见下文`参数`部分。
#### Server酱
以Server酱为例
**a.获取 SCKEY**
若开启订阅推送,无论成功与否,都会收到微信通知。
- 使用 GitHub 登录 [sc.ftqq.com](http://sc.ftqq.com/?c=github&a=login) 创建账号
- 点击「[发送消息](http://sc.ftqq.com/?c=code)」,获取`SCKEY`
- 点击「[微信推送](http://sc.ftqq.com/?c=wechat&a=bind)」,完成微信绑定
**b.添加 SCKEY 到 Secrets**
- 建立名为`SCKEY`的 secret并添加获取的 SCKEY 值即可开启Server酱推送
其他推送方式请参考对应官方文档获取 KEY 或 TOKEN 等参数,再添加到`Secrets`里。
## 🧬参数
在`Settings`-->`Secrets`里添加的参数,`Name`必须为下列的参数名称之一,`Value`则填写对应获取的值
| 参数名称 | 是否必填 | 默认值 | 说明 |
|--- |--- |--- |--- |
| COOKIE | ✅ | | 米游社的Cookie |
| SCKEY | ❌ | | Server酱推送所需的SCKEY |
| COOL_PUSH_SKEY | ❌ | | Cool Push推送所需的SKEY |
| COOL_PUSH_MODE | ❌ | send | Cool Push推送方式,可选群组(group)或者微信(wx) |
| BARK_KEY | ❌ | | Bark推送所需的BARK_KEY |
| BARK_SOUND | ❌ | healthnotification | Bark推送的铃声,在APP内查看铃声列表 |
| TG_BOT_TOKEN | ❌ | | Telegram Bot的TOKEN |
| TG_USER_ID | ❌ | | 接收通知消息的Telegram用户的ID |
| DD_BOT_TOKEN | ❌ | | 钉钉机器人的webhook KEY |
| DD_BOT_SECRET | ❌ | | 钉钉加签密钥,机器人安全设置页面,加签一栏下面显示的SEC开头的字符串 |
| WW_BOT_KEY | ❌ | | 企业微信机器人的webhook KEY |
| IGOT_KEY | ❌ | | iGot推送所需的KEY |
| PUSH_PLUS_TOKEN | ❌ | | pushplus一对一推送或一对多推送下面的Token |
| PUSH_PLUS_USER | ❌ | 一对一推送 | pushplus一对多推送的'群组编码' |
## 🔨开发
如果需要重构或增加额外功能可参考以下数据:
<details>
<summary>查看数据</summary>
```python
# 角色信息
roles = Roles(cookie).get_roles()
roles = {
'retcode': 0,
'message': 'OK',
'data': {
'list': [
{
'game_biz': 'hk4e_cn',
'region': 'cn_gf01',
'game_uid': '111111111',
'nickname': '酸柚子',
'level': 48,
'is_chosen': False,
'region_name': '天空岛',
'is_official': True
}
]
}
}
```
```python
# 签到信息
infos = Sign(cookie).get_info()
infos = [
{
'retcode': 0,
'message': 'OK',
'data': {
'total_sign_day': 5,
'today': '2021-01-05',
'is_sign': True,
'first_bind': False,
'is_sub': False,
'month_first': False
}
}
]
```
</details>
- 建立名为`SCKEY`的 secret并添加获取的 SCKEY 值,即可开启订阅推送
## ❗️协议

499
genshin.py Normal file → Executable file
View File

@ -1,265 +1,286 @@
'''
@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
#!/usr/bin/env python3
import requests
from requests.exceptions import HTTPError
import argparse
import json
import uuid
import logging
import time
import random
import hashlib
import string
from requests.exceptions import *
from settings import log, CONFIG
from notify import Notify
logging.basicConfig(
level = logging.INFO,
format = '%(asctime)s %(levelname)s %(message)s',
datefmt = '%Y-%m-%dT%H:%M:%S')
def hexdigest(text):
class ConfMeta(type):
@property
def ref_url(self):
return 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?' \
'bbs_auth_required={}&act_id={}&utm_source={}&utm_medium={}&' \
'utm_campaign={}'.format('true', self.act_id, 'bbs', 'mys', 'icon')
@property
def award_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home?' \
'act_id={}'.format(self.act_id)
@property
def role_url(self):
return 'https://api-takumi.mihoyo.com/binding/api/' \
'getUserGameRolesByCookie?game_biz={}'.format('hk4e_cn')
@property
def info_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/info?' \
'region={}&act_id={}&uid={}'
@property
def sign_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
@property
def app_version(self):
return '2.3.0'
@property
def ua(self):
return 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) Apple' \
'WebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/{}'.format(self.app_version)
@property
def act_id(self):
return 'e202009291139501'
class Conf(metaclass=ConfMeta):
pass
class Roles(object):
def __init__(self, cookie:str=None):
if type(cookie) is not str:
raise TypeError('%s want a %s but got %s' %(
self.__class__, type(__name__), type(cookie)))
self._cookie = cookie
def get_header(self):
return {
'User-Agent': Conf.ua,
'Referer': Conf.ref_url,
'Accept-Encoding': 'gzip, deflate, br',
'Cookie': self._cookie
}
def get_awards(self):
try:
jdict = json.loads(
requests.Session().get(
Conf.award_url, headers = self.get_header()).text)
except Exception as e:
logging.error(e)
return jdict
def get_roles(self):
logging.info('准备获取账号信息...')
errstr = None
for i in range(1, 4):
try:
jdict = json.loads(requests.Session().get(
Conf.role_url, headers = self.get_header()).text)
except HTTPError as e:
logging.error('HTTP error when get user game roles, ' \
'retry %s time(s) ...' %(i))
logging.error('error is %s' %(e))
errstr = str(e)
continue
except KeyError as e:
logging.error('Wrong response to get user game roles, ' \
'retry %s time(s) ...' %(i))
logging.error('response is %s' %(e))
errstr = str(e)
continue
except Exception as e:
logging.error('Unknown error %s, die' %(e))
errstr = str(e)
raise
else:
break
try:
jdict
logging.info('账号信息获取完毕')
except AttributeError:
raise Exception(errstr)
return jdict
class Sign(object):
def __init__(self, cookie:str=None):
if type(cookie) is not str:
raise TypeError('%s want a %s but got %s' %(
self.__class__, type(__name__), type(cookie)))
self._cookie = cookie
def md5(self, text):
md5 = hashlib.md5()
md5.update(text.encode())
return md5.hexdigest()
def get_DS(self):
# n = self.md5(2.1.0) # v2.1.0 @Steesha
# n = 'cx2y9z9a29tfqvr1qsq6c7yz99b5jsqt' # v2.2.0 @Womsxd
n = 'h8w582wxwgqvahcdkpvdhbh2w9casgfl' # v2.3.0 web @povsister & @journey-ad
i = str(int(time.time()))
r = ''.join(random.sample(string.ascii_lowercase + string.digits, 6))
c = self.md5('salt=' + n + '&t='+ i + '&r=' + r)
return '{},{},{}'.format(i, r, c)
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):
return {
'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',
'Accept-Encoding': 'gzip, deflate, br',
'User-Agent': Conf.ua,
'Referer': Conf.ref_url,
'x-rpc-app_version': Conf.app_version,
'DS': self.get_DS(),
'Cookie': self._cookie
}
def get_header(self):
header = {
'User-Agent': CONFIG.USER_AGENT,
'Referer': CONFIG.REFERER_URL,
'Accept-Encoding': 'gzip, deflate, br',
'Cookie': self._cookie
}
return header
def get_info(self):
roles = Roles(self._cookie).get_roles()
try:
rolesList = roles['data']['list']
except Exception as e:
notify(sckey, '失败', roles['message'])
else:
logging.info('当前账号绑定了 {} 个角色'.format(len(rolesList)))
infoList = []
# cn_gf01: 天空岛
# cn_qd01: 世界树
self._regionList = [(i.get('region', 'NA')) for i in rolesList]
self._regionNameList = [(i.get('region_name', 'NA')) for i in rolesList]
self._uidList = [(i.get('game_uid', 'NA')) for i in rolesList]
@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()
logging.info('准备获取签到信息...')
for i in range(len(self._uidList)):
info_url = Conf.info_url.format(self._regionList[i],
Conf.act_id, self._uidList[i])
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)
infoList.append(json.loads(requests.Session().get(
info_url, headers = self.get_header()).text))
logging.info('签到信息获取完毕')
except Exception as e:
logging.error(e)
return response
return infoList
def get_roles(self, max_attempt_number: int = 4):
log.info('准备获取账号信息...')
error = None
response = dict()
def run(self):
logging.info('任务开始')
messageList = []
infoList = self.get_info()
status = '失败'
for i in range(len(infoList)):
today = infoList[i]['data']['today']
totalSignDay = infoList[i]['data']['total_sign_day']
awards = Roles(self._cookie).get_awards()['data']['awards']
uid = str(self._uidList[i]).replace(
str(self._uidList[i])[3:6], '***', 1)
if infoList[i]['data']['is_sign'] is True:
#if infoList[i]['data']['is_sign'] is False:
status = '成功'
messageList.append(self.message().format(today,
self._regionNameList[i], uid,
awards[totalSignDay - 1]['name'], awards[totalSignDay - 1]['cnt'],
totalSignDay, '旅行者 {} 号,你已经签到过了'.format(i + 1), ''))
elif infoList[i]['data']['first_bind'] is True:
messageList.append(' 旅行者 {} 号为首次绑定,请先前往米游社App手动签到一次'.format(i + 1))
else:
data = {
'act_id': Conf.act_id,
'region': self._regionList[i],
'uid': self._uidList[i]
}
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
logging.info('准备为旅行者 {} 号签到...' \
'\n 区服: {}\n UID: {}'.format(i + 1, self._regionNameList[i], uid))
try:
jdict = json.loads(requests.Session().post(
Conf.sign_url, headers = self.get_header(),
data = json.dumps(data, ensure_ascii=False)).text)
logging.info('签到完毕')
except Exception as e:
raise
else:
code = jdict['retcode']
# 0: success
# -5003: already signed in
if code == 0:
status = '成功'
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'])
messageList.append(self.message().format(today,
self._regionNameList[i], uid,
awards[totalSignDay]['name'], awards[totalSignDay]['cnt'],
totalSignDay + 1, jdict['message'], ''))
else:
messageList.append(jdict)
log.info('账号信息获取完毕')
return response
return notify(sckey, status, ",".join(messageList))
def message(self):
return '''
{:#^30}
🔅[{}]{}
今日奖励: {} × {}
本月累签: {}
签到结果: {}
{:#^30}'''
class Sign(Base):
def __init__(self, cookies: str = None):
super(Sign, self).__init__(cookies)
self._region_list = []
self._region_name_list = []
self._uid_list = []
def notify(sckey, status, message):
logging.info('签到{}: {}'.format(status, message))
if sckey.startswith('SC'):
logging.info('准备推送通知...')
url = 'https://sc.ftqq.com/{}.send'.format(sckey)
data = {'text': '原神签到小助手 签到{}'.format(status), 'desp': message}
try:
jdict = json.loads(
requests.Session().post(url, data = data).text)
except Exception as e:
logging.error(e)
raise HTTPError
else:
errmsg = jdict['errmsg']
if errmsg == 'success':
logging.info('推送成功')
else:
logging.error('{}: {}'.format('推送失败', jdict))
else:
logging.info('未配置 SCKEY,正在跳过通知推送')
@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
logging.info('任务结束')
if status == '失败':
return exit(-1)
if __name__ == '__main__':
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 = ''
secret = input().strip().split('#')
secret.append('')
cookie = secret[0]
sckey = secret[1]
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('任务结束')
Sign(cookie).run()

339
notify.py
View File

@ -1,339 +0,0 @@
'''
@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={}&timestamp={}&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='内容详情')

View File

@ -1,57 +0,0 @@
# 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