使用Python的程序批量下载文件到不同文件

dickwolf

如题所示 启用的excel表downloads.xlsx来下载  一个是下载链接 一个是保存路径
程序如下：
import pandas as pd
import requests
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock, Semaphore
import time
from tqdm import tqdm
import argparse
import sys
from urllib.parse import unquote, urlparse
import mimetypes
from typing import Tuple, Optional
import json

class MultiThreadDownloader:
    """多线程批量下载器 - 支持智能处理文件后缀名"""
   
    def __init__(self, excel_path, url_column, path_column, sheet_name=0,
                 max_workers=10, timeout=30, remove_extension=True,
                 detect_extension=False, default_extension=''):
        """
        初始化多线程下载器
        
        Args:
            excel_path: Excel文件路径
            url_column: 下载链接所在的列名或索引
            path_column: 保存路径所在的列名或索引
            sheet_name: 工作表名称或索引（默认：0）
            max_workers: 最大线程数（默认：10）
            timeout: 请求超时时间（秒，默认：30）
            remove_extension: 是否去掉文件后缀名（默认：True）
            detect_extension: 是否自动检测文件后缀名（默认：False）
            default_extension: 当无法检测时的默认后缀名（如：'.bin'）
        """
        self.excel_path = excel_path
        self.url_column = url_column
        self.path_column = path_column
        self.sheet_name = sheet_name
        self.max_workers = max_workers
        self.timeout = timeout
        self.remove_extension = remove_extension
        self.detect_extension = detect_extension
        self.default_extension = default_extension if default_extension.startswith('.') else f'.{default_extension}'
        
        # 线程安全统计
        self.lock = Lock()
        self.success_count = 0
        self.fail_count = 0
        self.failed_tasks = []
        self.total_size = 0
        self.downloaded_size = 0
        self.renamed_count = 0
        self.detected_count = 0
        
        # 进度条
        self.pbar = None
        
        # 读取数据
        self.tasks = self._read_excel()
        
    def _read_excel(self):
        """读取Excel文件"""
        try:
            df = pd.read_excel(self.excel_path, sheet_name=self.sheet_name)
            urls = df[self.url_column].tolist()
            paths = df[self.path_column].tolist()
            
            tasks = []
            for i, (url, path) in enumerate(zip(urls, paths)):
                if pd.notna(url) and pd.notna(path):
                    tasks.append({
                        'id': i + 1,
                        'row': i + 2,
                        'url': str(url).strip(),
                        'path': str(path).strip()
                    })
            
            print(f"✓ 成功读取Excel文件: {self.excel_path}")
            print(f"  工作表: {self.sheet_name if isinstance(self.sheet_name, str) else f'第{self.sheet_name+1}个'}")
            print(f"  有效任务: {len(tasks)} 个")
            print(f"  线程数: {self.max_workers}")
            print(f"  去掉后缀名: {'是' if self.remove_extension else '否'}")
            print(f"  自动检测后缀: {'是' if self.detect_extension else '否'}")
            if self.default_extension:
                print(f"  默认后缀: {self.default_extension}")
            
            return tasks
            
        except Exception as e:
            print(f"✗ 读取Excel文件失败: {e}")
            return []
   
    def _get_extension_from_url(self, url: str) -> str:
        """从URL中提取文件后缀名"""
        parsed_url = urlparse(url)
        path = parsed_url.path
        filename = os.path.basename(path)
        
        # 获取扩展名
        extension = os.path.splitext(filename)[1]
        
        # 处理查询参数中的扩展名
        if not extension and '?' in url:
            import urllib.parse
            query = urllib.parse.parse_qs(parsed_url.query)
            for key in ['format', 'ext', 'type']:
                if key in query:
                    extension = f".{query[key][0]}"
                    break
        
        return extension
   
    def _detect_extension_from_content(self, url: str) -> Tuple[str, str]:
        """
        通过HTTP响应头检测文件类型
        
        Returns:
            (extension, content_type)
        """
        try:
            # 发送HEAD请求
            response = requests.head(url, timeout=10, allow_redirects=True)
            
            if response.status_code == 200:
                content_type = response.headers.get('content-type', '')
               
                # MIME类型映射
                mime_to_ext = {
                    'application/pdf': '.pdf',
                    'image/jpeg': '.jpg',
                    'image/png': '.png',
                    'image/gif': '.gif',
                    'image/webp': '.webp',
                    'image/bmp': '.bmp',
                    'video/mp4': '.mp4',
                    'video/mpeg': '.mpg',
                    'video/quicktime': '.mov',
                    'video/x-msvideo': '.avi',
                    'audio/mpeg': '.mp3',
                    'audio/wav': '.wav',
                    'audio/ogg': '.ogg',
                    'application/zip': '.zip',
                    'application/x-rar-compressed': '.rar',
                    'application/x-7z-compressed': '.7z',
                    'application/msword': '.doc',
                    'application/vnd.openxmlformats-officedocument.wordprocessingml.document': '.docx',
                    'application/vnd.ms-excel': '.xls',
                    'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': '.xlsx',
                    'application/vnd.ms-powerpoint': '.ppt',
                    'application/vnd.openxmlformats-officedocument.presentationml.presentation': '.pptx',
                    'text/plain': '.txt',
                    'text/html': '.html',
                    'text/css': '.css',
                    'text/javascript': '.js',
                    'application/json': '.json',
                    'application/xml': '.xml',
                }
               
                # 精确匹配
                if content_type in mime_to_ext:
                    return mime_to_ext[content_type], content_type
               
                # 模糊匹配
                main_type = content_type.split('/')[0]
                if main_type == 'image':
                    sub_type = content_type.split('/')[1]
                    if sub_type in ['jpeg', 'png', 'gif', 'webp', 'bmp']:
                        return f'.{sub_type}', content_type
                    return '.jpg', content_type
                elif main_type == 'video':
                    return '.mp4', content_type
                elif main_type == 'audio':
                    return '.mp3', content_type
                elif main_type == 'text':
                    return '.txt', content_type
                elif main_type == 'application':
                    # 尝试从子类型获取
                    sub_type = content_type.split('/')[1]
                    if sub_type in ['pdf', 'zip', 'json', 'xml']:
                        return f'.{sub_type}', content_type
                    return '.bin', content_type
               
                # 从Content-Disposition获取
                content_disposition = response.headers.get('content-disposition', '')
                if 'filename=' in content_disposition:
                    import re
                    match = re.search(r'filename[^;=\n]*=(([\'"]).*?\2|[^;\n]*)', content_disposition)
                    if match:
                        filename = match.group(1).strip('"\'')
                        extension = os.path.splitext(filename)[1]
                        if extension:
                            return extension, content_type
            
            return '', ''
            
        except Exception as e:
            return '', ''
   
    def _get_filename_without_extension(self, filename: str) -> str:
        """获取不带后缀的文件名"""
        return os.path.splitext(filename)[0]
   
    def _get_final_filename(self, original_filename: str, detected_extension: str = '') -> str:
        """
        获取最终文件名
        
        Returns:
            最终文件名
        """
        if self.remove_extension:
            # 去掉所有后缀名
            return self._get_filename_without_extension(original_filename)
        else:
            # 保留后缀名：如果没有后缀，尝试使用检测到的
            current_ext = os.path.splitext(original_filename)[1]
            if current_ext:
                return original_filename
            elif detected_extension and self.detect_extension:
                # 添加检测到的后缀
                with self.lock:
                    self.detected_count += 1
                return original_filename + detected_extension
            elif self.default_extension:
                # 使用默认后缀
                return original_filename + self.default_extension
            else:
                return original_filename
   
    def _prepare_save_path(self, task) -> Tuple[str, str]:
        """
        准备保存路径
        
        Returns:
            (final_save_path, temp_path)
        """
        url = task['url']
        raw_path = task['path']
        
        # 处理相对路径
        save_path = os.path.expanduser(raw_path)
        
        # 判断是目录还是文件路径
        is_directory = save_path.endswith('/') or save_path.endswith('\\') or not os.path.splitext(save_path)[1]
        
        if is_directory:
            # 是目录，从URL提取原始文件名
            parsed_url = urlparse(url)
            original_filename = os.path.basename(parsed_url.path)
            if not original_filename:
                original_filename = f"downloaded_file_{task['id']}"
            original_filename = unquote(original_filename)
            
            # 检测文件类型（如果需要）
            detected_extension = ''
            if self.detect_extension and not os.path.splitext(original_filename)[1]:
                # 原始文件名无后缀，尝试检测
                detected_extension, content_type = self._detect_extension_from_content(url)
                if detected_extension:
                    print(f"  检测到文件类型: {detected_extension} ({content_type})")
            
            # 确定最终文件名
            final_filename = self._get_final_filename(original_filename, detected_extension)
            
            # 确保目录存在
            os.makedirs(save_path, exist_ok=True)
            final_save_path = os.path.join(save_path, final_filename)
            temp_path = final_save_path + '.temp'
            
        else:
            # 是完整文件路径
            dir_path = os.path.dirname(save_path)
            original_filename = os.path.basename(save_path)
            
            # 检测文件类型（如果需要）
            detected_extension = ''
            if self.detect_extension and not os.path.splitext(original_filename)[1]:
                detected_extension, content_type = self._detect_extension_from_content(url)
                if detected_extension:
                    print(f"  检测到文件类型: {detected_extension} ({content_type})")
            
            # 确定最终文件名
            final_filename = self._get_final_filename(original_filename, detected_extension)
            
            # 确保目录存在
            if dir_path:
                os.makedirs(dir_path, exist_ok=True)
            
            final_save_path = os.path.join(dir_path, final_filename)
            temp_path = final_save_path + '.temp'
        
        return final_save_path, temp_path
   
    def download_single_file(self, task) -> Tuple[bool, int, str, str, str]:
        """
        下载单个文件（供多线程调用）
        
        Returns:
            (success, row_num, url, save_path, message)
        """
        url = task['url']
        raw_path = task['path']
        row_num = task['row']
        
        # 准备保存路径
        final_save_path, temp_path = self._prepare_save_path(task)
        filename = os.path.basename(final_save_path)
        
        # 检查文件是否已存在且完整
        if os.path.exists(final_save_path):
            try:
                # 获取远程文件大小
                response = requests.head(url, timeout=10, allow_redirects=True)
                if response.status_code == 200:
                    remote_size = int(response.headers.get('content-length', 0))
                    local_size = os.path.getsize(final_save_path)
                    
                    if remote_size > 0 and local_size == remote_size:
                        self.pbar.set_description(f"✓ 跳过: {filename}")
                        with self.lock:
                            self.success_count += 1
                        self.pbar.update(1)
                        return True, row_num, url, final_save_path, "文件已存在"
                    else:
                        self.pbar.set_description(f"⚠ 不完整，重新下载: {filename}")
                        os.remove(final_save_path)
            except:
                # 无法验证，跳过
                self.pbar.set_description(f"⚠ 跳过(未验证): {filename}")
                with self.lock:
                    self.success_count += 1
                self.pbar.update(1)
                return True, row_num, url, final_save_path, "文件已存在"
        
        try:
            # 下载文件
            response = requests.get(url, stream=True, timeout=self.timeout)
            response.raise_for_status()
            
            # 获取文件大小
            content_length = response.headers.get('content-length')
            if content_length:
                file_size = int(content_length)
                with self.lock:
                    self.total_size += file_size
            
            # 写入临时文件
            with open(temp_path, 'wb') as f:
                downloaded = 0
                for chunk in response.iter_content(chunk_size=8192 * 4):
                    if chunk:
                        f.write(chunk)
                        downloaded += len(chunk)
                        with self.lock:
                            self.downloaded_size += len(chunk)
                        
                        # 更新进度描述（可选）
                        if content_length and downloaded % (1024 * 1024) < 8192:  # 每1MB更新一次
                            progress = (downloaded / file_size) * 100
                            self.pbar.set_description(f"下载中: {filename} ({progress:.1f}%)")
            
            # 下载完成，重命名为最终文件名
            os.rename(temp_path, final_save_path)
            
            # 显示状态
            status = "完成"
            if self.remove_extension:
                status = "完成(无后缀)"
            elif self.detect_extension and not os.path.splitext(filename)[1]:
                status = "完成(已检测)"
            
            self.pbar.set_description(f"&#10003; {status}: {filename}")
            self.pbar.update(1)
            
            with self.lock:
                self.success_count += 1
            
            return True, row_num, url, final_save_path, status
            
        except requests.exceptions.Timeout:
            error_msg = f"超时（{self.timeout}秒）"
            self.pbar.set_description(f"&#10007; 超时: {filename}")
            self.pbar.update(1)
            with self.lock:
                self.fail_count += 1
                self.failed_tasks.append({
                    'row': row_num,
                    'url': url,
                    'path': raw_path,
                    'error': error_msg
                })
            if os.path.exists(temp_path):
                os.remove(temp_path)
            return False, row_num, url, raw_path, error_msg
            
        except Exception as e:
            error_msg = str(e)
            self.pbar.set_description(f"&#10007; 失败: {filename}")
            self.pbar.update(1)
            with self.lock:
                self.fail_count += 1
                self.failed_tasks.append({
                    'row': row_num,
                    'url': url,
                    'path': raw_path,
                    'error': error_msg
                })
            if os.path.exists(temp_path):
                os.remove(temp_path)
            return False, row_num, url, raw_path, error_msg
   
    def start_download(self):
        """启动多线程下载"""
        if not self.tasks:
            print("没有找到下载任务")
            return
        
        print("\n" + "=" * 60)
        print("开始多线程批量下载")
        print("=" * 60)
        
        start_time = time.time()
        
        # 创建进度条
        self.pbar = tqdm(total=len(self.tasks), desc="下载进度", unit="个", ncols=80)
        
        # 使用线程池
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有任务
            futures = {executor.submit(self.download_single_file, task): task for task in self.tasks}
            
            # 等待所有任务完成
            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    task = futures[future]
                    with self.lock:
                        self.fail_count += 1
                        self.failed_tasks.append({
                            'row': task['row'],
                            'url': task['url'],
                            'path': task['path'],
                            'error': str(e)
                        })
                    self.pbar.update(1)
        
        self.pbar.close()
        
        elapsed_time = time.time() - start_time
        self._print_summary(elapsed_time)
   
    def _print_summary(self, elapsed_time):
        """打印下载统计信息"""
        print("\n" + "=" * 60)
        print("下载完成统计")
        print("=" * 60)
        print(f"  总任务数: {len(self.tasks)}")
        print(f"  成功: {self.success_count} 个 &#10003;")
        print(f"  失败: {self.fail_count} 个 &#10007;")
        print(f"  成功率: {self.success_count/len(self.tasks)*100:.1f}%")
        print(f"  总耗时: {elapsed_time:.2f} 秒")
        print(f"  平均速度: {len(self.tasks)/elapsed_time:.2f} 个/秒")
        
        if self.detect_extension and self.detected_count > 0:
            print(f"  自动检测文件类型: {self.detected_count} 个")
        
        if self.total_size > 0:
            speed_mb = (self.downloaded_size / 1024 / 1024) / elapsed_time
            print(f"  下载总量: {self.downloaded_size / 1024 / 1024:.2f} MB")
            print(f"  平均网速: {speed_mb:.2f} MB/s")
        
        if self.failed_tasks:
            print("\n失败的下载任务：")
            for task in self.failed_tasks[:10]:
                print(f"  行号 {task['row']}: {task['url'][:80]}")
                print(f"    错误: {task['error']}")
            if len(self.failed_tasks) > 10:
                print(f"  ... 还有 {len(self.failed_tasks) - 10} 个失败任务")
        
        # 保存失败任务
        if self.failed_tasks:
            fail_file = "failed_tasks.csv"
            fail_df = pd.DataFrame(self.failed_tasks)
            fail_df.to_csv(fail_file, index=False, encoding='utf-8-sig')
            print(f"\n失败任务已保存到: {fail_file}")
   
    def save_progress(self):
        """保存下载进度"""
        progress_file = "download_progress.json"
        progress = {
            'completed': self.success_count,
            'failed': self.fail_count,
            'failed_tasks': self.failed_tasks,
            'total_tasks': len(self.tasks),
            'timestamp': time.time(),
            'remove_extension': self.remove_extension,
            'detect_extension': self.detect_extension
        }
        with open(progress_file, 'w', encoding='utf-8') as f:
            json.dump(progress, f, ensure_ascii=False, indent=2)
        print(f"\n进度已保存到: {progress_file}")

