Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 52 additions & 11 deletions src/tchmaterial_parser/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"

Expand Down Expand Up @@ -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"

Expand All @@ -242,21 +277,27 @@ 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 等)
for resources in data["relations"].values():
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: # 其他类型资源
Expand All @@ -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: # 音频资源不是必需的,获取失败时直接跳过
Expand Down
71 changes: 57 additions & 14 deletions src/tchmaterial_parser/ui/download_panel.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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))
Expand All @@ -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()

Expand All @@ -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)
Expand All @@ -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)) # 显示警告对话框
Expand Down
10 changes: 9 additions & 1 deletion tests/test_audio_parse.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand Down
81 changes: 81 additions & 0 deletions tests/test_download_paths.py
Original file line number Diff line number Diff line change
@@ -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()
25 changes: 25 additions & 0 deletions tests/test_resource_title.py
Original file line number Diff line number Diff line change
@@ -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()