SanYeCao-Nonebot/src/clover_music/cloud_music/cloud_music.py

# -*- coding: utf-8 -*-
import os
import qrcode
import base64
import codecs
import json
import requests
from io import BytesIO
from Crypto.Cipher import AES
from graiax import silkcoder
import src.clover_music.cloud_music.agent as agent
from src.clover_image.delete_file import delete_file


requests.packages.urllib3.disable_warnings()
headers = {'User-Agent': agent.get_user_agents(), 'Referer': 'https://music.163.com/'}


# 解密params和encSecKey值
def keys(key):
    while len(key) % 16 != 0:
        key += '\0'
    return str.encode(key)

#登录加密
def AES_aes(t, key, iv):
    def p(s): return s + (AES.block_size - len(s) %AES.block_size) * chr(AES.block_size - len(s) % AES.block_size)
    encrypt = str(base64.encodebytes(AES.new( keys(key), AES.MODE_CBC,keys(iv)).encrypt(str.encode(p(t)))),encoding='utf-8')
    return encrypt

def RSA_rsa(i, e, f):
    return format(int(codecs.encode(
        i[::-1].encode('utf-8'), 'hex_codec'), 16) ** int(e, 16) % int(f, 16), 'x').zfill(256)

# 获取的参数
key = agent.S()  # i6c的值
d = str({'key': key, 'type': "1", 'csrf_token': ""})
e = "010001"  # (["流泪", "强"])的值
f = "00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee341f56135fccf695280104e0312ecbda92557c93870114af6c9d05c4f7f0c3685b7a46bee255932575cce10b424d813cfe4875d3e82047b97ddef52741d546b8e289dc6935b3ece0462db0a22b8e7"
g = "0CoJUm6Qyw8W8jud"  # (["爱心", "女孩", "惊恐", "大笑"])的值
iv = "0102030405060708"  # 偏移量
i = agent.a()  # 随机生成长度为16的字符串


def login_params(u):
    if u is None:
        return AES_aes(AES_aes(d, g, iv), i, iv)  # g 和 i 都是key代替
    else:
        return AES_aes(AES_aes(u,g,iv),i,iv)  # g 和 i 都是key代替

def login_encSecKey():
    return RSA_rsa(i, e, f)

def get_music(id):
    return agent.get_params(id)




"""
使用二维码登录网易云音乐，需要先获取二维码的key，然后使用该key生成二维码，扫描二维码登录，最后通过登录接口 返回cookie 保存起来
"""
save_path = os.getcwd()+'/src/clover_music/netease_music'
os.makedirs(save_path, exist_ok=True)
qrcode_path = os.getcwd()+'/src/clover_music'


# 判断cookie是否有效
async def netease_cloud_music_is_login(session):
    """
    验证登录状态
    Args:
        session:

    Returns:

    """
    try:
        session.cookies.load(ignore_discard=True)
    except Exception:
        pass
    csrf_token = session.cookies.get('__csrf')
    if csrf_token is None:
        return session, False,None
    else:
        try:
            loginurl = session.post(f'https://music.163.com/weapi/w/nuser/account/get?csrf_token={csrf_token}',data={'params': login_params(None), 'encSecKey': login_encSecKey()}, headers=headers).json()
            if loginurl['account'] is not None:
                return session, True,loginurl['account']['id']
            else:
                return session, False,None
        except BaseException:
            return session, False,None

# 获取二维码的key
async def get_qr_key(session):
    """
    获取unikey
    Args:
        session:

    Returns:

    """
    url = f"https://music.163.com/weapi/login/qrcode/unikey"
    data = {"params": login_params(None),"encSecKey": login_encSecKey()}
    response = session.post(url, headers=headers,params=data)
    result = json.loads(response.text)
    if result.get("code") == 200:
        unikey = result.get("unikey")
        return unikey
    else:
        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):
    """
    二维码生成
    Args:
        unikey:

    Returns:

    """
    # 添加数据
    png_url = f"http://music.163.com/login?codekey={unikey}"
    qr.add_data(png_url)
    img = qr.make_image()
    a = BytesIO()
    img.save(a, 'png')
    img.save(os.path.join(qrcode_path, 'qrcode.png'))
    return  qrcode_path + '/qrcode.png'

