From b6d812d0e6571f564fecec8980f4569603c12ce6 Mon Sep 17 00:00:00 2001 From: maohd <949894706@qq.com> Date: Tue, 8 Sep 2026 15:15:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(ui):=20=E4=BC=98=E5=8C=96=E8=B5=84?= =?UTF-8?q?=E6=BA=90=E5=88=97=E8=A1=A8=E6=94=AF=E6=8C=81=E5=A4=8D=E9=80=89?= =?UTF-8?q?=E6=A1=86=E5=8B=BE=E9=80=89=E4=B8=8E=E9=93=BE=E6=8E=A5=E5=90=8C?= =?UTF-8?q?=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在资源树中添加复选框,支持多选和半选状态表示分类勾选情况 - 支持按分类层级展开勾选,自动同步更新 URL 输入框中的资源链接 - 新增勾选数量显示标签,提升勾选反馈信息 - 优化封面图标与复选框图标合成显示,支持主题切换动态更新 - 支持通过点击或空格键切换树项勾选状态,增强键盘操作体验 - 实现分类与资源链接的增减同步,避免重复和遗漏 - 新增链接的批量解析功能后台处理,提升界面响应速度 - 资源下载支持分类子目录归类,自动创建缺失目录 - 为网络请求添加超时设置,避免请求长时间阻塞 - 修改剪贴板复制逻辑,增加复制内容校验和异常处理 - 改进下载进度显示,支持多任务进度汇总与失败提示 - 优化资源版别和分类目录提取,实现更细粒度的文件归类 --- src/tchmaterial_parser/api.py | 23 +- src/tchmaterial_parser/app.py | 2 +- src/tchmaterial_parser/network.py | 2 + src/tchmaterial_parser/ui/download_panel.py | 225 ++++++++++------ src/tchmaterial_parser/ui/resource_tree.py | 269 ++++++++++++++++---- src/tchmaterial_parser/ui/theme.py | 10 + tests/test_download.py | 2 +- tests/test_download_paths.py | 70 ++++- tests/test_download_progress.py | 106 ++++++++ tests/test_parse_collection.py | 51 ++++ tests/test_resource_check.py | 133 ++++++++++ tests/test_resource_urls.py | 100 ++++++++ 12 files changed, 857 insertions(+), 136 deletions(-) create mode 100644 tests/test_download_progress.py create mode 100644 tests/test_parse_collection.py create mode 100644 tests/test_resource_check.py create mode 100644 tests/test_resource_urls.py diff --git a/src/tchmaterial_parser/api.py b/src/tchmaterial_parser/api.py index 5b5d99a4..147c2e3c 100644 --- a/src/tchmaterial_parser/api.py +++ b/src/tchmaterial_parser/api.py @@ -5,7 +5,7 @@ from typing import NamedTuple from urllib.parse import urlparse, parse_qs -from .network import headers, request_headers, session +from .network import REQUEST_TIMEOUT, headers, request_headers, session from .platform_utils import print_error class ResourceInfo(NamedTuple): @@ -14,6 +14,7 @@ class ResourceInfo(NamedTuple): file_format: str chapters: list[dict] edition: str | None = None + relative_dir: tuple[str, ...] = () # 按资源树层级(学段/学科/版本)分类存放的子目录 def get_edition_name(resource_data: dict) -> str | None: """读取资源分类中的教材版别,例如 “人教版”“北师大版”。""" @@ -22,6 +23,17 @@ def get_edition_name(resource_data: dict) -> str | None: return tag["tag_name"] return None +TAG_DIMENSIONS = ("zxxxd", "zxxxk", "zxxbb") # 学段、学科、版本,与资源树的分类层级一致 + +def get_relative_dir(resource_data: dict) -> tuple[str, ...]: + """按资源树层级从 tag_list 提取分类目录(学段/学科/版本),供批量下载时分类存放。""" + names: dict[str, str] = {} + for tag in resource_data.get("tag_list") or []: + dimension = tag.get("tag_dimension_id") + if dimension in TAG_DIMENSIONS and tag.get("tag_name") and dimension not in names: + names[dimension] = tag["tag_name"] + return tuple(names[dimension] for dimension in TAG_DIMENSIONS if dimension in names) + def combine_resource_title(root_title: str | None, resource_title: str) -> str: """组合专题标题与实际资源标题,并避免平台重复标题造成超长文件名。""" if not root_title: @@ -128,6 +140,7 @@ def parse(url: str, bookmarks: bool) -> list[ResourceInfo] | None: # 解析资 data: dict = response.json() root_edition = get_edition_name(data) + root_dir = get_relative_dir(data) # 3. 获取资源标题、下载链接及章节目录 def get_resource_info(resource_data: dict, root_title: str | None = None, edition: str | None = None) -> ResourceInfo | None: @@ -192,7 +205,11 @@ def get_resource_info(resource_data: dict, root_title: str | None = None, editio if mapping_url: # a. 下载 mapping 文件获取页码和 ebook_id。 # mapping 也在 ndr-private 上,必须按 URL 现算 X-ND-AUTH,不能用全局占位头。 - map_resp = session.get(mapping_url, headers=request_headers(mapping_url)) + map_resp = session.get( + mapping_url, + headers=request_headers(mapping_url), + timeout=REQUEST_TIMEOUT, + ) map_data: dict = map_resp.json() ebook_id: str = map_data.get("ebook_id") @@ -251,6 +268,7 @@ def process_tree_nodes(nodes: list[dict]) -> list[dict]: resource_format, chapters, edition or get_edition_name(resource_data), + get_relative_dir(resource_data) or root_dir, ) def get_audio_info(audio_data: dict, root_title: str | None = None, edition: str | None = None) -> ResourceInfo | None: # 解析教材关联的音频资源(如英语教材听力) @@ -284,6 +302,7 @@ def get_audio_info(audio_data: dict, root_title: str | None = None, edition: str resource_format, [], edition or get_edition_name(audio_data), + get_relative_dir(audio_data) or root_dir, ) if content_type == "thematic_course": # 专题课程 diff --git a/src/tchmaterial_parser/app.py b/src/tchmaterial_parser/app.py index 82ed515d..c413cb2d 100644 --- a/src/tchmaterial_parser/app.py +++ b/src/tchmaterial_parser/app.py @@ -281,7 +281,7 @@ def on_description_resize(event: tk.Event) -> None: # 窗口宽度变化时, copy_btn.pack(side="right", padx=(0, scaled(8))) # 下载相关的控件全部就位后,写入下载面板模块,供其中的解析与下载流程使用 - download_panel.bind_widgets(url_text, bookmark_var, download_btn, download_progress_bar, progress_label) + download_panel.bind_widgets(url_text, bookmark_var, download_btn, copy_btn, download_progress_bar, progress_label) # 最后打包内容区,使其占据剩余的全部空间 paned.pack(side="top", fill="both", expand=True, pady=(scaled(14), 0)) diff --git a/src/tchmaterial_parser/network.py b/src/tchmaterial_parser/network.py index c2ec91c5..855de26e 100644 --- a/src/tchmaterial_parser/network.py +++ b/src/tchmaterial_parser/network.py @@ -11,6 +11,8 @@ session = requests.Session() # 初始化请求 session.trust_env = False # 不读取系统或环境变量中的代理配置 +REQUEST_TIMEOUT = (10, 60) # 连接 / 相邻两次收数据的超时秒数;requests 没有全局默认超时,缺失时卡住的请求会永久挂起 + headers = { # 设置请求头部,包含认证信息 "Authorization": "Bearer 0", "Origin": "https://basic.smartedu.cn", diff --git a/src/tchmaterial_parser/ui/download_panel.py b/src/tchmaterial_parser/ui/download_panel.py index ae668993..857a20b6 100644 --- a/src/tchmaterial_parser/ui/download_panel.py +++ b/src/tchmaterial_parser/ui/download_panel.py @@ -15,7 +15,7 @@ from .. import config from ..api import ResourceInfo, parse from ..bookmarks import add_bookmarks -from ..network import request_headers, session +from ..network import REQUEST_TIMEOUT, request_headers, session from ..platform_utils import print_error download_states: list[dict] = [] # 初始化下载状态 @@ -76,7 +76,12 @@ def request_download(url: str): while True: try: _pace_request() - response = session.get(candidate_url, headers=request_headers(candidate_url), stream=True) + response = session.get( + candidate_url, + headers=request_headers(candidate_url), + stream=True, + timeout=REQUEST_TIMEOUT, + ) except RequestException as e: last_exception = e break @@ -189,8 +194,10 @@ def allocate_download_paths(resources: list[ResourceInfo], directory: str) -> li reserved_paths: set[str] = set() allocated_paths: list[str] = [] - for filename in edition_filenames: - candidate = os.path.join(directory, filename) + for resource, filename in zip(resources, edition_filenames): + # 按资源的分类层级(学段/学科/版本)归入子目录,段名中的非法字符换为全角 + subdirectory = os.path.join(directory, *(sanitize_filename(part) for part in resource.relative_dir)) + candidate = os.path.join(subdirectory, filename) stem, extension = os.path.splitext(candidate) sequence = 2 @@ -207,104 +214,159 @@ def allocate_download_paths(resources: list[ResourceInfo], directory: str) -> li allocated_paths.append(candidate) return allocated_paths -def bind_widgets(text: tk.Text, bookmark: tk.BooleanVar, button: ttk.Button, progress_bar: ttk.Progressbar, label: ttk.Label) -> None: # 由 app.py 在创建控件后写入 - global url_text, bookmark_var, download_btn, download_progress_bar, progress_label - url_text, bookmark_var, download_btn, download_progress_bar, progress_label = text, bookmark, button, progress_bar, label +def bind_widgets(text: tk.Text, bookmark: tk.BooleanVar, button: ttk.Button, copy_button: ttk.Button, progress_bar: ttk.Progressbar, label: ttk.Label) -> None: # 由 app.py 在创建控件后写入 + global url_text, bookmark_var, download_btn, copy_btn, download_progress_bar, progress_label + url_text, bookmark_var, download_btn, copy_btn, download_progress_bar, progress_label = text, bookmark, button, copy_button, progress_bar, label -def parse_and_copy() -> None: # 解析并复制链接 - urls = {line.strip() for line in url_text.get("1.0", "end").splitlines() if line.strip()} # 获取所有非空行并去重 +def downloads_active() -> bool: # 是否存在尚未完成的下载任务 + return bool(download_states) and not all(state["finished"] for state in download_states) + +def show_parse_progress(current: int, total: int) -> None: # 后台解析大量链接时在进度标签上反馈进度;下载进行中则让位给下载进度 + if downloads_active(): + return + ui_call(progress_label.config, text=f"正在解析链接 {current}/{total}") + +def refresh_download_progress() -> None: # 汇总全部任务状态刷新进度条与标签,没有 Content-Length 或出现失败时也能看到进展 + states = list(download_states) # 下载线程会并发追加状态,先取快照避免遍历时被修改 + all_downloaded_size = sum(state["downloaded_size"] for state in states) + all_total_size = sum(state["total_size"] for state in states) + finished_number = len([state for state in states if state["finished"]]) + failed_number = len([state for state in states if state["failed_reason"]]) + total_number = len(states) + if all_total_size > 0: # 防止下面一行代码除以 0 而报错 + download_progress = (all_downloaded_size / all_total_size) * 100 + ui_call(download_progress_bar.config, value=download_progress) # 更新进度条 + progress_text = f"{format_bytes(all_downloaded_size)}/{format_bytes(all_total_size)} ({download_progress:.2f}%) 已下载 {finished_number}/{total_number}" + else: + progress_text = f"已下载 {format_bytes(all_downloaded_size)},已完成 {finished_number}/{total_number} 个文件" + if failed_number: + progress_text += f",{failed_number} 个失败" + ui_call(progress_label.config, text=progress_text) # 更新标签以显示当前下载进度 + +def collect_parsed_resources(parse_fn: callable, urls: list[str], bookmarks: bool, on_progress: callable | None = None) -> tuple[list[ResourceInfo], set[str]]: + """逐条解析链接并汇总结果:按资源直链去重,解析失败的链接单独收集。""" + resources_info_list: list[ResourceInfo] = [] resource_urls: set[str] = set() failed_urls: set[str] = set() - - for url in urls: - resources_info = parse(url, False) + for index, url in enumerate(urls): + if on_progress: + on_progress(index + 1, len(urls)) + resources_info = parse_fn(url, bookmarks) if not resources_info: - failed_urls.add(url) # 添加到失败链接 + failed_urls.add(url) continue for resource in resources_info: + if resource.url in resource_urls: # 直接使用 resources_info_list 会报错(list 不可哈希) + continue + resources_info_list.append(resource) resource_urls.add(resource.url) + return resources_info_list, failed_urls - if failed_urls: - messagebox.showwarning("警告", "以下 “行” 无法解析:\n" + "\n".join(failed_urls)) +def parse_urls_in_background(urls: list[str], bookmarks: bool, on_finished: callable) -> None: + """在后台线程逐条解析链接,完成后回到主线程执行 on_finished(资源列表, 失败链接集合)。 - if resource_urls: - try: - resource_urls_str = "\n".join(resource_urls) - url_text.clipboard_clear() - url_text.clipboard_append(resource_urls_str) # 将链接复制到剪贴板 - if url_text.clipboard_get() == resource_urls_str: # 检查剪贴板内容是否正确 - # 真实 X-ND-AUTH 必须按每条 URL 现算,不能把某一次的 nonce/mac 当作通用头复制出去。 - messagebox.showinfo( - "提示", - f'资源链接已复制到剪贴板。\n注意:链接可能无法直接下载。官网私有资源使用按地址单独计算的 X-ND-AUTH,请优先用本工具下载。{"若需手动请求,至少带上以下标头(含隐私信息,请勿分享):" if config.access_token else "未登录时可以尝试:"}\n\nAuthorization: Bearer {config.access_token or "0"}\nX-ND-AUTH: MAC id="{config.access_token or "0"}",nonce="0",mac="0"', - ) - else: + 批量选择的链接可能多达上百条,逐条解析需多次网络请求,放在主线程会让界面未响应。 + """ + def worker() -> None: + resources_info_list, failed_urls = collect_parsed_resources(parse, urls, bookmarks, show_parse_progress) + ui_call(on_finished, resources_info_list, failed_urls) + + thread_it(worker) + +def parse_and_copy() -> None: # 解析并复制链接 + urls = {line.strip() for line in url_text.get("1.0", "end").splitlines() if line.strip()} # 获取所有非空行并去重 + if not urls: + return + + copy_btn.config(state="disabled") # 解析期间禁用按钮,避免重复触发 + + def copy_urls(resources_info_list: list[ResourceInfo], failed_urls: set[str]) -> None: # 解析完成后在主线程复制链接 + copy_btn.config(state="normal") # 恢复按钮为启用状态 + if not downloads_active(): + progress_label.config(text="等待下载") # 解析进度已无用,恢复默认文案 + + resource_urls = {resource.url for resource in resources_info_list} + if failed_urls: + messagebox.showwarning("警告", "以下 “行” 无法解析:\n" + "\n".join(failed_urls)) + + if resource_urls: + try: + resource_urls_str = "\n".join(resource_urls) + url_text.clipboard_clear() + url_text.clipboard_append(resource_urls_str) # 将链接复制到剪贴板 + if url_text.clipboard_get() == resource_urls_str: # 检查剪贴板内容是否正确 + # 真实 X-ND-AUTH 必须按每条 URL 现算,不能把某一次的 nonce/mac 当作通用头复制出去。 + messagebox.showinfo( + "提示", + f'资源链接已复制到剪贴板。\n注意:链接可能无法直接下载。官网私有资源使用按地址单独计算的 X-ND-AUTH,请优先用本工具下载。{"若需手动请求,至少带上以下标头(含隐私信息,请勿分享):" if config.access_token else "未登录时可以尝试:"}\n\nAuthorization: Bearer {config.access_token or "0"}\nX-ND-AUTH: MAC id="{config.access_token or "0"}",nonce="0",mac="0"', + ) + else: + messagebox.showerror("错误", "无法将链接复制到剪贴板,请手动复制。") + except Exception as e: + print_error(e) messagebox.showerror("错误", "无法将链接复制到剪贴板,请手动复制。") - except Exception as e: - print_error(e) - messagebox.showerror("错误", "无法将链接复制到剪贴板,请手动复制。") + + parse_urls_in_background(list(urls), False, copy_urls) def download() -> None: # 下载资源文件 global download_states download_btn.config(state="disabled") # 设置下载按钮为禁用状态 + download_progress_bar.config(value=0) # 重置上一批任务可能残留的进度 download_states = [] # 初始化下载状态 urls = {line.strip() for line in url_text.get("1.0", "end").splitlines() if line.strip()} # 获取所有非空行并去重 - resources_info_list: list[ResourceInfo] = [] - resource_urls: set[str] = set() - failed_urls: set[str] = set() if config.access_token and not config.access_token.isascii(): # 判断 Access Token 中是否包含非 ASCII 字符 messagebox.showwarning("警告", "Access Token 不正确(包含非 ASCII 字符),请点击“设置 Token”按钮重新填写。") download_btn.config(state="normal") # 恢复下载按钮为启用状态 return - for url in urls: - resources_info = parse(url, bookmark_var.get()) - if not resources_info: - failed_urls.add(url) - continue - for resource in resources_info: - resource_url = resource.url - if resource_url in resource_urls: # 直接使用 resources_info_list 会报错(list 不可哈希) - continue - resources_info_list.append(resource) - resource_urls.add(resource_url) + if not urls: + download_btn.config(state="normal") # 恢复下载按钮为启用状态 + return - if len(resources_info_list) > 1: - messagebox.showinfo("提示", "您将下载多个文件,请选择要下载文件的位置,本程序将在选定的文件夹中使用资源名称作为文件名进行下载。") - dir_path = filedialog.askdirectory() # 选择文件夹 - if not dir_path: # 用户取消或关闭对话框 - download_btn.config(state="normal") # 恢复下载按钮为启用状态 + def start_downloads(resources_info_list: list[ResourceInfo], failed_urls: set[str]) -> None: # 解析完成后在主线程选择保存位置并开始下载 + def restore_download_btn() -> None: # 未产生下载任务时恢复界面状态 + if not downloads_active(): + progress_label.config(text="等待下载") + download_btn.config(state="normal") # 设置下载按钮为启用状态 + + if len(resources_info_list) > 1: + messagebox.showinfo("提示", "您将下载多个文件,请选择要下载文件的位置。本程序将在该文件夹中按教材分类创建子文件夹,并以资源名称命名文件。") + dir_path = filedialog.askdirectory() # 选择文件夹 + if not dir_path: # 用户取消或关闭对话框 + restore_download_btn() + return + dir_path = os.path.normpath(dir_path) + # 路径必须在任何线程启动前统一预留,否则同名资源仍可能同时打开同一个 .tmp 文件。 + download_targets = list(zip(resources_info_list, allocate_download_paths(resources_info_list, dir_path))) + elif resources_info_list: + download_targets: list[tuple[ResourceInfo, str]] = [] + for resource in resources_info_list: + save_path = filedialog.asksaveasfilename( # 选择保存路径 + defaultextension=f".{resource.file_format}", + filetypes=[(f"{resource.file_format.upper()} 文件", f"*.{resource.file_format}"), ("所有文件", "*.*")], + initialfile=sanitize_filename(resource.title or "download"), + ) + if not save_path: # 用户取消了文件保存操作 + restore_download_btn() + return + save_path = os.path.normpath(save_path) + download_targets.append((resource, save_path)) + else: # 没有可下载的资源 + restore_download_btn() + if failed_urls: + messagebox.showwarning("警告", "以下 “行” 无法解析:\n" + "\n".join(failed_urls)) # 显示警告对话框 return - dir_path = os.path.normpath(dir_path) - else: - dir_path = None - if dir_path: - # 路径必须在任何线程启动前统一预留,否则同名资源仍可能同时打开同一个 .tmp 文件。 - download_targets = list(zip(resources_info_list, allocate_download_paths(resources_info_list, dir_path))) - else: - download_targets: list[tuple[ResourceInfo, str]] = [] - for resource in resources_info_list: - save_path = filedialog.asksaveasfilename( # 选择保存路径 - defaultextension=f".{resource.file_format}", - filetypes=[(f"{resource.file_format.upper()} 文件", f"*.{resource.file_format}"), ("所有文件", "*.*")], - initialfile=sanitize_filename(resource.title or "download"), - ) - if not save_path: # 用户取消了文件保存操作 - download_btn.config(state="normal") # 恢复下载按钮为启用状态 - return - save_path = os.path.normpath(save_path) - download_targets.append((resource, save_path)) + for resource, save_path in download_targets: + thread_it(download_file, resource.url, save_path, resource.chapters) # 开始下载(多线程,防止窗口卡死) - for resource, save_path in download_targets: - thread_it(download_file, resource.url, save_path, resource.chapters) # 开始下载(多线程,防止窗口卡死) + progress_label.config(text=f"正在下载 {len(download_targets)} 个文件") - if failed_urls: - messagebox.showwarning("警告", "以下 “行” 无法解析:\n" + "\n".join(failed_urls)) # 显示警告对话框 + if failed_urls: + messagebox.showwarning("警告", "以下 “行” 无法解析:\n" + "\n".join(failed_urls)) # 显示警告对话框 - if not resources_info_list: - download_btn.config(state="normal") # 设置下载按钮为启用状态 + parse_urls_in_background(list(urls), bookmark_var.get(), start_downloads) def download_file(url: str, save_path: str, chapters: list[dict] | None = None) -> None: # 下载文件 current_state = { "download_url": url, "save_path": save_path, "downloaded_size": 0, "total_size": 0, "finished": False, "failed_reason": None } @@ -322,6 +384,7 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) else: current_state["total_size"] = int(response.headers.get("Content-Length", 0)) + os.makedirs(os.path.dirname(save_path), exist_ok=True) # 分类下载时子目录可能尚不存在 with open(temp_path, "wb") as file: for chunk in response.iter_content( # 分块下载 chunk_size=131072 if current_state["total_size"] < 20971520 else 262144 if current_state["total_size"] < 52428800 else 524288 @@ -329,15 +392,7 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) if chunk: # 过滤掉 Keep-Alive 块 file.write(chunk) current_state["downloaded_size"] += len(chunk) - all_downloaded_size = sum(state["downloaded_size"] for state in download_states) - all_total_size = sum(state["total_size"] for state in download_states) - downloaded_number = len([state for state in download_states if state["finished"]]) - total_number = len(download_states) - - if all_total_size > 0: # 防止下面一行代码除以 0 而报错 - download_progress = (all_downloaded_size / all_total_size) * 100 - ui_call(download_progress_bar.config, value=download_progress) # 更新进度条 - ui_call(progress_label.config, text=f"{format_bytes(all_downloaded_size)}/{format_bytes(all_total_size)} ({download_progress:.2f}%) 已下载 {downloaded_number}/{total_number}") # 更新标签以显示当前下载进度 + refresh_download_progress() if current_state["total_size"] > 0 and current_state["downloaded_size"] != current_state["total_size"]: # 文件下载不完整 current_state["failed_reason"] = f"文件下载不完整,需下载 {current_state['total_size']} 字节,实际下载 {current_state['downloaded_size']} 字节" @@ -368,6 +423,8 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) if response is not None: response.close() + refresh_download_progress() # 每个任务结束时刷新一次,重试等待期间也能看到完成数与失败数 + if all(state["finished"] for state in download_states): # 所有文件下载完成 ui_call(download_progress_bar.config, value=0) # 重置进度条 ui_call(progress_label.config, text="等待下载") # 清空进度标签 @@ -376,7 +433,7 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) failed_states = [state for state in download_states if state["failed_reason"]] if failed_states: # 存在下载失败的文件 failed_message = "\n\n".join( - f"{state['download_url']}\n{state['failed_reason']}" + f"{os.path.basename(state['save_path'])}\n{state['failed_reason']}" for state in failed_states ) ui_call( diff --git a/src/tchmaterial_parser/ui/resource_tree.py b/src/tchmaterial_parser/ui/resource_tree.py index 7f3356b4..ab5963dd 100644 --- a/src/tchmaterial_parser/ui/resource_tree.py +++ b/src/tchmaterial_parser/ui/resource_tree.py @@ -1,11 +1,12 @@ # -*- coding: utf-8 -*- -# 左侧资源列表:搜索筛选、封面按需加载与悬停预览 +# 左侧资源列表:勾选教材或分类、搜索筛选、封面按需加载与悬停预览 import io import tkinter as tk +from collections.abc import Iterator from tkinter import ttk import tkinter.font as tkfont -from PIL import Image, ImageOps, ImageTk +from PIL import Image, ImageDraw, ImageOps, ImageTk from . import runtime, theme from .runtime import scaled, thread_it, ui_call @@ -15,17 +16,84 @@ from ..network import session from ..platform_utils import os_name, print_error +def build_resource_url(item_path: str, resource_data: dict) -> str: # 根据树项路径与资源数据生成资源页面链接 + resource_type = resource_data.get("resource_type_code") or "assets_document" + content_id = resource_data.get("content_id") or item_path.split(":")[-1] + root_id = item_path.split(":")[0] + if resource_type == "teachingmaterials": + return f"https://basic.smartedu.cn/syncClassroom{'/prepare' if root_id == '__internal_prepare_lesson' else ''}?defaultTag={'%2F'.join(item_path.split(':')[1:])}" + return f"https://basic.smartedu.cn/tchMaterial/detail?contentType={resource_type}&contentId={content_id}&catalogType=tchMaterial&subCatalog=tchMaterial" + +def iter_leaf_resources(items: dict[str, dict], parent_path: str = "") -> Iterator[tuple[str, dict]]: # 遍历分类子树,产出每个末级资源的(树项路径, 资源数据) + for option_id, option_data in items.items(): + item_path = f"{parent_path}:{option_id}" if parent_path else option_id + children: dict[str, dict] = option_data.get("children", {}) + if children: # 分类节点继续向下遍历 + yield from iter_leaf_resources(children, item_path) + else: + yield item_path, option_data + +def collect_resource_urls(items: dict[str, dict], parent_path: str = "") -> list[str]: # 递归收集分类子树中所有末级资源的链接 + return [build_resource_url(item_path, resource_data) for item_path, resource_data in iter_leaf_resources(items, parent_path)] + +def find_tree_node(items: dict[str, dict], item_path: str) -> dict | None: # 按树项路径在分类树中定位节点数据 + node = None + branch = items + for segment in item_path.split(":"): + node = branch.get(segment) + if node is None: + return None + branch = node.get("children", {}) + return node + +def category_check_state(leaf_ids: list[str], checked_items: set[str]) -> str: # 依据子树内末级资源的勾选情况得出分类的三态 + if not leaf_ids: + return "unchecked" + checked_count = sum(1 for leaf_id in leaf_ids if leaf_id in checked_items) + if checked_count == 0: + return "unchecked" + return "checked" if checked_count == len(leaf_ids) else "partial" + +def should_check_category(leaf_ids: list[str], checked_items: set[str]) -> bool: # 点击分类时的目标状态:未全选时补全勾选,已全选时取消 + return category_check_state(leaf_ids, checked_items) != "checked" + +def draw_checkbox_image(size: int, state: str, colors: dict[str, str]) -> Image.Image: # 绘制跟随主题配色的三态复选框图标 + image = Image.new("RGBA", (size, size), (0, 0, 0, 0)) + draw = ImageDraw.Draw(image) + selected = state in ("checked", "partial") + border_width = max(2, size // 10) + draw.rounded_rectangle( + (border_width // 2, border_width // 2, size - 1 - border_width // 2, size - 1 - border_width // 2), + radius=max(2, size // 5), + fill=colors["selbg"] if state == "checked" else colors["surface"], + outline=colors["selbg"] if selected else colors["muted"], + width=border_width, + ) + if state == "checked": # 对勾 + draw.line( + (size * 0.24, size * 0.53, size * 0.44, size * 0.74, size * 0.78, size * 0.3), + fill=colors["selfg"], + width=max(2, size // 7), + joint="curve", + ) + elif state == "partial": # 半选横线 + inset = size * 0.32 + draw.line((inset, size / 2, size - inset, size / 2), fill=colors["selbg"], width=max(2, size // 10)) + return image + def build_resource_tree(pane: ttk.Frame, resource_list: dict[str, dict], url_text: tk.Text) -> None: # 在给定的子框架内构建资源列表 pane.columnconfigure(0, weight=1) pane.rowconfigure(2, weight=1) treeview_header = ttk.Frame(pane) treeview_header.grid(row=0, column=0, columnspan=2, sticky="ew", pady=(0, scaled(6))) - treeview_header.columnconfigure(0, weight=1) + treeview_header.columnconfigure(1, weight=1) treeview_label = ttk.Label(treeview_header, text="资源列表", style="Heading.TLabel") # 添加树视图标签 treeview_label.grid(row=0, column=0, sticky="w") + checked_count_label = ttk.Label(treeview_header, style="Caption.TLabel") # 显示当前勾选的教材数量 + checked_count_label.grid(row=0, column=1, sticky="e", padx=(0, scaled(8))) search_status_label = ttk.Label(treeview_header, style="Caption.TLabel") - search_status_label.grid(row=0, column=1, sticky="e") + search_status_label.grid(row=0, column=2, sticky="e") search_frame = ttk.Frame(pane) search_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(0, scaled(8))) @@ -40,7 +108,7 @@ def build_resource_tree(pane: ttk.Frame, resource_list: dict[str, dict], url_tex clear_search_btn = ttk.Button(search_frame, text="清除", width=5, command=lambda: search_var.set("")) clear_search_btn.grid(row=0, column=2, padx=(scaled(6), 0)) - treeview = ttk.Treeview(pane, style="Custom.Treeview", show="tree", selectmode="browse", height=12) # 创建树视图,使用自定义样式(该样式在 apply_theme() 中配置),隐藏列标题,设置选择模式为单选 + treeview = ttk.Treeview(pane, style="Custom.Treeview", show="tree", selectmode="extended", height=12) # 创建树视图,使用自定义样式(该样式在 apply_theme() 中配置),隐藏列标题;勾选状态用复选框图标表达,选择模式仅供键盘导航 treeview.column("#0", stretch=False) treeview.grid(row=2, column=0, sticky="nsew") treeview_scrollbar = ttk.Scrollbar(pane, orient="vertical", command=treeview.yview) @@ -51,12 +119,18 @@ def build_resource_tree(pane: ttk.Frame, resource_list: dict[str, dict], url_tex tree_item_data: dict[str, dict] = {} # 键为树项 ID,值为资源数据 tree_item_paths: dict[str, tuple[str, ...]] = {} # 保存完整分类路径,用于悬停提示 - tree_item_images: dict[str, ImageTk.PhotoImage] = {} # 缓存已加载的封面,筛选后继续复用 + tree_item_images: dict[str, ImageTk.PhotoImage] = {} # 持有树项图标(复选框与封面的合成图)的引用防止被回收,筛选后继续复用 + tree_cover_images: dict[str, Image.Image] = {} # 已加载封面的缩放图,勾选状态变化时与复选框重新合成 tree_preview_images: dict[str, ImageTk.PhotoImage] = {} # 缓存大尺寸封面,用于悬停预览 loading_tree_images: set[str] = set() + checked_items: set[str] = set() # 已勾选末级资源的树项路径,搜索重建树视图后仍保留 + checkbox_pils: dict[str, Image.Image] = {} # 三态复选框底图,跟随主题配色重建 + checkbox_icons: dict[str, ImageTk.PhotoImage] = {} # 无封面树项直接使用的复选框图标(已含右侧间距) tree_font = tkfont.nametofont("AppBodyFont") tree_cover_size = (scaled(26), scaled(28)) tree_cover_gap = scaled(8) # 用透明区域拉开封面与标题,避免改变封面尺寸 + checkbox_size = scaled(18) + checkbox_gap = scaled(6) # 复选框与封面、标题之间的间距 preview_cover_size = (scaled(80), scaled(112)) tree_content_width = 0 @@ -76,7 +150,7 @@ def build_tree_items(parent: str, items: dict[str, dict], parent_names: tuple[st "end", iid=item_id, text=display_name, - image=tree_item_images.get(item_id, ""), + image=compose_item_image(item_id), open=expand_all or not parent, ) children: dict[str, dict] = option_data.get("children", {}) @@ -84,24 +158,67 @@ def build_tree_items(parent: str, items: dict[str, dict], parent_names: tuple[st build_tree_items(item_id, children, path_names, expand_all) depth_width = len(path_names) * scaled(20) - image_width = tree_cover_size[0] + get_tree_cover_gap(display_name) + scaled(4) if option_data.get("custom_properties", {}).get("thumbnails") else 0 + if option_data.get("custom_properties", {}).get("thumbnails"): + image_width = checkbox_size + checkbox_gap + tree_cover_size[0] + get_tree_cover_gap(display_name) + scaled(4) + else: + image_width = checkbox_size + checkbox_gap tree_content_width = max(tree_content_width, depth_width + image_width + tree_font.measure(display_name) + scaled(20)) def resize_tree_column(width: int) -> None: # 让树列至少铺满可视区域,内容过长时启用横向滚动 treeview.column("#0", width=max(tree_content_width, width - scaled(2))) + def rebuild_checkbox_images() -> None: # 按当前主题配色生成三态复选框图标 + checkbox_pils.clear() + checkbox_icons.clear() + for state in ("checked", "partial", "unchecked"): + checkbox_pils[state] = draw_checkbox_image(checkbox_size, state, theme.current_colors) + checkbox_icons[state] = ImageTk.PhotoImage(ImageOps.expand(checkbox_pils[state], border=(0, 0, checkbox_gap, 0), fill=(0, 0, 0, 0))) + + def on_theme_changed() -> None: # 主题切换后重建复选框配色并刷新全部树项图标 + rebuild_checkbox_images() + for item_id in list(tree_item_data): + refresh_item_image(item_id) + + def item_check_state(item_id: str) -> str: # 末级资源为勾选/未勾选两态,分类按后代整体勾选情况显示三态 + node = find_tree_node(resource_list, item_id) + if node is None: + return "unchecked" + children = node.get("children") + if children: + leaf_ids = [leaf_id for leaf_id, _leaf_data in iter_leaf_resources(children, item_id)] + return category_check_state(leaf_ids, checked_items) + return "checked" if item_id in checked_items else "unchecked" + + def compose_item_image(item_id: str) -> ImageTk.PhotoImage: # 合成树项图标:勾选状态对应的复选框与已加载的封面 + state = item_check_state(item_id) + cover = tree_cover_images.get(item_id) + if cover is None: # 尚未加载封面的树项直接复用带间距的复选框图标 + return checkbox_icons[state] + checkbox = checkbox_pils[state] + height = max(checkbox.height, cover.height) + image = Image.new("RGBA", (checkbox.width + checkbox_gap + cover.width, height), (0, 0, 0, 0)) + image.paste(checkbox, (0, (height - checkbox.height) // 2), checkbox) + image.paste(cover, (checkbox.width + checkbox_gap, (height - cover.height) // 2), cover) + return ImageTk.PhotoImage(image) + + def refresh_item_image(item_id: str) -> None: # 重新合成并应用树项图标(勾选状态或封面变化后调用) + image = compose_item_image(item_id) + tree_item_images[item_id] = image + if treeview.exists(item_id): + treeview.item(item_id, image=image) + def apply_tree_icon(item_id: str, image: Image.Image | None) -> None: loading_tree_images.discard(item_id) - if image is None: - return - tree_image = fit_cover_image(image, tree_cover_size) - cover_gap = get_tree_cover_gap(tree_item_data[item_id]["display_name"]) - if cover_gap: - tree_image = ImageOps.expand(tree_image, border=(0, 0, cover_gap, 0), fill=(0, 0, 0, 0)) - tree_item_images[item_id] = ImageTk.PhotoImage(tree_image) - tree_preview_images[item_id] = ImageTk.PhotoImage(image) - if treeview.exists(item_id): - treeview.item(item_id, image=tree_item_images[item_id]) + if image is not None: + tree_image = fit_cover_image(image, tree_cover_size) + # 搜索重建后树项可能暂不在当前视图中,仍缓存封面供下次合成复用 + resource_data = tree_item_data.get(item_id) or find_tree_node(resource_list, item_id) or {} + cover_gap = get_tree_cover_gap(resource_data.get("display_name", "")) + if cover_gap: + tree_image = ImageOps.expand(tree_image, border=(0, 0, cover_gap, 0), fill=(0, 0, 0, 0)) + tree_cover_images[item_id] = tree_image + tree_preview_images[item_id] = ImageTk.PhotoImage(image) + refresh_item_image(item_id) def load_tree_icon(item_id: str, url: str) -> None: # 在线程中下载封面,在主线程中更新控件 try: @@ -118,7 +235,7 @@ def load_tree_icon(item_id: str, url: str) -> None: # 在线程中下载封面 def queue_tree_icon(item_id: str) -> None: resource_data = tree_item_data[item_id] thumbnails = resource_data.get("custom_properties", {}).get("thumbnails") - if thumbnails and item_id not in tree_item_images and item_id not in loading_tree_images: + if thumbnails and item_id not in tree_cover_images and item_id not in loading_tree_images: loading_tree_images.add(item_id) thread_it(load_tree_icon, item_id, thumbnails[0]) @@ -149,37 +266,92 @@ def refresh_resource_tree() -> None: # 根据搜索词重建树视图 clear_search_btn.state(["!disabled"] if query else ["disabled"]) ui_call(load_visible_tree_icons) - def on_tree_select(event: tk.Event) -> None: # 处理树视图选择事件 - selection = treeview.selection() - if not selection: + def insert_resource_urls(urls: list[str]) -> None: # 将链接追加到 URL 输入框,跳过已存在的行 + existing_lines = set(url_text.get("1.0", "end").splitlines()) + new_urls = [url for url in dict.fromkeys(urls) if url and url not in existing_lines] # 保序去重,并跳过已存在的链接 + if not new_urls: return - item = selection[0] - children = treeview.get_children(item) - if children: # 如果选中的项有子项,则加载子项的预览图,否则插入 URL - for child in children: # 遍历子项,设置子项的图片 - queue_tree_icon(child) - else: - resource_data = tree_item_data.get(item) - if not resource_data: - return + url_text_content = url_text.get("1.0", "end")[:-1] # 获取 URL 输入框的内容,去掉最后一个换行符 + # URL 输入框为空或最后一个字符为换行符时,插入的内容前面不加换行 + prefix = "" if not url_text_content or url_text_content[-1] == "\n" else "\n" + url_text.insert("end", prefix + "\n".join(new_urls)) + url_text.see("end") # 滚动到文本框底部 - resource_type = resource_data.get("resource_type_code") or "assets_document" - content_id = resource_data.get("content_id") or item.split(":")[-1] - root_id = item.split(":")[0] - if resource_type == "teachingmaterials": - url = f"https://basic.smartedu.cn/syncClassroom{'/prepare' if root_id == '__internal_prepare_lesson' else ''}?defaultTag={'%2F'.join(item.split(':')[1:])}" - else: - url = f"https://basic.smartedu.cn/tchMaterial/detail?contentType={resource_type}&contentId={content_id}&catalogType=tchMaterial&subCatalog=tchMaterial" + def remove_resource_urls(urls: list[str]) -> None: # 从 URL 输入框移除已取消勾选资源的链接行 + if not urls: + return + url_set = set(urls) + lines = url_text.get("1.0", "end").splitlines() + kept_lines = [line for line in lines if line not in url_set] + if len(kept_lines) == len(lines): + return + url_text.delete("1.0", "end") + if kept_lines: + url_text.insert("1.0", "\n".join(kept_lines)) - url_text_content = url_text.get("1.0", "end")[:-1] # 获取 URL 输入框的内容,去掉最后一个换行符 - if url in url_text_content.splitlines(): # 如果 URL 已经存在于输入框中,则不再插入 - return - if not url_text_content or url_text_content[-1] == "\n": # URL 输入框为空或最后一个字符为换行符时,插入的内容前面不加换行 - url_text.insert("end", url) + def update_checked_count() -> None: # 更新已勾选教材数量提示 + checked_count_label.config(text=f"已选 {len(checked_items)} 项" if checked_items else "") + + def toggle_item(item_id: str) -> None: # 切换树项勾选状态:分类按三态决定目标状态并级联其下所有末级资源 + node = find_tree_node(resource_list, item_id) + if node is None: + return + children = node.get("children") + if children: + leafs = list(iter_leaf_resources(children, item_id)) + checked = should_check_category([leaf_id for leaf_id, _leaf_data in leafs], checked_items) + else: + leafs = [(item_id, node)] + checked = item_id not in checked_items + set_items_checked(leafs, checked) + + def set_items_checked(leafs: list[tuple[str, dict]], checked: bool) -> None: # 批量更新末级资源勾选状态,级联刷新图标并同步 URL 输入框 + changed_leafs: list[tuple[str, dict]] = [] + for leaf_id, leaf_data in leafs: + if checked == (leaf_id in checked_items): + continue + if checked: + checked_items.add(leaf_id) else: - url_text.insert("end", f"\n{url}") - url_text.see("end") # 滚动到文本框底部 + checked_items.discard(leaf_id) + changed_leafs.append((leaf_id, leaf_data)) + if not changed_leafs: + return + + refresh_ids: set[str] = set() # 状态变化的末级资源及其各级祖先分类都需要刷新图标 + for leaf_id, _leaf_data in changed_leafs: + segments = leaf_id.split(":") + refresh_ids.update(":".join(segments[:index]) for index in range(1, len(segments) + 1)) + for refresh_id in refresh_ids: + if refresh_id in tree_item_data: + refresh_item_image(refresh_id) + + urls = [build_resource_url(leaf_id, leaf_data) for leaf_id, leaf_data in changed_leafs] + if checked: + insert_resource_urls(urls) + else: + remove_resource_urls(urls) + update_checked_count() + + def on_tree_press(event: tk.Event) -> str | None: # 按下鼠标时隐藏悬停提示;左键点击标题或封面(含复选框)时切换勾选,点击箭头或缩进保持展开收起 + hide_tree_tooltip() + if event.num != 1 or treeview.identify("element", event.x, event.y) not in ("text", "image"): + return None + item_id = treeview.identify_row(event.y) + if not item_id: + return None + treeview.focus_set() # 确保随后可以直接用方向键与空格操作 + treeview.selection_set(item_id) + treeview.focus(item_id) + toggle_item(item_id) + return "break" + + def on_tree_space(_event: tk.Event) -> str: # 空格键切换当前焦点树项的勾选状态 + item_id = treeview.focus() + if item_id: + toggle_item(item_id) + return "break" tooltip_window: tk.Toplevel | None = None tooltip_after_id: str | None = None @@ -294,15 +466,18 @@ def on_tree_shift_mousewheel(event: tk.Event) -> str: delta_unit = 1 if os_name == "Darwin" else 120 return scroll_tree_horizontally(-event.delta / delta_unit) + rebuild_checkbox_images() # 构建树项前先生成三态复选框图标 + theme.on_theme_applied(on_theme_changed) # 主题切换后重建复选框配色并刷新树项图标 + update_checked_count() refresh_resource_tree() # 初始展示完整资源树并展开一级目录 search_var.trace_add("write", schedule_search) treeview.configure(yscrollcommand=on_tree_view_change) - treeview.bind("<>", on_tree_select) + treeview.bind("", on_tree_space) treeview.bind("<>", lambda _event: ui_call(load_visible_tree_icons)) treeview.bind("", lambda event: resize_tree_column(event.width)) treeview.bind("", on_tree_motion) treeview.bind("", lambda _event: leave_tree()) - treeview.bind("", lambda _event: hide_tree_tooltip()) + treeview.bind("", on_tree_press) treeview.bind("", on_tree_shift_mousewheel) treeview.bind("", lambda _event: scroll_tree_horizontally(-1)) treeview.bind("", lambda _event: scroll_tree_horizontally(1)) diff --git a/src/tchmaterial_parser/ui/theme.py b/src/tchmaterial_parser/ui/theme.py index 5fd9af68..2cb57bd4 100644 --- a/src/tchmaterial_parser/ui/theme.py +++ b/src/tchmaterial_parser/ui/theme.py @@ -16,6 +16,10 @@ current_theme = "light" # 当前主题,若 switched_theme 为 `system` 则 current_theme 为系统主题(`light` 或 `dark`) current_colors: dict[str, str] = {} # 当前主题的配色,在 apply_theme() 中填充 themed_widgets: set[tk.Widget] = set() # 当前仍存在且需要跟随主题调整配色的 tk 原生控件 +theme_actions: list[callable] = [] # 主题应用后需要额外执行的回调(如重建跟随主题的图片资源) + +def on_theme_applied(action: callable) -> None: # 登记回调,在每次 apply_theme() 末尾执行 + theme_actions.append(action) # 主题配色,surface 为 sv-ttk 卡片贴图的填充色,须与之一致,否则卡片内会出现色差; # page 比 surface 略深,用作页面底色,让卡片、列表、文本框显出层次 @@ -164,4 +168,10 @@ def apply_theme(theme: Literal["system", "light", "dark"]) -> None: # 应用浅 for widget in themed_widgets: apply_widget_theme(widget) + for action in theme_actions: # 重建跟随主题的图片等资源 + try: + action() + except Exception as e: + print_error(e) + apply_titlebar_theme(runtime.root) diff --git a/tests/test_download.py b/tests/test_download.py index e08669f9..dc0109d0 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -56,7 +56,7 @@ def setUp(self) -> None: self.addCleanup(setattr, download_panel, "_MIN_REQUEST_INTERVAL", previous_interval) download_panel._MIN_REQUEST_INTERVAL = 0 widget = FakeWidget() - download_panel.bind_widgets(widget, widget, widget, widget, widget) + download_panel.bind_widgets(widget, widget, widget, widget, widget, widget) def failure_reason(self, status_code: int) -> str: download_panel.session = FakeSession(status_code) diff --git a/tests/test_download_paths.py b/tests/test_download_paths.py index 639d6485..d59d2c50 100644 --- a/tests/test_download_paths.py +++ b/tests/test_download_paths.py @@ -2,7 +2,7 @@ import tempfile import unittest -from src.tchmaterial_parser.api import ResourceInfo +from src.tchmaterial_parser.api import ResourceInfo, get_relative_dir from src.tchmaterial_parser.ui.download_panel import ( allocate_download_paths, download_filename, @@ -15,6 +15,7 @@ def resource( resource_key: str, edition: str | None = None, file_format: str = "pdf", + relative_dir: tuple[str, ...] = (), ) -> ResourceInfo: return ResourceInfo( title=title, @@ -22,6 +23,7 @@ def resource( file_format=file_format, chapters=[], edition=edition, + relative_dir=relative_dir, ) @@ -80,6 +82,41 @@ def test_avoids_existing_final_and_temporary_files(self) -> None: self.assertEqual([os.path.basename(path) for path in paths], ["已有教材 (2).pdf", "未完成教材 (2).pdf"]) + def test_places_files_into_category_subdirectories(self) -> None: + resources = [ + resource("必修上册", "book-1", relative_dir=("高中", "语文", "统编版")), + resource("必修上册", "book-2", relative_dir=("高中", "语文", "人教版")), + ] + + with tempfile.TemporaryDirectory() as directory: + paths = allocate_download_paths(resources, directory) + + self.assertEqual(paths, [ + os.path.join(directory, "高中", "语文", "统编版", "必修上册.pdf"), + os.path.join(directory, "高中", "语文", "人教版", "必修上册.pdf"), + ]) + + def test_same_title_in_different_categories_needs_no_sequence(self) -> None: + resources = [ + resource("必修上册", "book-1", relative_dir=("高中", "语文")), + resource("必修上册", "book-2", relative_dir=("初中", "语文")), + ] + + with tempfile.TemporaryDirectory() as directory: + paths = allocate_download_paths(resources, directory) + + self.assertEqual([os.path.basename(path) for path in paths], ["必修上册.pdf", "必修上册.pdf"]) + + def test_sanitizes_category_directory_names(self) -> None: + resources = [ + resource("教材", "book-1", relative_dir=("特殊:学段", "语*文")), + ] + + with tempfile.TemporaryDirectory() as directory: + paths = allocate_download_paths(resources, directory) + + self.assertEqual(paths, [os.path.join(directory, "特殊:学段", "语*文", "教材.pdf")]) + def test_issue_86_replaces_windows_illegal_filename_characters(self) -> None: resources = [ resource( @@ -136,6 +173,37 @@ def test_sanitized_titles_that_collide_still_get_a_sequence(self) -> None: ]) +class GetRelativeDirTest(unittest.TestCase): + @staticmethod + def tag(dimension: str, name: str, order: int = 0) -> dict: + return {"tag_dimension_id": dimension, "tag_name": name, "order_num": order} + + def test_builds_dir_in_stage_subject_edition_order(self) -> None: + data = {"tag_list": [ + self.tag("zxxnj", "一年级"), + self.tag("zxxbb", "统编版"), + self.tag("zxxxd", "小学"), + self.tag("zxxxk", "道德与法治"), + self.tag("tagView", "教材"), + ]} + + self.assertEqual(get_relative_dir(data), ("小学", "道德与法治", "统编版")) + + def test_skips_missing_dimensions(self) -> None: + data = {"tag_list": [self.tag("zxxxd", "高中"), self.tag("zxxxk", "语文")]} + + self.assertEqual(get_relative_dir(data), ("高中", "语文")) + + def test_returns_empty_dir_without_tag_list(self) -> None: + self.assertEqual(get_relative_dir({}), ()) + self.assertEqual(get_relative_dir({"tag_list": []}), ()) + + def test_ignores_other_dimensions_and_blank_names(self) -> None: + data = {"tag_list": [self.tag("zxxcc", "上册"), self.tag("zxxbb", ""), self.tag("5036342742", "电子教材")]} + + self.assertEqual(get_relative_dir(data), ()) + + class SanitizeFilenameTest(unittest.TestCase): def test_uses_fullwidth_punctuation_and_strips_trailing_dots(self) -> None: self.assertEqual(sanitize_filename('a<>:"/\\|?*b'), "a<>:"/\|?*b") diff --git a/tests/test_download_progress.py b/tests/test_download_progress.py new file mode 100644 index 00000000..d77fe901 --- /dev/null +++ b/tests/test_download_progress.py @@ -0,0 +1,106 @@ +import unittest + +from src.tchmaterial_parser.network import REQUEST_TIMEOUT +from src.tchmaterial_parser.ui import download_panel + + +class RecordingWidget: + def __init__(self) -> None: + self.configs: list[dict] = [] + + def config(self, **kwargs: dict) -> None: + self.configs.append(kwargs) + + +class FakeResponse: + def __init__(self, status_code: int) -> None: + self.ok = status_code < 400 + self.status_code = status_code + + def close(self) -> None: + pass + + +class TimeoutRecordingSession: + def __init__(self) -> None: + self.timeouts: list[tuple | None] = [] + + def get(self, url: str, **kwargs: dict) -> FakeResponse: + self.timeouts.append(kwargs.get("timeout")) + return FakeResponse(200) + + +class DownloadProgressTest(unittest.TestCase): + def setUp(self) -> None: + # 让 ui_call 直接同步执行,便于断言标签与进度条的实际更新内容 + previous_call = download_panel.ui_call + self.addCleanup(setattr, download_panel, "ui_call", previous_call) + download_panel.ui_call = lambda func, *args, **kwargs: func(*args, **kwargs) + + previous_states = download_panel.download_states + self.addCleanup(setattr, download_panel, "download_states", previous_states) + download_panel.download_states = [] + + self.label = RecordingWidget() + self.bar = RecordingWidget() + previous_label = download_panel.progress_label + previous_bar = download_panel.download_progress_bar + self.addCleanup(setattr, download_panel, "progress_label", previous_label) + self.addCleanup(setattr, download_panel, "download_progress_bar", previous_bar) + download_panel.progress_label = self.label + download_panel.download_progress_bar = self.bar + + def latest_label_text(self) -> str: + return self.label.configs[-1]["text"] + + def test_shows_percentage_when_total_size_known(self) -> None: + download_panel.download_states = [ + {"downloaded_size": 50, "total_size": 100, "finished": True, "failed_reason": None}, + {"downloaded_size": 50, "total_size": 100, "finished": False, "failed_reason": None}, + ] + + download_panel.refresh_download_progress() + + self.assertIn("50.00%", self.latest_label_text()) + self.assertIn("已下载 1/2", self.latest_label_text()) + self.assertEqual(self.bar.configs[-1], {"value": 50.0}) + + def test_shows_finished_count_without_total_size(self) -> None: + download_panel.download_states = [ + {"downloaded_size": 0, "total_size": 0, "finished": True, "failed_reason": None}, + {"downloaded_size": 0, "total_size": 0, "finished": False, "failed_reason": None}, + ] + + download_panel.refresh_download_progress() + + self.assertIn("已完成 1/2 个文件", self.latest_label_text()) + self.assertEqual(self.bar.configs, []) # 未知总大小时不驱动进度条 + + def test_counts_failed_downloads(self) -> None: + download_panel.download_states = [ + {"downloaded_size": 0, "total_size": 0, "finished": True, "failed_reason": "服务器返回 HTTP 状态码 403"}, + {"downloaded_size": 0, "total_size": 0, "finished": False, "failed_reason": None}, + ] + + download_panel.refresh_download_progress() + + self.assertIn("1 个失败", self.latest_label_text()) + + def test_private_download_requests_carry_timeout(self) -> None: + fake_session = TimeoutRecordingSession() + previous_session = download_panel.session + self.addCleanup(setattr, download_panel, "session", previous_session) + download_panel.session = fake_session + previous_interval = download_panel._MIN_REQUEST_INTERVAL + self.addCleanup(setattr, download_panel, "_MIN_REQUEST_INTERVAL", previous_interval) + download_panel._MIN_REQUEST_INTERVAL = 0 + + response, _attempted_urls = download_panel.request_download("https://r1-ndr-private.ykt.cbern.com.cn/edu_product/esp/assets/test.pkg/book.pdf") + + self.assertTrue(response.ok) + self.assertTrue(fake_session.timeouts) + self.assertTrue(all(timeout == REQUEST_TIMEOUT for timeout in fake_session.timeouts)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_parse_collection.py b/tests/test_parse_collection.py new file mode 100644 index 00000000..b54d67db --- /dev/null +++ b/tests/test_parse_collection.py @@ -0,0 +1,51 @@ +import unittest + +from src.tchmaterial_parser.api import ResourceInfo +from src.tchmaterial_parser.ui.download_panel import collect_parsed_resources + + +def make_resource(url: str, title: str = "资源") -> ResourceInfo: + return ResourceInfo(title, url, "pdf", []) + + +class CollectParsedResourcesTest(unittest.TestCase): + def test_flattens_multiple_resources_per_url(self) -> None: + def fake_parse(url: str, bookmarks: bool): + return [make_resource("https://example.com/a.pdf"), make_resource("https://example.com/b.mp3")] + + resources, failed = collect_parsed_resources(fake_parse, ["https://example.com/1"], False) + + self.assertEqual([resource.url for resource in resources], ["https://example.com/a.pdf", "https://example.com/b.mp3"]) + self.assertEqual(failed, set()) + + def test_deduplicates_by_resource_url_across_urls(self) -> None: + def fake_parse(url: str, bookmarks: bool): + return [make_resource("https://example.com/a.pdf")] + + resources, failed = collect_parsed_resources(fake_parse, ["https://example.com/1", "https://example.com/2"], False) + + self.assertEqual([resource.url for resource in resources], ["https://example.com/a.pdf"]) + self.assertEqual(failed, set()) + + def test_collects_urls_that_fail_to_parse(self) -> None: + def fake_parse(url: str, bookmarks: bool): + return None if url.endswith("bad") else [make_resource("https://example.com/a.pdf")] + + resources, failed = collect_parsed_resources(fake_parse, ["https://example.com/good", "https://example.com/bad"], False) + + self.assertEqual([resource.url for resource in resources], ["https://example.com/a.pdf"]) + self.assertEqual(failed, {"https://example.com/bad"}) + + def test_reports_progress_for_each_url(self) -> None: + def fake_parse(url: str, bookmarks: bool): + return None + + progress: list[tuple[int, int]] = [] + + collect_parsed_resources(fake_parse, ["https://example.com/1", "https://example.com/2", "https://example.com/3"], True, lambda current, total: progress.append((current, total))) + + self.assertEqual(progress, [(1, 3), (2, 3), (3, 3)]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resource_check.py b/tests/test_resource_check.py new file mode 100644 index 00000000..176b4001 --- /dev/null +++ b/tests/test_resource_check.py @@ -0,0 +1,133 @@ +import unittest + +from PIL import Image + +from src.tchmaterial_parser.ui.resource_tree import ( + category_check_state, + draw_checkbox_image, + find_tree_node, + iter_leaf_resources, + should_check_category, +) +from src.tchmaterial_parser.ui.theme import THEME_COLORS + + +RESOURCE_ITEMS = { + "books": { + "display_name": "电子教材", + "children": { + "primary": { + "display_name": "小学", + "children": { + "chinese": { + "display_name": "语文", + "children": { + "book-a": {"display_name": "语文 一年级上册", "content_id": "a001"}, + "book-b": {"display_name": "语文 一年级下册", "content_id": "a002"}, + }, + }, + }, + }, + "middle": { + "display_name": "初中", + "children": { + "math": { + "display_name": "数学", + "children": { + "book-c": {"display_name": "数学 七年级上册", "content_id": "a003"}, + }, + }, + }, + }, + }, + }, +} + + +class IterLeafResourcesTest(unittest.TestCase): + def test_yields_paths_and_data_of_all_leaf_nodes(self) -> None: + leafs = list(iter_leaf_resources(RESOURCE_ITEMS)) + + self.assertEqual( + [leaf_id for leaf_id, _leaf_data in leafs], + ["books:primary:chinese:book-a", "books:primary:chinese:book-b", "books:middle:math:book-c"], + ) + self.assertEqual(leafs[0][1]["content_id"], "a001") + + def test_yields_only_leaves_under_given_parent_path(self) -> None: + leafs = list(iter_leaf_resources(RESOURCE_ITEMS["books"]["children"], "books")) + + self.assertEqual([leaf_id for leaf_id, _leaf_data in leafs], ["books:primary:chinese:book-a", "books:primary:chinese:book-b", "books:middle:math:book-c"]) + + def test_treats_empty_children_as_leaf(self) -> None: # 与 count_resource_items 的口径一致:children 为空字典的节点视为末级资源 + self.assertEqual(list(iter_leaf_resources({})), []) + leafs = list(iter_leaf_resources({"empty": {"display_name": "空分类", "children": {}}}, "books")) + self.assertEqual([leaf_id for leaf_id, _leaf_data in leafs], ["books:empty"]) + + +class FindTreeNodeTest(unittest.TestCase): + def test_locates_node_by_colon_path(self) -> None: + node = find_tree_node(RESOURCE_ITEMS, "books:primary:chinese:book-b") + + self.assertEqual(node["content_id"], "a002") + + def test_locates_category_and_root_nodes(self) -> None: + self.assertEqual(find_tree_node(RESOURCE_ITEMS, "books")["display_name"], "电子教材") + self.assertEqual(find_tree_node(RESOURCE_ITEMS, "books:primary:chinese")["display_name"], "语文") + + def test_returns_none_for_unknown_path(self) -> None: + self.assertIsNone(find_tree_node(RESOURCE_ITEMS, "books:unknown:chinese")) + self.assertIsNone(find_tree_node(RESOURCE_ITEMS, "not-exist")) + + +class CategoryCheckStateTest(unittest.TestCase): + LEAF_IDS = ["books:primary:chinese:book-a", "books:primary:chinese:book-b", "books:middle:math:book-c"] + + def test_returns_unchecked_when_no_leaf_is_checked(self) -> None: + self.assertEqual(category_check_state(self.LEAF_IDS, set()), "unchecked") + self.assertEqual(category_check_state([], {"books:middle:math:book-c"}), "unchecked") + + def test_returns_partial_when_some_leaves_are_checked(self) -> None: + checked = {"books:primary:chinese:book-a"} + + self.assertEqual(category_check_state(self.LEAF_IDS, checked), "partial") + + def test_returns_checked_when_all_leaves_are_checked(self) -> None: + checked = set(self.LEAF_IDS) + + self.assertEqual(category_check_state(self.LEAF_IDS, checked), "checked") + + def test_ignores_checked_items_outside_the_category(self) -> None: + checked = {"books:primary:chinese:book-a", "books:primary:chinese:book-b", "books:other:book-x"} + + self.assertEqual(category_check_state(self.LEAF_IDS[:2], checked), "checked") + + +class ShouldCheckCategoryTest(unittest.TestCase): + LEAF_IDS = ["book-a", "book-b"] + + def test_checks_when_not_all_leaves_are_checked(self) -> None: + self.assertTrue(should_check_category(self.LEAF_IDS, set())) + self.assertTrue(should_check_category(self.LEAF_IDS, {"book-a"})) + + def test_unchecks_when_all_leaves_are_already_checked(self) -> None: + self.assertFalse(should_check_category(self.LEAF_IDS, {"book-a", "book-b"})) + + +class DrawCheckboxImageTest(unittest.TestCase): + def test_draws_all_states_on_transparent_canvas(self) -> None: + colors = THEME_COLORS["light"] + + for state in ("checked", "partial", "unchecked"): + with self.subTest(state=state): + image = draw_checkbox_image(18, state, colors) + + self.assertIsInstance(image, Image.Image) + self.assertEqual(image.size, (18, 18)) + self.assertEqual(image.mode, "RGBA") + # 中心区域被复选框底色填充(不透明) + self.assertEqual(image.getpixel((9, 9))[3], 255) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resource_urls.py b/tests/test_resource_urls.py new file mode 100644 index 00000000..58bfad8e --- /dev/null +++ b/tests/test_resource_urls.py @@ -0,0 +1,100 @@ +import unittest + +from src.tchmaterial_parser.ui.resource_tree import build_resource_url, collect_resource_urls + + +RESOURCE_ITEMS = { + "books": { + "display_name": "电子教材", + "children": { + "high": { + "display_name": "高中", + "children": { + "chinese": { + "display_name": "语文", + "children": { + "tbb": { + "display_name": "统编版", + "children": { + "book-a": { + "display_name": "普通高中教科书·语文 必修上册", + "resource_type_code": "assets_document", + "content_id": "a0001", + }, + "book-b": { + "display_name": "普通高中教科书·语文 必修下册", + "resource_type_code": "assets_document", + "content_id": "a0002", + }, + }, + }, + "rjb": { + "display_name": "人教版", + "children": { + "book-c": { + "display_name": "普通高中教科书·语文 选择性必修上册", + "resource_type_code": "assets_document", + "content_id": "a0003", + }, + }, + }, + }, + }, + }, + }, + }, + }, +} + + +def detail_url(content_id: str) -> str: + return f"https://basic.smartedu.cn/tchMaterial/detail?contentType=assets_document&contentId={content_id}&catalogType=tchMaterial&subCatalog=tchMaterial" + + +def chinese_subject() -> dict: + return RESOURCE_ITEMS["books"]["children"]["high"]["children"]["chinese"] + + +class BuildResourceUrlTest(unittest.TestCase): + def test_builds_detail_url_for_regular_book(self) -> None: + book = chinese_subject()["children"]["tbb"]["children"]["book-a"] + + url = build_resource_url("books:high:chinese:tbb:book-a", book) + + self.assertEqual(url, detail_url("a0001")) + + def test_falls_back_to_default_type_and_path_id(self) -> None: + url = build_resource_url("books:book-x", {"display_name": "未知资源"}) + + self.assertEqual(url, detail_url("book-x")) + + def test_builds_sync_classroom_url_for_teaching_materials(self) -> None: + url = build_resource_url("__internal_national_lesson:tag1:tag2", {"resource_type_code": "teachingmaterials"}) + + self.assertEqual(url, "https://basic.smartedu.cn/syncClassroom?defaultTag=tag1%2Ftag2") + + def test_prepare_lesson_root_adds_prepare_path(self) -> None: + url = build_resource_url("__internal_prepare_lesson:tag1", {"resource_type_code": "teachingmaterials"}) + + self.assertEqual(url, "https://basic.smartedu.cn/syncClassroom/prepare?defaultTag=tag1") + + +class CollectResourceUrlsTest(unittest.TestCase): + def test_collects_all_leaf_urls_under_subject(self) -> None: + urls = collect_resource_urls(chinese_subject()["children"], "books:high:chinese") + + self.assertEqual(urls, [detail_url("a0001"), detail_url("a0002"), detail_url("a0003")]) + + def test_single_edition_only_collects_its_own_books(self) -> None: + tbb = chinese_subject()["children"]["tbb"] + + urls = collect_resource_urls(tbb["children"], "books:high:chinese:tbb") + + self.assertEqual(urls, [detail_url("a0001"), detail_url("a0002")]) + + def test_empty_category_collects_nothing(self) -> None: + self.assertEqual(collect_resource_urls({}), []) + + +if __name__ == "__main__": + unittest.main()