diff --git a/src/tchmaterial_parser/api.py b/src/tchmaterial_parser/api.py index d9d17af..68f3e2e 100644 --- a/src/tchmaterial_parser/api.py +++ b/src/tchmaterial_parser/api.py @@ -2,14 +2,42 @@ # 解析单个资源页面，获取资源标题、下载直链、文件格式与章节目录 import re +from typing import NamedTuple from urllib.parse import urlparse, parse_qs from .network import headers, session from .platform_utils import print_error -def parse(url: str, bookmarks: bool) -> list[tuple[str, str, str, list[dict]]] | None: # 解析资源，获取资源下载链接 +class ResourceInfo(NamedTuple): + title: str + url: str + file_format: str + chapters: list[dict] + edition: str | None = None + +def get_edition_name(resource_data: dict) -> str | None: + """读取资源分类中的教材版别，例如“人教版”“北师大版”。""" + for tag in resource_data.get("tag_list") or []: + if tag.get("tag_dimension_id") == "zxxbb" and tag.get("tag_name"): + return tag["tag_name"] + return None + +def combine_resource_title(root_title: str | None, resource_title: str) -> str: + """组合专题标题与实际资源标题，并避免平台重复标题造成超长文件名。""" + if not root_title: + return resource_title + + # 例如“体育与健康教师用书 基本运动技能（全一册）”的专题父记录与内部 PDF 标题相同； + # 先折叠连续空白再比较，命中时保留子资源原文，避免生成“标题 - 标题”的超长文件名。 + normalized_root = " ".join(root_title.split()).casefold() + normalized_resource = " ".join(resource_title.split()).casefold() + if normalized_root == normalized_resource: + return resource_title + return f"{root_title} - {resource_title}" + +def parse(url: str, bookmarks: bool) -> list[ResourceInfo] | None: # 解析资源，获取资源下载链接 try: - resources_info: list[tuple[str, str, str, list[dict]]] = [] + resources_info: list[ResourceInfo] = [] # 1. 提取 URL 中的 contentId 与 contentType content_id: str | None = None @@ -99,12 +127,13 @@ def parse(url: str, bookmarks: bool) -> list[tuple[str, str, str, list[dict]]] | response = session.get(f"https://s-file-1.ykt.cbern.com.cn/zxx/ndrs/special_edu/resources/details/{content_id}.json") data: dict = response.json() + root_edition = get_edition_name(data) # 3. 获取资源标题、下载链接及章节目录 - def get_resource_info(resource_data: dict, root_title: str | None = None) -> tuple[str, str, str, list[dict]] | None: + def get_resource_info(resource_data: dict, root_title: str | None = None, edition: str | None = None) -> ResourceInfo | None: title_data = resource_data.get("global_title") resource_title: str = title_data.get("zh-CN") or title_data.get("en") if isinstance(title_data, dict) else title_data or resource_data.get("title") or resource_data.get("id") - title = f"{root_title} - {resource_title}" if root_title else resource_title + title = combine_resource_title(root_title, resource_title) resource_url: str | None = None resource_format = "pdf" @@ -215,13 +244,19 @@ def process_tree_nodes(nodes: list[dict]) -> list[dict]: print_error(e) chapters = [] - return title, resource_url, resource_format, chapters + return ResourceInfo( + title, + resource_url, + resource_format, + chapters, + edition or get_edition_name(resource_data), + ) - def get_audio_info(audio_data: dict, root_title: str | None = None) -> tuple[str, str, str, list[dict]] | None: # 解析教材关联的音频资源（如英语教材听力） + def get_audio_info(audio_data: dict, root_title: str | None = None, edition: str | None = None) -> ResourceInfo | None: # 解析教材关联的音频资源（如英语教材听力） # 音频资源的标题存放在 global_title 字典中（键为语言代码，如 zh-CN） title_data = audio_data.get("global_title") audio_title: str = title_data.get("zh-CN") or title_data.get("en") if isinstance(title_data, dict) else title_data or audio_data.get("title") or audio_data.get("id") - title = f"{root_title} - {audio_title}" if root_title else audio_title + title = combine_resource_title(root_title, audio_title) resource_url: str | None = None resource_format = "mp3" @@ -242,13 +277,19 @@ def get_audio_info(audio_data: dict, root_title: str | None = None) -> tuple[str if not resource_url: return None - return title, resource_url, resource_format, [] + return ResourceInfo( + title, + resource_url, + resource_format, + [], + edition or get_edition_name(audio_data), + ) if content_type == "thematic_course": # 专题课程 resources_resp = session.get(f"https://s-file-1.ykt.cbern.com.cn/zxx/ndrs/special_edu/thematic_course/{content_id}/resources/list.json") resources_data: list[dict] = resources_resp.json() for resource in resources_data: - resource_info = get_resource_info(resource, data["title"]) + resource_info = get_resource_info(resource, data["title"], root_edition) if resource_info: resources_info.append(resource_info) elif data.get("relations"): # 课程包等多资源页面（含导学案、课件、PPT 等） @@ -256,7 +297,7 @@ def get_audio_info(audio_data: dict, root_title: str | None = None) -> tuple[str if not isinstance(resources, list): continue for resource in resources: - resource_info = get_resource_info(resource, data.get("title")) + resource_info = get_resource_info(resource, data.get("title"), root_edition) if resource_info: resources_info.append(resource_info) else: # 其他类型资源 @@ -269,7 +310,7 @@ def get_audio_info(audio_data: dict, root_title: str | None = None) -> tuple[str audios_resp = session.get(f"https://s-file-1.ykt.cbern.com.cn/zxx/ndrs/resources/{content_id}/relation_audios.json") audios_data: list[dict] = audios_resp.json() for audio in audios_data: - audio_info = get_audio_info(audio, data.get("title")) + audio_info = get_audio_info(audio, data.get("title"), root_edition) if audio_info: resources_info.append(audio_info) except Exception: # 音频资源不是必需的，获取失败时直接跳过 diff --git a/src/tchmaterial_parser/ui/download_panel.py b/src/tchmaterial_parser/ui/download_panel.py index 231809d..28e55df 100644 --- a/src/tchmaterial_parser/ui/download_panel.py +++ b/src/tchmaterial_parser/ui/download_panel.py @@ -4,6 +4,7 @@ import os, re, traceback import tkinter as tk +from collections import Counter from tkinter import ttk, messagebox, filedialog from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from xml.etree import ElementTree @@ -12,7 +13,7 @@ from .runtime import thread_it, ui_call from .. import config -from ..api import parse +from ..api import ResourceInfo, parse from ..bookmarks import add_bookmarks from ..network import headers, session from ..platform_utils import print_error @@ -109,6 +110,45 @@ def download_failure_reason(response, attempted_urls: list[str]) -> str: reason += f"，已尝试 {len(attempted_urls)} 个下载镜像" return reason +def download_filename(resource: ResourceInfo) -> str: + return f"{resource.title or 'download'}.{resource.file_format}" + +def filename_key(filename: str) -> str: + """以跨平台保守方式比较文件名，提前避开 Windows/macOS 上的大小写冲突。""" + return os.path.normcase(filename).casefold() + +def allocate_download_paths(resources: list[ResourceInfo], directory: str) -> list[str]: + """在线程启动前为批量任务分配唯一目标路径，防止多个线程共用同一个 .tmp 文件。""" + base_filenames = [download_filename(resource) for resource in resources] + base_counts = Counter(filename_key(filename) for filename in base_filenames) + + edition_filenames: list[str] = [] + for resource, filename in zip(resources, base_filenames): + # 例如人教版与北师大版的“普通高中教科书·英语必修 第三册”同名时，优先使用易读的版别前缀区分。 + if base_counts[filename_key(filename)] > 1 and resource.edition: + filename = f"[{resource.edition}] {filename}" + edition_filenames.append(filename) + + reserved_paths: set[str] = set() + allocated_paths: list[str] = [] + for filename in edition_filenames: + candidate = os.path.join(directory, filename) + stem, extension = os.path.splitext(candidate) + sequence = 2 + + # 同时检查最终文件和可辨识的“最终文件.tmp”；后者可能属于另一个仍在运行的程序实例。 + while ( + filename_key(candidate) in reserved_paths + or os.path.exists(candidate) + or os.path.exists(f"{candidate}.tmp") + ): + candidate = f"{stem} ({sequence}){extension}" + sequence += 1 + + reserved_paths.add(filename_key(candidate)) + 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 @@ -124,7 +164,7 @@ def parse_and_copy() -> None: # 解析并复制链接 failed_urls.add(url) # 添加到失败链接 continue for resource in resources_info: - resource_urls.add(resource[1]) + resource_urls.add(resource.url) if failed_urls: messagebox.showwarning("警告", "以下 “行” 无法解析：\n" + "\n".join(failed_urls)) @@ -150,7 +190,7 @@ def download() -> None: # 下载资源文件 download_btn.config(state="disabled") # 设置下载按钮为禁用状态 download_states = [] # 初始化下载状态 urls = {line.strip() for line in url_text.get("1.0", "end").splitlines() if line.strip()} # 获取所有非空行并去重 - resources_info_list: list[tuple[str, str, str, list[dict]]] = [] + resources_info_list: list[ResourceInfo] = [] resource_urls: set[str] = set() failed_urls: set[str] = set() @@ -165,7 +205,7 @@ def download() -> None: # 下载资源文件 failed_urls.add(url) continue for resource in resources_info: - resource_url = resource[1] + resource_url = resource.url if resource_url in resource_urls: # 直接使用 resources_info_list 会报错（list 不可哈希） continue resources_info_list.append(resource) @@ -181,22 +221,25 @@ def download() -> None: # 下载资源文件 else: dir_path = None - for resource in resources_info_list: - title, resource_url, resource_format, chapters = resource - default_filename = title or "download" - if dir_path: - save_path = os.path.join(dir_path, f"{default_filename}.{resource_format}") # 构造完整路径 - else: + 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_format}", - filetypes=[(f"{resource_format.upper()} 文件", f"*.{resource_format}"), ("所有文件", "*.*")], - initialfile=default_filename, + defaultextension=f".{resource.file_format}", + filetypes=[(f"{resource.file_format.upper()} 文件", f"*.{resource.file_format}"), ("所有文件", "*.*")], + initialfile=resource.title or "download", ) if not save_path: # 用户取消了文件保存操作 download_btn.config(state="normal") # 恢复下载按钮为启用状态 return save_path = os.path.normpath(save_path) - thread_it(download_file, resource_url, save_path, chapters) # 开始下载（多线程，防止窗口卡死） + download_targets.append((resource, save_path)) + + for resource, save_path in download_targets: + thread_it(download_file, resource.url, save_path, resource.chapters) # 开始下载（多线程，防止窗口卡死） if failed_urls: messagebox.showwarning("警告", "以下 “行” 无法解析：\n" + "\n".join(failed_urls)) # 显示警告对话框 diff --git a/tests/test_audio_parse.py b/tests/test_audio_parse.py index 476df0f..f2a0d35 100644 --- a/tests/test_audio_parse.py +++ b/tests/test_audio_parse.py @@ -25,6 +25,12 @@ def get(self, url: str, *args: tuple, **kwargs: dict) -> FakeResponse: DETAILS = { "id": "book-1", "title": "英语七年级上册", + "tag_list": [ + { + "tag_dimension_id": "zxxbb", + "tag_name": "人教版", + }, + ], "ti_items": [ { "ti_is_source_file": True, @@ -69,7 +75,7 @@ class AudioParseTest(unittest.TestCase): def setUp(self) -> None: self.addCleanup(setattr, api, "session", api.session) - def parse_book(self) -> list[tuple[str, str, str, list[dict]]] | None: + def parse_book(self) -> list[api.ResourceInfo] | None: api.session = FakeSession(DETAILS, AUDIOS) url = "https://basic.smartedu.cn/tchMaterial/detail?contentType=assets_document&contentId=book-1&catalogType=tchMaterial&subCatalog=tchMaterial" return api.parse(url, False) @@ -79,9 +85,11 @@ def test_textbook_with_audio_returns_pdf_and_mp3s(self) -> None: self.assertIsNotNone(results) self.assertEqual(len(results), 3) self.assertEqual(results[0][2], "pdf") + self.assertEqual(results[0].edition, "人教版") self.assertEqual(results[1][1], "https://r1-ndr-private.ykt.cbern.com.cn/edu_product/esp/assets/audio-1.t/1.mp3") self.assertEqual(results[1][2], "mp3") self.assertEqual(results[1][0], "英语七年级上册 - 1 Starter Section 2 Activity 2") + self.assertEqual(results[1].edition, "人教版") self.assertEqual(results[2][0], "英语七年级上册 - 2 Starter Section 3 Activity 1") def test_textbook_without_audio_returns_only_pdf(self) -> None: diff --git a/tests/test_download_paths.py b/tests/test_download_paths.py new file mode 100644 index 0000000..b130d73 --- /dev/null +++ b/tests/test_download_paths.py @@ -0,0 +1,81 @@ +import os +import tempfile +import unittest + +from src.tchmaterial_parser.api import ResourceInfo +from src.tchmaterial_parser.ui.download_panel import allocate_download_paths + + +def resource( + title: str, + resource_key: str, + edition: str | None = None, + file_format: str = "pdf", +) -> ResourceInfo: + return ResourceInfo( + title=title, + url=f"https://example.com/{resource_key}.{file_format}", + file_format=file_format, + chapters=[], + edition=edition, + ) + + +class DownloadPathTest(unittest.TestCase): + def test_keeps_unique_filenames_unchanged(self) -> None: + resources = [ + resource("语文第一册", "book-1", "人教版"), + resource("数学第一册", "book-2", "北师大版"), + ] + + 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_uses_edition_prefix_for_same_title_from_different_editions(self) -> None: + title = "普通高中教科书·英语必修 第三册" + resources = [ + resource(title, "bf54b36f-4c75-4c91-8b9c-53ce15e4f903", "人教版"), + resource(title, "1e2e7507-0db6-4505-af12-87baac887bc1", "北师大版"), + ] + + with tempfile.TemporaryDirectory() as directory: + paths = allocate_download_paths(resources, directory) + + self.assertEqual([os.path.basename(path) for path in paths], [ + "[人教版] 普通高中教科书·英语必修 第三册.pdf", + "[北师大版] 普通高中教科书·英语必修 第三册.pdf", + ]) + self.assertEqual(len({f"{path}.tmp" for path in paths}), 2) + + def test_uses_sequence_when_edition_cannot_resolve_collision(self) -> None: + resources = [ + resource("同名教材", "aaaaaaaa-1111-2222-3333-444444444444", "人教版"), + resource("同名教材", "bbbbbbbb-1111-2222-3333-444444444444", "人教版"), + ] + + with tempfile.TemporaryDirectory() as directory: + paths = allocate_download_paths(resources, directory) + + self.assertEqual([os.path.basename(path) for path in paths], [ + "[人教版] 同名教材.pdf", + "[人教版] 同名教材 (2).pdf", + ]) + + def test_avoids_existing_final_and_temporary_files(self) -> None: + resources = [ + resource("已有教材", "book-1"), + resource("未完成教材", "book-2"), + ] + + with tempfile.TemporaryDirectory() as directory: + open(os.path.join(directory, "已有教材.pdf"), "wb").close() + open(os.path.join(directory, "未完成教材.pdf.tmp"), "wb").close() + paths = allocate_download_paths(resources, directory) + + self.assertEqual([os.path.basename(path) for path in paths], ["已有教材 (2).pdf", "未完成教材 (2).pdf"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_resource_title.py b/tests/test_resource_title.py new file mode 100644 index 0000000..c4fd97a --- /dev/null +++ b/tests/test_resource_title.py @@ -0,0 +1,25 @@ +import unittest + +from src.tchmaterial_parser.api import combine_resource_title + + +class ResourceTitleTest(unittest.TestCase): + def test_keeps_child_title_when_parent_is_missing(self) -> None: + self.assertEqual(combine_resource_title(None, "子资源"), "子资源") + + def test_joins_distinct_parent_and_child_titles(self) -> None: + self.assertEqual(combine_resource_title("英语七年级上册", "听力 1"), "英语七年级上册 - 听力 1") + + def test_deduplicates_titles_that_only_differ_in_whitespace(self) -> None: + child_title = "义务教育教科书•体育与健康教师用书 基本运动技能（全一册）" + parent_title = "义务教育教科书•体育与健康教师用书 基本运动技能（全一册）" + self.assertEqual(combine_resource_title(parent_title, child_title), child_title) + + def test_issue_76_title_stays_within_common_filesystem_limit(self) -> None: + title = "（根据2022年版课程标准修订）义务教育教科书•体育与健康教师用书 基本运动技能（全一册）" + filename = f"{combine_resource_title(title, title)}.pdf.tmp" + self.assertLessEqual(len(filename.encode("utf-8")), 255) + + +if __name__ == "__main__": + unittest.main()