diff --git a/src/tchmaterial_parser/network.py b/src/tchmaterial_parser/network.py index b36e108..b518f93 100644 --- a/src/tchmaterial_parser/network.py +++ b/src/tchmaterial_parser/network.py @@ -11,5 +11,6 @@ "Origin": "https://basic.smartedu.cn", "Referer": "https://basic.smartedu.cn/", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", - "X-ND-AUTH": 'MAC id="0",nonce="0",mac="0"', # “MAC id” 即为 Access Token,“nonce” 和 “mac” 不可缺省但可为任意非空值 + # 保留旧版匿名请求格式;设置 Token 后,私有 CDN 的主要鉴权入口由下载 URL 的 accessToken 参数提供。 + "X-ND-AUTH": 'MAC id="0",nonce="0",mac="0"', } diff --git a/src/tchmaterial_parser/ui/download_panel.py b/src/tchmaterial_parser/ui/download_panel.py index 0918576..231809d 100644 --- a/src/tchmaterial_parser/ui/download_panel.py +++ b/src/tchmaterial_parser/ui/download_panel.py @@ -2,9 +2,13 @@ # 下载面板:解析并复制直链、下载资源文件与进度反馈 # 本模块持有与下载相关的几个控件句柄,因此这些控件的读写不必跨模块 -import os, traceback +import os, re, traceback import tkinter as tk from tkinter import ttk, messagebox, filedialog +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit +from xml.etree import ElementTree + +from requests import RequestException from .runtime import thread_it, ui_call from .. import config @@ -14,6 +18,96 @@ from ..platform_utils import print_error download_states: list[dict] = [] # 初始化下载状态 +PRIVATE_DOWNLOAD_HOSTS = tuple(f"r{index}-ndr-private.ykt.cbern.com.cn" for index in range(1, 4)) + +def redact_access_token(text: str) -> str: + """隐藏 URL 查询参数里的 Token,防止网络异常把凭据带入日志或弹窗。""" + return re.sub(r"([?&]accessToken=)[^&\s'\"]+", r"\1<已隐藏>", text, flags=re.IGNORECASE) + +def authenticated_download_url(url: str) -> str: + """仅在真正发起请求时为私有资源附加 Token,避免把凭据写入状态或错误信息。""" + parts = urlsplit(url) + hostname = parts.hostname or "" + if not config.access_token or not hostname.endswith("-ndr-private.ykt.cbern.com.cn"): + return url + + query = [(name, value) for name, value in parse_qsl(parts.query, keep_blank_values=True) if name != "accessToken"] + query.append(("accessToken", config.access_token)) + return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) + +def download_mirror_urls(url: str) -> list[str]: + """按原地址优先的顺序生成私有 CDN 镜像,普通下载地址保持不变。""" + parts = urlsplit(url) + hostname = parts.hostname or "" + if hostname not in PRIVATE_DOWNLOAD_HOSTS: + return [url] + + ordered_hosts = [hostname, *(host for host in PRIVATE_DOWNLOAD_HOSTS if host != hostname)] + return [urlunsplit((parts.scheme, host, parts.path, parts.query, parts.fragment)) for host in ordered_hosts] + +def request_download(url: str): + """请求资源并在镜像出错时自动切换,返回最终响应和已尝试的无凭据地址。""" + attempted_urls: list[str] = [] + last_response = None + last_exception: RequestException | None = None + + for candidate_url in download_mirror_urls(url): + attempted_urls.append(candidate_url) + try: + response = session.get(authenticated_download_url(candidate_url), headers=headers, stream=True) + except RequestException as e: + last_exception = e + continue + + if response.ok: + if last_response is not None: + last_response.close() + return response, attempted_urls + + if last_response is not None: + last_response.close() + last_response = response + + # 认证失败通常与镜像无关,立即返回以免重复请求。 + if response.status_code in (401, 403): + break + + if last_response is not None: + return last_response, attempted_urls + if last_exception is not None: + # requests 的异常文字通常包含完整请求 URL,此处重新包装以清除查询参数中的 Token。 + raise RuntimeError(redact_access_token(str(last_exception))) from None + raise RuntimeError("没有可用的下载地址") + +def storage_error_code(response) -> str | None: + """读取对象存储返回的 XML 错误码;非 XML 响应保持原有通用提示。""" + try: + root = ElementTree.fromstring(response.content) + return root.findtext("Code") + except (AttributeError, ElementTree.ParseError, TypeError): + return None + +def download_failure_reason(response, attempted_urls: list[str]) -> str: + status_code = response.status_code + error_code = storage_error_code(response) + reason = f"服务器返回 HTTP 状态码 {status_code}" + if error_code: + reason += f"({error_code})" + + if status_code in (401, 403): + if config.access_token: + reason += ",Access Token 可能已过期或无效,请重新设置" + else: + reason += ",该资源需要有效的 Access Token,请先设置" + elif status_code == 400 and error_code == "InvalidArgument": + if config.access_token: + reason += ",私有资源鉴权失败,Access Token 可能已过期或无效,请重新设置" + else: + reason += ",该私有资源需要有效的 Access Token,请先设置" + + if len(attempted_urls) > 1: + reason += f",已尝试 {len(attempted_urls)} 个下载镜像" + return reason 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 @@ -115,12 +209,13 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) download_states.append(current_state) temp_path = f"{save_path}.tmp" + response = None try: - response = session.get(url, headers=headers, stream=True) + response, attempted_urls = request_download(url) if not response.ok: # 服务器返回表示错误的 HTTP 状态码 current_state["finished"] = True - current_state["failed_reason"] = f"服务器返回 HTTP 状态码 {response.status_code}" + (",Access Token 可能已过期或无效,请重新设置" if response.status_code in (401, 403) else "") + current_state["failed_reason"] = download_failure_reason(response, attempted_urls) else: current_state["total_size"] = int(response.headers.get("Content-Length", 0)) @@ -161,11 +256,14 @@ def download_file(url: str, save_path: str, chapters: list[dict] | None = None) print_error(e) current_state["downloaded_size"], current_state["total_size"] = 0, 0 current_state["finished"] = True - current_state["failed_reason"] = traceback.format_exc().rstrip() + current_state["failed_reason"] = redact_access_token(traceback.format_exc().rstrip()) try: os.remove(temp_path) except Exception: pass + finally: + if response is not None: + response.close() if all(state["finished"] for state in download_states): # 所有文件下载完成 ui_call(download_progress_bar.config, value=0) # 重置进度条 diff --git a/tests/test_download.py b/tests/test_download.py index d5a56fe..767b2d6 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -1,20 +1,29 @@ import unittest +from requests import ConnectionError + from src.tchmaterial_parser.ui import download_panel, runtime class FakeResponse: - def __init__(self, status_code: int) -> None: - self.ok = False + def __init__(self, status_code: int, content: bytes = b"") -> None: + self.ok = status_code < 400 self.status_code = status_code + self.content = content + + def close(self) -> None: + pass class FakeSession: - def __init__(self, status_code: int) -> None: - self.status_code = status_code + def __init__(self, status_code: int | list[int]) -> None: + self.status_codes = status_code if isinstance(status_code, list) else [status_code] + self.requested_urls: list[str] = [] def get(self, *args: tuple, **kwargs: dict) -> FakeResponse: - return FakeResponse(self.status_code) + self.requested_urls.append(args[0]) + status_code = self.status_codes[min(len(self.requested_urls) - 1, len(self.status_codes) - 1)] + return FakeResponse(status_code) class FakeWidget: @@ -22,12 +31,19 @@ def config(self, **kwargs: dict) -> None: pass +class FailingSession: + def get(self, url: str, **kwargs: dict) -> FakeResponse: + raise ConnectionError(f"无法访问 {url}") + + class DownloadFailureTest(unittest.TestCase): def setUp(self) -> None: # 置为关闭状态后 ui_call() 不会真正执行回调,因此桩控件只需提供 config 属性 runtime.app_closing = True self.addCleanup(setattr, runtime, "app_closing", False) self.addCleanup(setattr, download_panel, "session", download_panel.session) + previous_token = download_panel.config.access_token + self.addCleanup(setattr, download_panel.config, "access_token", previous_token) widget = FakeWidget() download_panel.bind_widgets(widget, widget, widget, widget, widget) @@ -41,6 +57,7 @@ def test_reports_server_errors_unrelated_to_the_token(self) -> None: self.assertEqual(self.failure_reason(404), "服务器返回 HTTP 状态码 404") def test_appends_a_token_hint_to_authentication_failures(self) -> None: + download_panel.config.access_token = "private-token" for status_code in (401, 403): with self.subTest(status_code=status_code): self.assertEqual( @@ -48,6 +65,96 @@ def test_appends_a_token_hint_to_authentication_failures(self) -> None: f"服务器返回 HTTP 状态码 {status_code},Access Token 可能已过期或无效,请重新设置", ) + def test_asks_for_token_when_anonymous_request_requires_authentication(self) -> None: + download_panel.config.access_token = None + self.assertEqual( + self.failure_reason(401), + "服务器返回 HTTP 状态码 401,该资源需要有效的 Access Token,请先设置", + ) + + def test_adds_access_token_only_to_private_request_url(self) -> None: + token = "private-token" + download_panel.config.access_token = token + fake_session = FakeSession(404) + download_panel.session = fake_session + original_url = "https://r1-ndr-private.ykt.cbern.com.cn/book.pdf?source=catalog" + + download_panel.download_states = [] + download_panel.download_file(original_url, "book.pdf") + + requested_url = fake_session.requested_urls[0] + self.assertIn("source=catalog", requested_url) + self.assertIn("accessToken=private-token", requested_url) + self.assertEqual(download_panel.download_states[0]["download_url"], original_url) + self.assertNotIn(token, download_panel.download_states[0]["failed_reason"]) + + def test_keeps_anonymous_and_non_private_urls_unchanged(self) -> None: + private_url = "https://r1-ndr-private.ykt.cbern.com.cn/book.pdf" + public_url = "https://example.com/book.pdf" + + download_panel.config.access_token = None + self.assertEqual(download_panel.authenticated_download_url(private_url), private_url) + + download_panel.config.access_token = "private-token" + self.assertEqual(download_panel.authenticated_download_url(public_url), public_url) + + def test_retries_private_download_on_the_next_mirror(self) -> None: + download_panel.config.access_token = "private-token" + fake_session = FakeSession([500, 200]) + download_panel.session = fake_session + original_url = "https://r1-ndr-private.ykt.cbern.com.cn/book.pdf" + + response, attempted_urls = download_panel.request_download(original_url) + + self.assertEqual(response.status_code, 200) + self.assertEqual([url.split("/", 3)[2] for url in attempted_urls], [ + "r1-ndr-private.ykt.cbern.com.cn", + "r2-ndr-private.ykt.cbern.com.cn", + ]) + self.assertTrue(all("accessToken=private-token" in url for url in fake_session.requested_urls)) + self.assertTrue(all("accessToken" not in url for url in attempted_urls)) + + def test_does_not_retry_authentication_failures_or_public_urls(self) -> None: + private_url = "https://r1-ndr-private.ykt.cbern.com.cn/book.pdf" + public_url = "https://example.com/book.pdf" + + private_session = FakeSession(401) + download_panel.session = private_session + _response, private_attempts = download_panel.request_download(private_url) + self.assertEqual(private_attempts, [private_url]) + + public_session = FakeSession(500) + download_panel.session = public_session + _response, public_attempts = download_panel.request_download(public_url) + self.assertEqual(public_attempts, [public_url]) + + def test_explains_private_storage_authentication_errors(self) -> None: + response = FakeResponse(400, b"InvalidArgument") + attempted_urls = ["https://r1.example/book.pdf", "https://r2.example/book.pdf"] + + download_panel.config.access_token = None + self.assertEqual( + download_panel.download_failure_reason(response, attempted_urls), + "服务器返回 HTTP 状态码 400(InvalidArgument),该私有资源需要有效的 Access Token,请先设置,已尝试 2 个下载镜像", + ) + + download_panel.config.access_token = "private-token" + self.assertEqual( + download_panel.download_failure_reason(response, attempted_urls), + "服务器返回 HTTP 状态码 400(InvalidArgument),私有资源鉴权失败,Access Token 可能已过期或无效,请重新设置,已尝试 2 个下载镜像", + ) + + def test_redacts_token_from_network_exceptions(self) -> None: + token = "private-token" + download_panel.config.access_token = token + download_panel.session = FailingSession() + + with self.assertRaises(RuntimeError) as context: + download_panel.request_download("https://r1-ndr-private.ykt.cbern.com.cn/book.pdf") + + self.assertNotIn(token, str(context.exception)) + self.assertIn("accessToken=<已隐藏>", str(context.exception)) + if __name__ == "__main__": unittest.main()