Building/download_from_data_links.py

#!/usr/bin/env python3
import argparse
import json
import os
import re
import ssl
import urllib.error
import urllib.parse
import urllib.request


def sanitize_filename(name: str) -> str:
    sanitized = name.strip()
    sanitized = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', sanitized)
    sanitized = re.sub(r'\s+', ' ', sanitized)
    return sanitized


def unique_path(path: str) -> str:
    base, ext = os.path.splitext(path)
    counter = 1
    while os.path.exists(path):
        path = f"{base} ({counter}){ext}"
        counter += 1
    return path


def download_file(url: str, dest_path: str, verify_ssl: bool = True) -> None:
    req = urllib.request.Request(url, headers={
        'User-Agent': 'Mozilla/5.0 (Python urllib)'
    })
    context = None
    if not verify_ssl:
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE

    with urllib.request.urlopen(req, timeout=60, context=context) as response:
        data = response.read()
    with open(dest_path, 'wb') as out_file:
        out_file.write(data)


def get_extension_from_url(url: str) -> str:
    path = urllib.parse.urlparse(url).path
    ext = os.path.splitext(path)[1]
    return ext if ext else '.bin'


def main() -> None:
    parser = argparse.ArgumentParser(
        description='从 数据链接.txt 下载每个 file_url，并根据 name 重命名到指定目录。'
    )
    parser.add_argument(
        '--input', '-i', default='数据链接.txt',
        help='输入 JSON 文件，默认：数据链接.txt'
    )
    parser.add_argument(
        '--output-dir', '-o', default='downloads',
        help='输出目录，默认：downloads'
    )
    parser.add_argument(
        '--ignore-ssl', action='store_true',
        help='忽略 SSL 证书验证，适用于证书链不完整的服务器'
    )
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)

    with open(args.input, 'r', encoding='utf-8') as infile:
        source = json.load(infile)

    records = []
    if isinstance(source, dict):
        records = source.get('data', {}).get('records', [])
    elif isinstance(source, list):
        records = source

    if not isinstance(records, list):
        raise ValueError('无法从输入文件中解析 records 列表。')

    for record in records:
        if not isinstance(record, dict):
            continue
        name = record.get('name')
        url = record.get('file_url')
        if not name or not url:
            print('跳过：缺少 name 或 file_url。', record)
            continue

        clean_name = sanitize_filename(name)
        ext = get_extension_from_url(url)
        dest_name = f'{clean_name}{ext}'
        dest_path = os.path.join(args.output_dir, dest_name)
        dest_path = unique_path(dest_path)

        try:
            print(f'正在下载: {url} -> {dest_path}')
            download_file(url, dest_path, verify_ssl=not args.ignore_ssl)
        except urllib.error.URLError as exc:
            if 'CERTIFICATE_VERIFY_FAILED' in str(exc) and not args.ignore_ssl:
                print('SSL 证书验证失败，重试时忽略证书验证。')
                try:
                    download_file(url, dest_path, verify_ssl=False)
                except Exception as exc2:
                    print(f'下载失败: {url}\n原因: {exc2}')
            else:
                print(f'下载失败: {url}\n原因: {exc}')
        except Exception as exc:
            print(f'保存失败: {dest_path}\n原因: {exc}')

    print('下载完成。')


if __name__ == '__main__':
    main()