renil/lx-music-api-server
1
0
Fork 0
mirror of https://github.com/MeoProject/lx-music-api-server.git synced 2025-05-23 19:17:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: 增加kg token报活

Browse Source 
feat: 增加kg token报活
This commit is contained in:
ikun 2024-04-07 00:14:48 +08:00 committed by GitHub
commit a769b23cdf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 143 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
common/config.py
View File

@ -201,6 +201,12 @@ default = {
"desc": "mix_songmid的获取方式, 默认auto, 可以改成一个数字手动",
"value": "auto"
}
},
"refresh_token": {
"desc": "酷狗token保活相关配置，30天不刷新token会失效，enable是否启动，interval刷新间隔。默认appid=1005时有效，3116需要更换signatureKey",
"enable": False,
"interval": 86000,
"login_url": "http://login.user.kugou.com/v4/login_by_token"
}
}
},

1
modules/kg/__init__.py
View File

@ -22,6 +22,7 @@ from common.exceptions import FailedException
from common import Httpx
from common import utils
import asyncio
from . import refresh_token
async def info(hash_):
tasks = []

124
modules/kg/refresh_token.py Normal file
View File

@ -0,0 +1,124 @@
# ----------------------------------------
# - mode: python -
# - author: helloplhm-qwq - (feat. Huibq)
# - name: refresh_token.py -
# - project: lx-music-api-server -
# - license: MIT -
# ----------------------------------------
# This file is part of the "lx-music-api-server" project.
import time
from common import variable
from common import scheduler
from common import config
from common import log
from .utils import signRequest, tools, aes_sign
import ujson as json
logger = log.log('kg_refresh_token')
async def refresh():
if (not config.read_config('module.kg.user.token')):
return
if (not config.read_config('module.kg.user.refresh_token.enable')):
return
if (config.read_config('module.kg.client.appid') == '1005'):
ts = int(time.time() * 1000)
p3 = aes_sign(json.dumps({'clienttime': ts // 1000, 'token': config.read_config('module.kg.user.token')}))
data = {
'p3': p3,
'clienttime_ms': ts,
't1': 0,
't2': 0,
'userid': config.read_config('module.kg.user.userid')
}
params = {
'dfid': '-',
'appid': tools.appid,
'mid': tools.mid,
'clientver': tools.clientver,
'clienttime': ts // 1000
}
headers = {
'User-Agent': 'Android712-AndroidPhone-8983-18-0-NetMusic-wifi',
'KG-THash': '3e5ec6b',
'KG-Rec': '1',
'KG-RC': '1',
}
login_url = config.read_config('module.kg.user.refresh_token.login_url')
req = await signRequest(login_url, params, {'method': 'POST', 'json': data, 'headers': headers})
body = req.json()
if body['error_code'] != 0:
logger.warning('刷新登录失败, code: ' +
str(body['error_code']) + f'\n响应体: {body}')
return
else:
logger.info('刷新登录成功')
config.write_config('module.kg.user.userid',
str(body['data']['userid']))
logger.info(f'已通过相应数据更新userid')
config.write_config('module.kg.user.token',
body['data']['token'])
logger.info('已通过相应数据更新kg_token')
elif (config.read_config('module.kg.client.appid') == '3116'):
ts = int(time.time() * 1000)
p3 = aes_sign(json.dumps({'clienttime': ts // 1000, 'token': config.read_config('module.kg.user.token')}), key=b'c24f74ca2820225badc01946dba4fdf7', iv=b'adc01946dba4fdf7')
data = {
'p3': p3,
'clienttime_ms': ts,
't1': 0,
't2': 0,
'userid': config.read_config('module.kg.user.userid')
}
params = {
'dfid': '-',
'appid': tools.appid,
'mid': tools.mid,
'clientver': tools.clientver,
'clienttime': ts // 1000
}
headers = {
'User-Agent': 'Android712-AndroidPhone-8983-18-0-NetMusic-wifi',
'KG-THash': '3e5ec6b',
'KG-Rec': '1',
'KG-RC': '1',
}
login_url = config.read_config('module.kg.user.refresh_token.login_url')
req = await signRequest(login_url, params, {'method': 'POST', 'json': data, 'headers': headers})
body = req.json()
if body['error_code'] != 0:
logger.warning('刷新登录失败, code: ' +
str(body['error_code']) + f'\n响应体: {body}')
return
else:
logger.info('刷新登录成功')
config.write_config('module.kg.user.userid',
str(body['data']['userid']))
logger.info(f'已通过相应数据更新userid')
config.write_config('module.kg.user.token',
body['data']['token'])
logger.info('已通过相应数据更新kg_token')
if (not variable.use_cookie_pool):
kgconfig = config.read_config('module.kg')
refresh_login_info = kgconfig.get('refresh_token')
if (refresh_login_info):
kgconfig['user']['refresh_token'] = refresh_login_info
kgconfig.pop('refresh_login')
config.write_config('module.kg', kgconfig)
if (config.read_config('module.kg.user.refresh_token.enable') and not variable.use_cookie_pool):
scheduler.append('kg_refresh_token', refresh,
config.read_config('module.kg.user.refresh_token.interval'))
async def refresh_login_for_pool(user_info):
# TODO
pass
def reg_refresh_login_pool_task():
# TODO
pass
if (variable.use_cookie_pool):
# TODO
pass

14
modules/kg/utils.py
View File

@ -6,10 +6,12 @@
# - license: MIT -
# ----------------------------------------
# This file is part of the "lx-music-api-server" project.
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from common import utils
from common import config
from common import Httpx
import json
createObject = utils.CreateObject
@ -23,6 +25,7 @@ tools = createObject({
"version": config.read_config("module.kg.tracker.version"),
"extra_params": config.read_config("module.kg.tracker.extra_params"),
"appid": config.read_config("module.kg.client.appid"),
'mid': config.read_config('module.kg.user.mid'),
"pid": config.read_config("module.kg.client.pid"),
'qualityHashMap': {
'128k': 'hash_128',
@ -49,14 +52,21 @@ def buildRequestParams(dictionary: dict):
return joined_str
def sign(params, body = "", signkey = tools["signkey"]):
if (isinstance(body, dict)):
body = json.dumps(body)
params = utils.sortDict(params)
params = buildSignatureParams(params, body)
return utils.createMD5(signkey + params + signkey)
async def signRequest(url, params, options, signkey = tools["signkey"]):
params['signature'] = sign(params, options.get("body") if options.get("body") else (options.get("data") if options.get("data") else ""), signkey)
params['signature'] = sign(params, options.get("body") if options.get("body") else (options.get("data") if options.get("data") else (options.get("json") if options.get("json") else "")), signkey)
url = url + "?" + buildRequestParams(params)
return await Httpx.AsyncRequest(url, options)
def getKey(hash_, user_info):
return utils.createMD5(hash_.lower() + tools.pidversec + tools.appid + user_info['mid'] + user_info['userid'])
def aes_sign(plain_text, key=b'90b8382a1bb4ccdcf063102053fd75b8', iv=b'f063102053fd75b8'):
cipher = AES.new(key, AES.MODE_CBC, iv)
crypto = cipher.encrypt(pad(plain_text.encode(), AES.block_size))
return crypto.hex()