Compare commits

...

22 Commits

Author SHA1 Message Date
雨落凋殇
15c958d2d9
Update README.md 2021-04-07 13:53:53 +08:00
y1ndan
fefc5611b6 docs: update README.md 2021-01-13 16:29:56 +08:00
y1ndan
e61c608262 chore: add .python-version to .gitignore 2021-01-13 16:29:14 +08:00
y1ndan
197100ceda refactor: change auto merge method 2021-01-13 16:27:29 +08:00
y1ndan
21db1d8884 feature: push all in one 2021-01-13 16:24:00 +08:00
y1ndan
7e21828f63 refactor: output improvements 2021-01-07 18:24:14 +08:00
y1ndan
693740c436 fix: wrong award output when already signed 2021-01-07 18:22:14 +08:00
y1ndan
6ba5f1f2dd
Merge pull request #99 from PomeloWang/master
refactor: update code
2021-01-06 18:59:55 +08:00
zhipeng wang
b85f7ba48b refactor: update code
统一message推送的格式, 之前的版本is_sign、first_bind的情况会直接推送json
2021-01-06 18:49:36 +08:00
y1ndan
d32d0f30d0
Merge pull request #95 from PomeloWang/master
fix: message format error
2021-01-06 11:56:12 +08:00
zhipeng wang
4fba9c3e5b fix: message format error 2021-01-06 11:33:03 +08:00
zhipeng wang
9c07408659 Merge remote-tracking branch 'upstream/master' 2021-01-06 11:21:49 +08:00
y1ndan
f869766801 Merge ; commit '8006bd471ea5491f7e7214b283fee2b3714470dc'
Conflicts:
	genshin.py
2021-01-05 20:51:59 +08:00
PomeloWang
5d2af3b422 refactor: genshin.py 2021-01-05 20:46:31 +08:00
y1ndan
07b184c7b7 Merge branch 'PomeloWang-master' 2021-01-05 12:19:56 +00:00
PomeloWang
4cf05465ad Refactor code 2021-01-05 20:13:19 +08:00
y1ndan
8006bd471e
rm 2021-01-05 19:46:52 +08:00
zhipeng wang
e02d2c105e update: use env set timezone 2021-01-05 18:08:21 +08:00
zhipeng wang
4158cab643 docs: update README.md 2021-01-05 18:02:01 +08:00
zhipeng wang
c68414a325 fix messages is None 2021-01-05 16:33:00 +08:00
zhipeng wang
ff7d8d6fd2 refactor 2021-01-05 15:38:58 +08:00
PomeloWang
bcbd35e770
Merge pull request #1 from y1ndan/master
Master
2021-01-01 19:25:04 +08:00
7 changed files with 873 additions and 296 deletions

6
.github/pull.yml vendored Normal file
View File

@ -0,0 +1,6 @@
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:
# auto merge from y1ndan/genshin-impact-helper, default: false
ALLOW_MERGE: 'false'
RUN_ENV: 'prod'
TZ: 'Asia/Shanghai'
jobs:
build:
@ -19,16 +19,7 @@ jobs:
uses: actions/checkout@v2
with:
fetch-depth: 0
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
# ref: master
- name: Set up python
uses: actions/setup-python@v2
@ -40,8 +31,23 @@ 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
echo '${{ secrets.COOKIE }}' | tr '#' '\n' | sed 's/$/&#${{ secrets.SCKEY }}/g' | xargs -I {} sh -c 'echo "{}" | python3 ./genshin.py'
python3 ./genshin.py
# echo '${{ secrets.COOKIE }}' | tr '#' '\n' | sed 's/$/&#${{ secrets.SCKEY }}/g' | xargs -I {} sh -c 'echo "{}" | python3 ./genshin.py'

57
.gitignore vendored Normal file
View File

