本帖最后由 话痨司机啊 于 2022-11-12 19:01 编辑
【多线程源码程序下载】https://pan.baidu.com/s/1MuC6H_EwnmAxM0_YuEzIFg?pwd=3ikq 提取码: 3ikq
【异步程序下载源码】https://pan.baidu.com/s/1ThCKKGlTJfQBu951soLFIQ?pwd=quzw 提取码: quzw
【演示说明下载】https://pan.baidu.com/s/1RgZ-X8YNk2rGfHRp3fJK6g?pwd=1cbn 提取码: 1cbn
【喜马拉雅下载程序成品EXE】链接： https://pan.baidu.com/s/1ZbeHr4FGgunKGPHsbAFsfg?pwd=t5wm 提取码: t5wm 【20221112更新】
【特别说明】
坑1：对于执行源码的朋友，你们得改一下 execjs库的源码，路径是所在库lib/site-packages/execjs/_external_runtime.py 第99行 增加个encoding = 'utf-8',应该属于源码库的bug，如果js代码里没有特殊字符是不会报错的。
坑2：需要安装Nodejs 下载地址：https://nodejs.org/zh-cn/download/ ，windows自带的 'JScript'有可能执行不了扣下来的JS代码。（成品运行不了也需要安装这个Nodejs，log日志不报错应该就是这个原因）
特别说明截图:
效果图:
用异步协程写了一个程序，主代码如下（感觉异步写起来简单很多）：
[Python] 纯文本查看 复制代码 import asyncio
import logging
import aiofiles
import aiohttp
from rich.console import Console
from contants.contants import NetConstant
from untils.util import *
from untils.ximalaya_execjs import *
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename='ximalaya.log',
filemode='a')
class Ximalaya:
def __init__(self) -> None:
# 初始化解密类
self.exjs = Execjs_Xi_Ma()
# 专辑api接口
self.albumId_url = 'https://www.ximalaya.com/revision/album/v1/getTracksList'
# 音频下载api接口
self.soundapi_url = 'https://mobile.ximalaya.com/mobile-playpage/track/v3/baseInfo'
# 初始化常量类
self.contents = NetConstant()
# 初始化输出类
self.console = Console()
@retry(logger=logging)
async def get_album_tracks_list(self,session:aiohttp.ClientSession,id:str,pageNum:int) -> list:
'''
获取专辑列表
'''
params = {'albumId':id,'pageNum':pageNum}
async with session.get(self.albumId_url,headers=self.contents.headers,params=params) as response:
album_list = await response.json(content_type='text/plain',encoding='utf-8')
return album_list['data']["tracks"]
async def js_from_string(self,session,trackID:str) -> str:
'''
解密地址
'''
url = f'{self.soundapi_url}/{time.time()}'
params = {'device':'web','trackId':trackID}
async with session.get(url,headers=self.contents.headers,params=params) as response:
_result = await response.json(content_type='application/json',encoding='utf-8')
result = _result['trackInfo']['playUrlList'][0]['url']
return self.exjs.js_from_string(result)
@retry(logger=logging)
async def tracks_downloads(self,session:aiohttp.ClientSession,url:str,filename:str) -> list:
'''
下载并保存音频
'''
async with session.get(url,headers=self.contents.headers) as response:
mkdir(os.path.dirname(filename))
await self.save_audio(await response.read(),filename)
@retry(logger=logging)
async def save_audio(self, content:bytes, filename:str) -> None:
'''
保存音频
'''
async with aiofiles.open(filename, 'wb') as f:
await f.write(content)
now = time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
self.console.print(f'[yellow]创建时间:{now}\n {filename} 下载完成!')
self.console.print(f"{'[blue]-'*100}")
def user_input(self) -> str:
'''
获取用户输入
'''
_url = self.console.input('请输入专辑网址:')
# _url = 'https://www.ximalaya.com/album/9723091'
_file_name = self.console.input('请输入保存的专辑文件夹名称:')
_file_name = _file_name if _file_name else '合辑'
file_name = filename_filter(_file_name)
id_url = _url.split('/')[-1]
return id_url,file_name
async def main(self):
id,file_name = self.user_input()
pageNum = 1
async with aiohttp.ClientSession() as session:
# 异步获取pageNum错误会报超时，未设置pageNum边界,取消timeout
# async with async_timeout.timeout(10):
while True:
result = await self.get_album_tracks_list(session,id,pageNum)
pageNum += 1
for item in result:
title = item['title']
trackId = item['trackId']
url = await self.js_from_string(session,trackId)
await self.tracks_downloads(session,url,f'{file_name}/{title}.m4a')
if __name__ == "__main__":
ximalaya = Ximalaya()
loop = asyncio.get_event_loop()
loop.run_until_complete(ximalaya.main())
| 
[复制链接]
发表于 2022-5-13 23:37
|
发表于 2022-5-15 10:32