# 检查二维码状态是否被扫描
async def check_qr_code(unikey,session):
    token_url = f"https://music.163.com/weapi/login/qrcode/client/login?csrf_token="
    u = str({'key': unikey, 'type': "1", 'csrf_token': ""})
    qrcode_data = session.post( token_url,data={'params': login_params(u),'encSecKey': login_encSecKey()},headers=headers).json()
    return qrcode_data.get('code')

async def netease_music_search(keyword,session):
    """
    歌曲搜索
    Args:
        keyword:
        session:

    Returns:

    """
    url = "http://music.163.com/api/search/get"
    params = {
        "s": keyword,
        "type": 1,  # 1 表示搜索歌曲，2 表示搜索专辑，3 表示搜索歌手等
        "limit": 10,  # 限制搜索结果的数量
        "offset": 0,  # 搜索结果的偏移量，可用于分页
        "sub": "false",
    }
    response = session.get(url, headers=headers, params=params)
    data = response.json()
    if "result" in data and "songs" in data["result"]:
        songs = data["result"]["songs"]
        if songs:
            first_song = songs[0]  # 获取第一首歌曲
            song_name = first_song["name"]
            singer = first_song["artists"][0]["name"]
            song_id = first_song["id"]
            song_url = f"https://music.163.com/song?id={song_id}"
            print(f"搜索结果：{song_name} - {singer}")
            print(f"歌曲链接：{song_url}")
            return song_id,song_name,singer,song_url
    return None, None, None, None

#仅限于免费歌曲
# def netease_music_download(song_id,song_name,singer,session):
#     if not os.path.exists(save_path):
#         os.makedirs(save_path)
#     download_url = f"http://music.163.com/song/media/outer/url?br=999000&id={song_id}.mp3"
#     response = session.get(download_url, headers=headers)
#     if response.status_code == 200:
#         print(f"正在下载 {song_name} - {singer} 歌曲...")
#         file_path = os.path.join(save_path, f"{song_name}-{singer}.mp3")
#         file_name = os.path.basename(f"{song_name}-{singer}.mp3")
#         with open(file_path, "wb") as f:
#             f.write(response.content)
#         output_silk_path = os.path.join(save_path, os.path.splitext(file_name)[0] + ".silk")
#         # 使用 graiax-silkcoder 进行转换
#         silkcoder.encode(file_path, output_silk_path)
#         return output_silk_path
#     else:
#         return None

#所有歌曲都可以下载
async def netease_music_download(song_id, song_name, singer, session):
    """
    歌曲下载
    Args:
        song_id:
        song_name:
        singer:
        session:

    Returns:

    """
    if not os.path.exists(save_path):
        os.makedirs(save_path)

    # 获取加密后的歌曲id信息
    result = get_music(song_id)
    data = {'params': result['encText'], 'encSecKey': result['encSecKey']}
    download_url = 'https://music.163.com/weapi/song/enhance/player/url/v1?br=999000'
    response_data = session.post(download_url, headers=headers, data=data).json()
    url = response_data['data'][0]['url']
    if url is None:
        return -1
    # 下载歌曲
    try:
        response = requests.get(url, stream=True)
        if response.status_code == 200:
            file_path = os.path.join(save_path, f"{song_name}-{singer}.wav")
            file_name = os.path.basename(f"{song_name}-{singer}.wav")

            with open(file_path, "wb") as file:
                for chunk in response.iter_content(chunk_size=8192):
                    file.write(chunk)

            output_silk_path = os.path.join(save_path, os.path.splitext(file_name)[0] + ".silk")
            # 使用 graiax-silkcoder 进行转换
            silkcoder.encode(file_path, output_silk_path, rate=32000, tencent=True, ios_adaptive=True)
            # 删除临时文件
            await delete_file(file_path)
            return output_silk_path
        else:
            return None
    except requests.RequestException as e:
        return None