@ -0,0 +1,57 @@
*.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,12 +32,18 @@ Genshin Impact Helper 可以自动化为你获取原神每日福利。
## 💡特性
- [x] **自动签到** 程序会在每天早上自动执行签到流程,也可以随时通过部署教程的`步骤4`手动触发,具体时间参照[此处](.github/workflows/main.yml)
- [x] **支持订阅** 通过配置`SCKEY`开启订阅,每天将签到结果推送到微信上
- [x] **支持多账号** 不同账号的`Cookie`之间用`#`分隔,如:`myCookie1#myCookie2`
- [x] **支持同步** 自动同步上游仓库,默认关闭
- [x] **支持订阅** 可选多种订阅方式,通过配置不同参数开启,每天将签到结果推送给订阅用户
- [x] **支持多账号** 不同账号的`Cookie`值之间用`#`分隔,如:`Cookie1#Cookie2#Cookie3`
- [x] **支持多角色** 支持绑定官服和B站渠道服角色的米游社账号
## 📐部署
1. Fork 仓库
2. 获取 Cookie
3. 添加 Cookie 至 Secrets
4. 启用 Actions
<details>
<summary>查看教程</summary>
@ -110,50 +116,177 @@ if (ask == true) {
## 🔍结果
当你完成上述流程,可以在`Actions`页面点击`Genshin Impact Helper`-->`build`-->`Run sign`查看结果。
当你完成上述流程,可以在`Actions`页面点击`Genshin Impact Helper`-->`build`-->`Run sign`查看运行日志,注意`签到结果`的提示
<details>
<summary>查看结果</summary>
### 签到成功
如果成功,会输出类似`"result": "Success"`的信息:
如果成功,会输出类似`签到结果: 成功: 1 | 失败: 0 `的信息:
```
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\"}}"
}
签到结果: 成功: 1 | 失败: 0
NO.1 账号:
#########2021-01-13#########
🔅[天空岛]1******9
今日奖励: 摩拉 × 8000
本月累签: 13 天
签到结果: OK
############################
#########2021-01-13#########
🔅[世界树]5******1
今日奖励: 精锻用良矿 × 3
本月累签: 2 天
签到结果: OK
############################
```
### 签到失败
如果失败,会输出类似`"result": "Failed"`的信息:
如果失败,会输出类似`签到结果: 成功: 0 | 失败: 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.
签到结果: 成功: 0 | 失败: 1
NO.1 账号:
登录失效,请重新登录
```
同时你会收到一封来自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)」,完成微信绑定
- 建立名为`SCKEY`的 secret并添加获取的 SCKEY 值,即可开启订阅推送
**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>
## ❗️协议

501
genshin.py Executable file → Normal file
View File

@ -1,286 +1,265 @@
#!/usr/bin/env python3
'''
@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
import argparse
import json
import uuid
import logging
import time
import random
import hashlib
import string
from requests.exceptions import *
from requests.exceptions import HTTPError
logging.basicConfig(
level = logging.INFO,
format = '%(asctime)s %(levelname)s %(message)s',
datefmt = '%Y-%m-%dT%H:%M:%S')
from settings import log, CONFIG
from notify import Notify
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):
def hexdigest(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)
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
}
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_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]
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:
infoList.append(json.loads(requests.Session().get(
info_url, headers = self.get_header()).text))
logging.info('签到信息获取完毕')
except Exception as e:
logging.error(e)
return infoList
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]
def get_header(self):
header = {
'User-Agent': CONFIG.USER_AGENT,
'Referer': CONFIG.REFERER_URL,
'Accept-Encoding': 'gzip, deflate, br',
'Cookie': self._cookie
}
return header
logging.info('准备为旅行者 {} 号签到...' \
'\n 区服: {}\n UID: {}'.format(i + 1, self._regionNameList[i], uid))
@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:
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 = '成功'
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)
messageList.append(self.message().format(today,
self._regionNameList[i], uid,
awards[totalSignDay]['name'], awards[totalSignDay]['cnt'],
totalSignDay + 1, jdict['message'], ''))
else:
messageList.append(jdict)
return response
return notify(sckey, status, ",".join(messageList))
def get_roles(self, max_attempt_number: int = 4):
log.info('准备获取账号信息...')
error = None
response = dict()
def message(self):
return '''
{:#^30}
🔅[{}]{}
今日奖励: {} × {}
本月累签: {}
签到结果: {}
{:#^30}'''
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
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,正在跳过通知推送')
class Sign(Base):
def __init__(self, cookies: str = None):
super(Sign, self).__init__(cookies)
self._region_list = []
self._region_name_list = []
self._uid_list = []
logging.info('任务结束')
if status == '失败':
return exit(-1)
@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
if __name__ == '__main__':
secret = input().strip().split('#')
secret.append('')
cookie = secret[0]
sckey = secret[1]
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 = ''
Sign(cookie).run()
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
View 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={}&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='内容详情')

57
settings.py Normal file
View 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