async def main(search_keyword, singer_name, page):
async with aiohttp.ClientSession() as session:
url = 'https://complexsearch.kugou.com/v2/search/song'
t = time.time()
params = {
'callback': 'callback123',
'page': page,
'keyword': search_keyword,
'pagesize': '30',
'bitrate': '0',
'isfuzzy': '0',
'inputtype': '0',
'platform': 'WebFilter',
'userid': '0',
'clientver': '2000',
'iscorrection': '1',
'privilege_filter': '0',
'token': '',
'srcappid': '2919',
'clienttime': str(t),
'mid': str(t),
'uuid': str(t),
'dfid': '-',
'signature': generate_signature(t, search_keyword, page)
}
async with session.get(url=url, headers=headers, params=params) as resp:
if resp.status == 200:
resp_text = await resp.text()
json_data = json.loads(resp_text[12:-2:])
status = json_data['status']
song_list = []
if status == 1:
for item in json_data['data']['lists']:
song_info = {'SongName': re.sub(r"[/\\:*?\"<>|]", "_", item['SongName']),
'AlbumID': item['AlbumID'],
'FileHash': item['FileHash'], 'SQFileHash': item['SQFileHash'],
'HQFileHash': item['HQFileHash'], 'MvHash': item['MvHash'],
'Audioid': item['Audioid'],
'SingerName': re.sub(r"[/\\:*?\"<>|]", "_", item['SingerName'])}
song_list.append(song_info)
else:
logging.error(f'获取歌曲列表失败: {json_data["error_msg"]}')
if singer_name is not None: # 根据歌手名过滤 song_list = [song_info for song_info in song_list if song_info["SingerName"] == singer_name]
n = len(song_list)
if n > 0:
logging.info(f'获取歌曲列表成功，总共{len(song_list)}首， 准备下载...')
tasks = [asyncio.create_task(get_song_play_addr(song)) for song in song_list]
await asyncio.wait(tasks)
else:
logging.error(f'歌曲搜索结果为: {n}, 请更换关键词或者歌手重试')
else:
logging.error(f'错误代码: {resp.status}, 稍后重试')
async def get_song_play_addr(song_info):
async with aiohttp.ClientSession() as session:
url = 'https://wwwapi.kugou.com/yy/index.php'
params = {
'r': 'play/getdata',
'callback': 'jQuery191035601158181920933_1653052693184',
'hash': song_info['FileHash'],
'dfid': '2mSZvv2GejpK2VDsgh0K7U0O',
'appid': '1014',
'mid': 'c18aeb062e34929c6e90e3af8f7e2512',
'platid': '4',
'album_id': song_info['AlbumID'],
'_': '1653050047389'
}
async with session.get(url=url, headers=headers, params=params) as resp:
if resp.status == 200:
resp_text = await resp.text()
json_data = json.loads(resp_text[42:-2:].replace('\\', '').encode('utf8').decode('unicode_escape'))
data = json_data['data'] if json_data.get('data') is not None else None
mp3_url = data['play_url'] if data is not None and data.get('play_url') is not None else ""
if not mp3_url:
logging.error("未找到mp3_url下载链接")
else:
await save_mp3(mp3_url, song_info['SongName'], song_info['SingerName'])
else:
logging.info(f'错误代码: {resp.status}, 请稍后再试')
async def save_mp3(url, song_name, singer_name, default_sv_dir="music"):
os.makedirs(default_sv_dir, exist_ok=True)
async with aiohttp.ClientSession() as session:
async with session.get(url=url, headers=headers) as resp:
async with aiofiles.open(f'{default_sv_dir}/{song_name}-{singer_name}.mp3', mode='wb') as f:
await f.write(await resp.content.read())
logging.info(f'{song_name}--{singer_name}--下载完成')
if __name__ == '__main__':
logging.basicConfig(filename="kgDownloader.log", format="%(asctime)s %(levelname)s %(message)s", level=logging.INFO)
logging.info(read_me)
parser = argparse.ArgumentParser(
prog='kuGouDownloader',
description='酷狗音乐下载')
parser.add_argument('keyword', help='歌曲关键词')
parser.add_argument('-s', '--singer', help='歌手名')
parser.add_argument('-p', '--page', help='页数', type=int)
args = parser.parse_args()
keyword, singer = args.keyword, args.singer
total_page = args.page if args.page is not None else 1
loop = asyncio.get_event_loop()
for i in range(1, total_page+1):
logging.info(f"搜索: {keyword}, 歌手: {singer if singer is not None else '' }, 搜索页数: {i}")
loop.run_until_complete(main(keyword, singer, i)) # 默认下载搜索列表的 第一页 共30首 基于原贴：https://www.52pojie.cn/thread-1638865-1-1.html，优化修改
[复制链接]
发表于 2022-11-28 11:28
