#!/usr/bin/env python3 """ 简单的文件下载管理器 支持后台运行,通过Web界面查看下载状态 """ import http.server import socketserver import urllib.request import urllib.error import os import sys import json import time import logging from pathlib import Path from datetime import datetime # 配置 PORT = 8888 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) DOWNLOAD_DIR = os.path.join(SCRIPT_DIR, "download") STATE_FILE = os.path.join(SCRIPT_DIR, "download_state.json") LOG_FILE = os.path.join(SCRIPT_DIR, "download_manager.log") PID_FILE = os.path.join(SCRIPT_DIR, "download_manager.pid") MAX_RETRIES = 2 # 最大重试次数 # 数据源配置 SOURCES_FILE = os.path.join(SCRIPT_DIR, "sources.txt") def load_sources(): """从 sources.txt 文件加载下载资源""" sources = [] if os.path.exists(SOURCES_FILE): with open(SOURCES_FILE, 'r', encoding='utf-8') as f: for line in f: url = line.strip() # 跳过空行和注释行 if url and not url.startswith('#'): sources.append(url) return sources SOURCES = load_sources() class DownloadManager: """下载管理器""" def __init__(self): self.tasks = [] self.load_state() self.setup_logging() def setup_logging(self): """设置日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOG_FILE, encoding='utf-8'), logging.StreamHandler(sys.stdout) ] ) self.logger = logging.getLogger(__name__) def load_state(self): """加载状态""" # 每次启动都重新加载 sources.txt,检测是否有新增或删除的 URL current_sources = load_sources() if os.path.exists(STATE_FILE): try: with open(STATE_FILE, 'r', encoding='utf-8') as f: self.tasks = json.load(f) # 检查 sources 是否有变化,有变化则重新初始化 existing_urls = {task['url'] for task in self.tasks} if set(current_sources) != existing_urls: print("检测到 sources.txt 有变化,重新初始化任务列表") self.tasks = [{"url": url, "status": "pending", "filename": "", "size": 0, "error": "", "retry": 0} for url in current_sources] self.save_state() except Exception as e: print(f"加载状态失败: {e}") self.tasks = [{"url": url, "status": "pending", "filename": "", "size": 0, "error": "", "retry": 0} for url in current_sources] else: # 初始化任务列表 self.tasks = [{"url": url, "status": "pending", "filename": "", "size": 0, "error": "", "retry": 0} for url in current_sources] def save_state(self): """保存状态""" try: with open(STATE_FILE, 'w', encoding='utf-8') as f: json.dump(self.tasks, f, ensure_ascii=False, indent=2) except Exception as e: self.logger.error(f"保存状态失败: {e}") def get_filename_from_url(self, url, response): """从URL或响应头获取文件名""" from urllib.parse import unquote, urlparse # 优先从 URL 的 ?n= 参数获取文件名 if '?n=' in url: try: file_name = unquote(url.split('?n=')[1].split('&')[0]) if file_name: # 确保有 .zip 后缀(如果文件本身没有扩展名) if '.' not in os.path.basename(file_name): file_name = file_name + '.zip' return file_name except: pass # 尝试从Content-Disposition获取 content_disp = response.headers.get('Content-Disposition', '') if content_disp: import re match = re.search(r'filename[*]?=["\']?([^"\';\s]+)["\']?', content_disp) if match: return match.group(1) # 从URL路径提取 path = urlparse(url).path filename = os.path.basename(unquote(path)) if filename: return filename # 默认文件名 return f"file_{int(time.time())}" def download_file(self, task): """下载单个文件(带重试机制)""" url = task['url'] retry_count = task.get('retry', 0) try: self.logger.info(f"开始下载: {url}" + (f" (重试 {retry_count}/{MAX_RETRIES})" if retry_count > 0 else "")) # 创建下载目录 os.makedirs(DOWNLOAD_DIR, exist_ok=True) # 先获取文件名和预期大小 req = urllib.request.Request(url, method='HEAD') try: with urllib.request.urlopen(req) as response: filename = self.get_filename_from_url(url, response) except: # HEAD请求失败,使用GET请求获取文件名 with urllib.request.urlopen(url) as response: filename = self.get_filename_from_url(url, response) filepath = os.path.join(DOWNLOAD_DIR, filename) # 检查文件是否已存在且完整 if self.check_file_complete(filepath, url): task['status'] = 'completed' task['filename'] = filename task['size'] = os.path.getsize(filepath) task['error'] = '' self.logger.info(f"文件已存在且完整,跳过下载: {filename}") self.save_state() return True # 下载文件 task['status'] = 'downloading' task['filename'] = filename self.save_state() with urllib.request.urlopen(url) as response: # 获取预期文件大小 expected_size = response.headers.get('Content-Length') if expected_size: expected_size = int(expected_size) downloaded = 0 with open(filepath, 'wb') as f: while True: chunk = response.read(8192) if not chunk: break f.write(chunk) downloaded += len(chunk) # 验证文件完整性 if expected_size and downloaded != expected_size: os.remove(filepath) raise IOError(f"文件不完整: 预期 {expected_size} 字节,实际 {downloaded} 字节") task['size'] = downloaded task['status'] = 'completed' task['error'] = '' self.logger.info(f"下载完成: {filename} ({downloaded} bytes)") self.save_state() return True except Exception as e: retry_count = task.get('retry', 0) + 1 task['retry'] = retry_count error_msg = str(e) if retry_count <= MAX_RETRIES: task['status'] = 'retrying' task['error'] = f'{error_msg} (准备第 {retry_count} 次重试)' self.logger.warning(f"下载失败: {url} - {error_msg},准备第 {retry_count} 次重试") self.save_state() time.sleep(2) # 等待2秒后重试 return self.download_file(task) else: task['status'] = 'failed' task['error'] = f'{error_msg} (已重试 {MAX_RETRIES} 次)' self.logger.error(f"下载失败: {url} - {error_msg} (已重试 {MAX_RETRIES} 次)") self.save_state() return False def check_file_complete(self, filepath, url): """检查文件是否已存在且完整""" if not os.path.exists(filepath): return False # 获取已下载文件的大小 local_size = os.path.getsize(filepath) # 尝试获取远程文件大小 try: req = urllib.request.Request(url, method='HEAD') with urllib.request.urlopen(req, timeout=10) as response: remote_size = response.headers.get('Content-Length') if remote_size: remote_size = int(remote_size) # 文件存在且大小一致 if local_size == remote_size: return True return False except: # 无法获取远程文件大小,检查本地文件是否大于0 return local_size > 0 return local_size > 0 def run_downloads(self): """执行所有下载任务""" self.logger.info("=" * 50) self.logger.info("开始下载任务") for task in self.tasks: if task['status'] in ['pending', 'downloading', 'retrying']: self.download_file(task) self.logger.info("所有下载任务完成") self.logger.info("=" * 50) class WebHandler(http.server.SimpleHTTPRequestHandler): """Web请求处理器""" def __init__(self, *args, manager=None, **kwargs): self.manager = manager super().__init__(*args, **kwargs) def do_GET(self): """处理GET请求""" if self.path == '/' or self.path == '/status': self.send_html() elif self.path == '/api/tasks': self.send_json(self.manager.tasks) elif self.path.startswith('/api/log'): self.send_log() else: super().do_GET() def send_html(self): """发送HTML页面""" html = """
| # | URL | 状态 | 重试 | 文件名 | 大小 | 错误信息 |
|---|
加载中...