#!/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或响应头获取文件名""" # 尝试从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路径提取 from urllib.parse import unquote, urlparse 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 = """ 下载管理器

📥 文件下载管理器

自动刷新
3秒
下载目录: """ + os.path.basename(DOWNLOAD_DIR) + """
日志: """ + os.path.basename(LOG_FILE) + """
# URL 状态 重试 文件名 大小 错误信息

📋 最近日志

更新于: --:--:--
加载中...
""" self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(html.encode('utf-8')) def send_json(self, data): """发送JSON响应""" self.send_response(200) self.send_header('Content-Type', 'application/json; charset=utf-8') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) def send_log(self): """发送日志内容""" try: if os.path.exists(LOG_FILE): with open(LOG_FILE, 'r', encoding='utf-8') as f: lines = f.readlines() # 返回最后50行 log_content = ''.join(lines[-50:]) else: log_content = '日志文件不存在' except Exception as e: log_content = f'读取日志失败: {e}' self.send_response(200) self.send_header('Content-Type', 'text/plain; charset=utf-8') self.end_headers() self.wfile.write(log_content.encode('utf-8')) def log_message(self, format, *args): """禁用默认日志""" pass def save_pid(pid): """保存PID""" with open(PID_FILE, 'w') as f: f.write(str(pid)) def run_server(manager): """运行Web服务器""" def handler(*args, **kwargs): return WebHandler(*args, manager=manager, **kwargs) with socketserver.TCPServer(("", PORT), handler) as httpd: print(f"\n🌐 Web服务器启动: http://localhost:{PORT}") print(f"📁 下载目录: {os.path.abspath(DOWNLOAD_DIR)}") print(f"📋 日志文件: {os.path.abspath(LOG_FILE)}") print(f"📄 PID文件: {os.path.abspath(PID_FILE)}\n") httpd.serve_forever() def main(): """主函数""" # 检查是否已有进程运行 if os.path.exists(PID_FILE): try: with open(PID_FILE, 'r') as f: old_pid = int(f.read().strip()) # 检查进程是否存在 try: os.kill(old_pid, 0) print(f"⚠️ 进程已在运行 (PID: {old_pid})") print("如需重启,请先运行: ./stop.sh") return except OSError: # 进程不存在,删除旧的PID文件 os.remove(PID_FILE) except: pass # 保存当前PID save_pid(os.getpid()) # 创建管理器 manager = DownloadManager() # 在后台线程执行下载 import threading download_thread = threading.Thread(target=manager.run_downloads, daemon=True) download_thread.start() # 启动Web服务器 run_server(manager) if __name__ == "__main__": main()