""" 人教社电子教材 WebUI 管理与下载系统 (webui.py) 基于 FastAPI + TailwindCSS 构建，提供与原版网站一致的可视化筛选与一键批量下载体验。 """ import os import sys import time import json import asyncio import webbrowser import threading from typing import List, Dict, Optional from pydantic import BaseModel import uvicorn from fastapi import FastAPI, BackgroundTasks, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pep_core import PepCatalog, PepDownloader, XD_ORDER, XK_ORDER_PREFIX, NJ_ORDER, get_base_dir if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") app = FastAPI(title="人教社电子教材下载器") DOWNLOAD_DIR = os.path.join(get_base_dir(), "downloads") os.makedirs(DOWNLOAD_DIR, exist_ok=True) # 全局任务状态管理 class TaskManager: def __init__(self): self.queue: List[Dict] = [] self.is_running = False self.current_book: Optional[Dict] = None self.current_page = 0 self.total_pages = 0 self.status_text = "空闲中" self.logs: List[str] = [] self._lock = threading.Lock() def log(self, text: str): with self._lock: self.logs.append(text) if len(self.logs) > 500: self.logs.pop(0) def get_status(self) -> Dict: with self._lock: return { "is_running": self.is_running, "queue_len": len(self.queue), "current_book": self.current_book, "current_page": self.current_page, "total_pages": self.total_pages, "progress_percent": int((self.current_page / self.total_pages * 100)) if self.total_pages > 0 else 0, "status_text": self.status_text, "recent_logs": self.logs[-25:] } def add_tasks(self, books: List[Dict]): with self._lock: existing_ids = {b["id"] for b in self.queue} if self.current_book: existing_ids.add(self.current_book["id"]) for b in books: if b["id"] not in existing_ids: self.queue.append(b) existing_ids.add(b["id"]) self.log(f"[*] 已添加 {len(books)} 本教材至下载队列。") def run_worker(self): """后台单线程顺序执行下载队列中的教材""" downloader = PepDownloader(headless=True, output_dir=DOWNLOAD_DIR) while True: book_to_download = None with self._lock: if self.queue: book_to_download = self.queue.pop(0) self.current_book = book_to_download self.is_running = True self.current_page = 0 self.total_pages = 0 self.status_text = f"正在准备下载《{book_to_download.get('title')}》..." else: self.current_book = None self.is_running = False self.current_page = 0 self.total_pages = 0 self.status_text = "所有任务已完成" if not book_to_download: time.sleep(1) continue def progress_callback(cur, total, txt): with self._lock: self.current_page = cur self.total_pages = total self.status_text = txt def log_callback(txt): self.log(txt) try: self.log(f"==================================================") self.log(f"[*] 开始下载教材: 《{book_to_download.get('title')}》") downloader.download_book( book_id=book_to_download["id"], custom_title=book_to_download.get("title"), progress_cb=progress_callback, log_cb=log_callback ) except Exception as e: self.log(f"[-] 下载异常: {e}") time.sleep(1) task_manager = TaskManager() # 启动后台下载线程 worker_thread = threading.Thread(target=task_manager.run_worker, daemon=True) worker_thread.start() class FilterQuery(BaseModel): xd: Optional[str] = "全部" xk: Optional[str] = "全部" nj: Optional[str] = "全部" keyword: Optional[str] = "" class BatchDownloadRequest(BaseModel): book_ids: List[str] @app.get("/api/structure") def get_structure(): """获取所有学段、学科、年级的结构（保持严格定制排序）""" return PepCatalog.get_structure() @app.post("/api/books") def list_books(query: FilterQuery): """根据条件筛选教材列表""" books = PepCatalog.filter_books( xd=query.xd, xk=query.xk, nj=query.nj, keyword=query.keyword ) return {"total": len(books), "books": books} @app.get("/api/status") def get_download_status(): """轮询当前下载状态""" return task_manager.get_status() @app.post("/api/download") def add_download(req: BatchDownloadRequest): """提交下载请求""" all_books = {b["id"]: b for b in PepCatalog.fetch_and_decrypt_all()} selected = [all_books[bid] for bid in req.book_ids if bid in all_books] if selected: task_manager.add_tasks(selected) return {"status": "ok", "added_count": len(selected)} @app.post("/api/refresh_catalog") def refresh_catalog(): """强制重新从官方服务器拉取并解密最新教材目录""" try: books = PepCatalog.fetch_and_decrypt_all(force_refresh=True) return {"status": "ok", "total": len(books)} except Exception as e: return {"status": "error", "message": str(e)} @app.post("/api/clear_cache") def clear_cache(): """清理本地临时缓存图片""" try: res = PepCatalog.clear_cache_files() return res except Exception as e: return {"status": "error", "message": str(e)} @app.post("/api/open_folder") def open_download_folder(): """在资源管理器中打开下载文件夹""" try: if sys.platform == "win32": os.startfile(DOWNLOAD_DIR) elif sys.platform == "darwin": os.system(f'open "{DOWNLOAD_DIR}"') else: os.system(f'xdg-open "{DOWNLOAD_DIR}"') return {"status": "ok"} except Exception as e: return {"status": "error", "message": str(e)} @app.get("/", response_class=HTMLResponse) def index_page(): """返回一体化前端页面""" html_content = """ 人教社中小学电子教材下载系统
📚

人教社电子教材下载器

免滑块验证 · 批量抓取 · 自动合成高清 PDF

当前状态: 空闲中
队列等待: 0 本
就绪 0%
🔍
学段：
学科：
年级：
已找到 0 本教材 已选中 0 本
""" return html_content def main(): port = 8000 url = f"http://127.0.0.1:{port}" print("=" * 65) print(" 🚀 人教社电子教材 WebUI 服务器正在启动...") print(f" 🔗 请在浏览器打开: {url}") print("=" * 65) # 自动打开默认浏览器 threading.Timer(1.5, lambda: webbrowser.open(url)).start() uvicorn.run(app, host="127.0.0.1", port=port, log_level="info") if __name__ == "__main__": main()