def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='多线程批量下载工具 - 从Excel读取下载链接和保存位置，智能处理文件后缀名',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
文件后缀名处理策略：
  1. 去掉后缀名 (--remove-extension): 移除所有文件扩展名（默认）
  2. 保留后缀名 (--keep-extension): 保留原有后缀名
  3. 自动检测 (--detect-extension): 从HTTP响应头检测文件类型，自动添加后缀
  4. 默认后缀 (--default-extension .bin): 当无法检测时使用默认后缀

示例：
  # 去掉所有后缀名（默认）
  python downloader.py data.xlsx "url" "path"
  
  # 保留后缀名
  python downloader.py data.xlsx "url" "path" --keep-extension
  
  # 自动检测并添加后缀
  python downloader.py data.xlsx "url" "path" --detect-extension
  
  # 自动检测，无法检测时使用默认后缀
  python downloader.py data.xlsx "url" "path" --detect-extension --default-extension .bin
  
  # 设置线程数
  python downloader.py data.xlsx "url" "path" --workers 20 --timeout 60
  
  # 使用列索引
  python downloader.py data.xlsx 0 1 --sheet Sheet1 --workers 15
        """
    )
   
    parser.add_argument('excel_path', help='Excel文件路径')
    parser.add_argument('url_column', help='下载链接所在的列名或索引（0开始）')
    parser.add_argument('path_column', help='保存路径所在的列名或索引（0开始）')
    parser.add_argument('--sheet', default=0, help='工作表名称或索引（默认：0）')
    parser.add_argument('--workers', '-w', type=int, default=10,
                       help='线程数（默认：10）')
    parser.add_argument('--timeout', '-t', type=int, default=30,
                       help='请求超时时间/秒（默认：30）')
   
    # 后缀名处理选项
    group = parser.add_mutually_exclusive_group()
    group.add_argument('--remove-extension', action='store_true',
                       help='去掉所有文件后缀名（默认）')
    group.add_argument('--keep-extension', action='store_true',
                       help='保留文件后缀名')
    group.add_argument('--detect-extension', action='store_true',
                       help='自动检测文件类型并添加后缀')
   
    parser.add_argument('--default-extension', default='',
                       help='当无法检测时的默认后缀（如：.bin）')
   
    args = parser.parse_args()
   
    # 确定后缀名处理策略
    remove_extension = args.remove_extension
    detect_extension = args.detect_extension
   
    if args.keep_extension:
        remove_extension = False
        detect_extension = False
    elif not remove_extension and not detect_extension:
        # 默认行为：去掉后缀名
        remove_extension = True
   
    # 处理列名/索引
    url_col = args.url_column
    path_col = args.path_column
    if url_col.isdigit():
        url_col = int(url_col)
    if path_col.isdigit():
        path_col = int(path_col)
   
    # 创建下载器
    downloader = MultiThreadDownloader(
        excel_path=args.excel_path,
        url_column=url_col,
        path_column=path_col,
        sheet_name=args.sheet,
        max_workers=args.workers,
        timeout=args.timeout,
        remove_extension=remove_extension,
        detect_extension=detect_extension,
        default_extension=args.default_extension
    )
   
    # 运行下载
    try:
        downloader.start_download()
    except KeyboardInterrupt:
        print("\n\n用户中断下载")
        downloader.save_progress()
        sys.exit(0)

if __name__ == "__main__":
    # 如果没有命令行参数，使用示例配置
    if len(sys.argv) == 1:
        print("=" * 70)
        print("多线程Excel批量下载工具 - 智能处理无后缀文件")
        print("=" * 70)
        print("\n文件后缀名处理方案：")
        print("  1. 去掉后缀名（默认）：下载后所有文件都没有扩展名")
        print("  2. 保留后缀名：保持原有文件名")
        print("  3. 自动检测：从HTTP响应头检测真实文件类型并添加后缀")
        print("\n使用示例：")
        print("  # 方案1: 去掉所有后缀（默认）")
        print("  python downloader.py data.xlsx url path")
        print("\n  # 方案2: 保留原有后缀")
        print("  python downloader.py data.xlsx url path --keep-extension")
        print("\n  # 方案3: 自动检测文件类型")
        print("  python downloader.py data.xlsx url path --detect-extension")
        print("\n  # 方案3增强: 自动检测 + 默认后缀")
        print("  python downloader.py data.xlsx url path --detect-extension --default-extension .bin")
        print("\n  # 高并发下载（20线程）")
        print("  python downloader.py data.xlsx url path --workers 20")
        print("\n" + "=" * 70)
        
        # 示例配置
        excel_file = "downloads.xlsx"
        if os.path.exists(excel_file):
            downloader = MultiThreadDownloader(
                excel_path=excel_file,
                url_column="下载链接",
                path_column="保存路径",
                max_workers=10,
                remove_extension=True  # 改为 False 保留后缀，或设置 detect_extension=True
            )
            try:
                downloader.start_download()
            except KeyboardInterrupt:
                print("\n\n下载已中断")
                downloader.save_progress()
        else:
            print(f"找不到示例文件: {excel_file}")
            print("请创建Excel文件或使用命令行参数指定文件")
    else:
        main()

苏紫方璇

代码插入可以参考置顶帖
【公告】发帖代码插入以及添加链接教程（有福利）
https://www.52pojie.cn/thread-713042-1-1.html
(出处: 吾爱破解论坛)