Ubuntu
281 lines
8.7 KiB
Python
281 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Loyal Books MP3 Downloader for Windows
|
|
从 mp3_link.txt 下载 MP3 压缩包
|
|
|
|
使用方法:
|
|
python loyalbooks_downloader.py
|
|
|
|
功能:
|
|
- 读取 mp3_link.txt，逐行下载
|
|
- 支持断点续传
|
|
- 自动更新下载状态 (0->1成功, 0->2失败)
|
|
- 失败链接可手动改为 0 重新尝试
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
print("请先安装 requests: pip install requests")
|
|
exit(1)
|
|
|
|
# 配置
|
|
LINKS_FILE = "mp3_link.txt" # 链接文件
|
|
DOWNLOAD_DIR = Path("downloads") # 下载目录
|
|
DELAY_SECONDS = 10 # 下载间隔（秒）
|
|
MIN_FILE_SIZE = 1024 * 1024 # 最小文件大小 1MB
|
|
MAX_RETRIES = 3 # 最大重试次数
|
|
CHUNK_SIZE = 8192 # 下载块大小
|
|
|
|
|
|
def sanitize_filename(name):
|
|
"""清理文件名"""
|
|
# 移除不安全字符
|
|
name = re.sub(r'[<>:"/\\|?*]', '_', name)
|
|
# 限制长度
|
|
return name[:100]
|
|
|
|
|
|
def parse_links_file():
|
|
"""解析 mp3_link.txt 文件"""
|
|
if not os.path.exists(LINKS_FILE):
|
|
print(f"❌ 找不到文件: {LINKS_FILE}")
|
|
print("请确保 mp3_link.txt 与脚本在同一目录")
|
|
return []
|
|
|
|
links = []
|
|
with open(LINKS_FILE, 'r', encoding='utf-8') as f:
|
|
for line_num, line in enumerate(f, 1):
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
|
|
parts = line.split('|', 1)
|
|
if len(parts) != 2:
|
|
print(f"⚠️ 第 {line_num} 行格式错误，跳过: {line}")
|
|
continue
|
|
|
|
status = parts[0].strip()
|
|
url = parts[1].strip()
|
|
|
|
if status not in ['0', '1', '2']:
|
|
print(f"⚠️ 第 {line_num} 行状态码无效，跳过: {line}")
|
|
continue
|
|
|
|
links.append({
|
|
'line_num': line_num,
|
|
'status': status,
|
|
'url': url,
|
|
'original_line': line
|
|
})
|
|
|
|
return links
|
|
|
|
|
|
def get_local_filename(url):
|
|
"""从 URL 生成本地文件名"""
|
|
# 从 URL 提取路径
|
|
parsed = urlparse(f"http://{url}")
|
|
path_parts = parsed.path.split('/')
|
|
|
|
# 使用最后一部分作为文件名
|
|
if len(path_parts) >= 2:
|
|
# 格式: archive.org/download/xxx/xxx.zip
|
|
# 使用最后两级目录名组合
|
|
folder = sanitize_filename(path_parts[-2]) if len(path_parts) >= 2 else "unknown"
|
|
filename = sanitize_filename(path_parts[-1])
|
|
if not filename.endswith('.zip'):
|
|
filename += '.zip'
|
|
return folder, filename
|
|
else:
|
|
return "unknown", sanitize_filename(path_parts[-1])
|
|
|
|
|
|
def update_link_status(line_num, new_status):
|
|
"""更新 mp3_link.txt 中的状态"""
|
|
try:
|
|
with open(LINKS_FILE, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()
|
|
|
|
# 修改指定行
|
|
if 1 <= line_num <= len(lines):
|
|
original_line = lines[line_num - 1].strip()
|
|
if '|' in original_line:
|
|
parts = original_line.split('|', 1)
|
|
lines[line_num - 1] = f"{new_status}|{parts[1]}\n"
|
|
|
|
with open(LINKS_FILE, 'w', encoding='utf-8') as f:
|
|
f.writelines(lines)
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f" ⚠️ 更新状态失败: {e}")
|
|
return False
|
|
|
|
|
|
def download_file(url, dest_path, retries=0):
|
|
"""
|
|
下载文件，支持断点续传
|
|
返回: (success: bool, size: int)
|
|
"""
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
|
}
|
|
|
|
# 确保目录存在
|
|
dest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 检查是否已存在有效文件
|
|
if dest_path.exists():
|
|
existing_size = dest_path.stat().st_size
|
|
if existing_size >= MIN_FILE_SIZE:
|
|
print(f" ⏭️ 已存在 ({existing_size // 1024 // 1024}MB)，跳过")
|
|
return True, existing_size
|
|
|
|
# 临时文件路径
|
|
temp_path = dest_path.with_suffix('.tmp')
|
|
resume_byte_pos = 0
|
|
|
|
# 检查是否有部分下载的文件
|
|
if temp_path.exists():
|
|
resume_byte_pos = temp_path.stat().st_size
|
|
print(f" 🔄 续传 ({resume_byte_pos // 1024 // 1024}MB)...", end=' ')
|
|
headers['Range'] = f'bytes={resume_byte_pos}-'
|
|
else:
|
|
print(f" 下载中...", end=' ')
|
|
|
|
try:
|
|
response = requests.get(f"http://{url}", headers=headers, timeout=300, stream=True)
|
|
|
|
# 如果服务器不支持断点续传，从头下载
|
|
if response.status_code == 416:
|
|
headers.pop('Range', None)
|
|
resume_byte_pos = 0
|
|
response = requests.get(f"http://{url}", headers=headers, timeout=300, stream=True)
|
|
|
|
response.raise_for_status()
|
|
|
|
# 确定写入模式
|
|
mode = 'ab' if resume_byte_pos > 0 and response.status_code == 206 else 'wb'
|
|
downloaded = resume_byte_pos
|
|
|
|
with open(temp_path, mode) as f:
|
|
for chunk in response.iter_content(chunk_size=CHUNK_SIZE):
|
|
if chunk:
|
|
f.write(chunk)
|
|
downloaded += len(chunk)
|
|
|
|
# 下载完成，重命名为正式文件
|
|
temp_path.rename(dest_path)
|
|
|
|
# 检查文件大小
|
|
final_size = dest_path.stat().st_size
|
|
if final_size < MIN_FILE_SIZE:
|
|
print(f"❌ 文件过小 ({final_size} bytes)")
|
|
return False, final_size
|
|
|
|
print(f"✅ {final_size // 1024 // 1024}MB")
|
|
return True, final_size
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"❌ 网络错误: {e}")
|
|
if retries < MAX_RETRIES:
|
|
print(f" 等待 5 秒后重试 ({retries + 1}/{MAX_RETRIES})...")
|
|
time.sleep(5)
|
|
return download_file(url, dest_path, retries + 1)
|
|
return False, 0
|
|
except Exception as e:
|
|
print(f"❌ 错误: {e}")
|
|
return False, 0
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("=" * 60)
|
|
print("📚 Loyal Books MP3 Downloader")
|
|
print("=" * 60)
|
|
print(f"💾 下载目录: {DOWNLOAD_DIR.absolute()}")
|
|
print(f"🔗 链接文件: {LINKS_FILE}")
|
|
print(f"⏱️ 下载间隔: {DELAY_SECONDS} 秒")
|
|
print()
|
|
|
|
# 解析链接文件
|
|
links = parse_links_file()
|
|
if not links:
|
|
print("❌ 没有需要下载的链接")
|
|
return
|
|
|
|
# 统计
|
|
pending = [l for l in links if l['status'] == '0']
|
|
completed = [l for l in links if l['status'] == '1']
|
|
skipped = [l for l in links if l['status'] == '2']
|
|
|
|
print(f"📊 链接统计:")
|
|
print(f" 总计: {len(links)}")
|
|
print(f" 待下载 (0): {len(pending)}")
|
|
print(f" 已完成 (1): {len(completed)}")
|
|
print(f" 跳过 (2): {len(skipped)}")
|
|
print()
|
|
|
|
if not pending:
|
|
print("✅ 所有文件已下载完成！")
|
|
return
|
|
|
|
print(f"即将开始下载 {len(pending)} 个文件...")
|
|
print("按 Ctrl+C 可随时暂停\n")
|
|
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
try:
|
|
for i, link in enumerate(pending, 1):
|
|
print(f"[{i}/{len(pending)}] 行 {link['line_num']}")
|
|
|
|
# 获取本地文件名
|
|
folder, filename = get_local_filename(link['url'])
|
|
dest_dir = DOWNLOAD_DIR / folder
|
|
dest_path = dest_dir / filename
|
|
|
|
print(f" URL: {link['url'][:60]}...")
|
|
print(f" 保存到: {dest_path}")
|
|
|
|
# 下载
|
|
success, size = download_file(link['url'], dest_path)
|
|
|
|
if success:
|
|
# 更新状态为 1 (已完成)
|
|
update_link_status(link['line_num'], '1')
|
|
success_count += 1
|
|
else:
|
|
# 更新状态为 2 (失败，不再尝试)
|
|
update_link_status(link['line_num'], '2')
|
|
fail_count += 1
|
|
|
|
# 间隔
|
|
if i < len(pending):
|
|
time.sleep(DELAY_SECONDS)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✅ 下载完成！")
|
|
print(f" 成功: {success_count}")
|
|
print(f" 失败: {fail_count}")
|
|
print("=" * 60)
|
|
print()
|
|
print("提示：")
|
|
print(" - 失败的链接可在 mp3_link.txt 中将状态改为 0 重新尝试")
|
|
print(f" - 下载的文件保存在: {DOWNLOAD_DIR.absolute()}")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n⚠️ 用户中断")
|
|
print("已保存进度，下次会继续从断点下载")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|