fix(cloud_music): 修复创建全局二维码对象导致第二次及其之后生成错误的二维码

1. 增加unikey缓存机制避免频繁生成二维码
2. 优化登录状态检查流程
This commit is contained in:
SlyAimer 2025-04-08 16:14:56 +08:00
parent 79293219bc
commit e22bb909d5
2 changed files with 56 additions and 48 deletions

View file

@ -62,7 +62,7 @@ def get_music(id):
""" """
save_path = os.getcwd()+'/src/clover_music/netease_music' save_path = os.getcwd()+'/src/clover_music/netease_music'
os.makedirs(save_path, exist_ok=True) os.makedirs(save_path, exist_ok=True)
qrcode_path = os.getcwd()+'/src/clover_music' qrcode_path = os.getcwd()+'/src/clover_music/'
# 判断cookie是否有效 # 判断cookie是否有效
@ -112,8 +112,6 @@ async def get_qr_key(session):
else: else:
return None return None
# 创建 QRCode 对象
qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=10, border=4, )
# 生成二维码 # 生成二维码
async def create_qr_code(unikey): async def create_qr_code(unikey):
""" """
@ -124,14 +122,12 @@ async def create_qr_code(unikey):
Returns: Returns:
""" """
# 添加数据 qr = qrcode.QRCode(version=1,error_correction=qrcode.constants.ERROR_CORRECT_L,box_size=10,border=4)
png_url = f"http://music.163.com/login?codekey={unikey}" qr.add_data(f"https://music.163.com/login?codekey={unikey}")
qr.add_data(png_url) #保存二维码
img = qr.make_image() img_path = os.path.join(qrcode_path, 'qrcode.png')
a = BytesIO() qr.make_image().save(img_path)
img.save(a, 'png') return img_path
img.save(os.path.join(qrcode_path, 'qrcode.png'))
return qrcode_path + '/qrcode.png'
# 检查二维码状态是否被扫描 # 检查二维码状态是否被扫描
async def check_qr_code(unikey,session): async def check_qr_code(unikey,session):

View file

@ -8,15 +8,18 @@ from nonebot.adapters.qq import MessageSegment,MessageEvent
from src.clover_music.cloud_music.cloud_music import * from src.clover_music.cloud_music.cloud_music import *
from src.clover_image.delete_file import delete_file from src.clover_image.delete_file import delete_file
unikey_cache = {'unikey': None, 'expires': 0}
music = on_command("点歌", rule=to_me(), priority=10,block=False) music = on_command("点歌", rule=to_me(), priority=10,block=False)
@music.handle() @music.handle()
async def handle_function(msg: MessageEvent): async def handle_function(msg: MessageEvent):
qr_path = ""
keyword = msg.get_plaintext().replace("/点歌", "").strip(" ") keyword = msg.get_plaintext().replace("/点歌", "").strip(" ")
if keyword == "": if keyword == "":
await music.finish("\n请输入“/点歌+歌曲名”喔🎶") await music.finish("\n请输入“/点歌+歌曲名”喔🎶")
#获取登录信息 可以获取更换高音质 #获取登录信息
session = requests.session() session = requests.session()
if not os.path.exists('cloud_music_cookies.cookie'): if not os.path.exists('cloud_music_cookies.cookie'):
with open('cloud_music_cookies.cookie', 'wb') as f: with open('cloud_music_cookies.cookie', 'wb') as f:
@ -26,9 +29,18 @@ async def handle_function(msg: MessageEvent):
session, status,user_id = await netease_cloud_music_is_login(session) session, status,user_id = await netease_cloud_music_is_login(session)
if not status: if not status:
await music.send("登录失效,请联系管理员进行登录") await music.send("登录失效,请联系管理员进行登录")
current_time = time.time()
# 检查缓存是否有效二维码有效期5分钟)
if unikey_cache['unikey'] and current_time < unikey_cache['expires']:
unikey = unikey_cache['unikey']
else:
# 获取新 unikey 并设置过期时间
unikey = await get_qr_key(session) unikey = await get_qr_key(session)
path = await create_qr_code(unikey) unikey_cache.update({
'unikey': unikey,
'expires': current_time + 300 # 大约是5分钟有效期 失效时间会有几秒误差
})
qr_path = await create_qr_code(unikey)
"""是否要发送到QQ上面登录 """ """是否要发送到QQ上面登录 """
# await clover_music.send(MessageSegment.file_image(Path(path))) # await clover_music.send(MessageSegment.file_image(Path(path)))
"""是否要发送到QQ上面登录 """ """是否要发送到QQ上面登录 """
@ -42,11 +54,12 @@ async def handle_function(msg: MessageEvent):
print('已确认,登入成功!') print('已确认,登入成功!')
break break
else: else:
print('二维码失效,请重获取!')
break break
await asyncio.sleep(2) await asyncio.sleep(5)
with open('cloud_music_cookies.cookie', 'wb') as f: with open('cloud_music_cookies.cookie', 'wb') as f:
pickle.dump(session.cookies, f) pickle.dump(session.cookies, f)
else:
#搜索歌曲 #搜索歌曲
song_id,song_name,singer,song_url = await netease_music_search(keyword,session) song_id,song_name,singer,song_url = await netease_music_search(keyword,session)
song_name = str(song_name).replace(".", "·").replace("/", "") song_name = str(song_name).replace(".", "·").replace("/", "")
@ -63,7 +76,6 @@ async def handle_function(msg: MessageEvent):
await music.send("歌曲音频获取失败了Σヽ(゚Д ゚; )ノ,请重试。") await music.send("歌曲音频获取失败了Σヽ(゚Д ゚; )ノ,请重试。")
else: else:
await music.send(MessageSegment.file_audio(Path(output_silk_path))) await music.send(MessageSegment.file_audio(Path(output_silk_path)))
#删除临时文件 #删除临时文件
await delete_file(output_silk_path) await delete_file(output_silk_path)
await music.finish() await music.finish()