fix: 修复私有资源下载鉴权与镜像重试 (#79)

* fix: 为私有资源请求附加访问令牌

* feat: 支持私有下载镜像自动重试

* fix: 完善私有资源鉴权错误提示

Author: happycola233

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"',
 }

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) # 重置进度条

tests/test_download.py

@@ -1,33 +1,49 @@
 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:
     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,13 +57,104 @@ 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(
                     self.failure_reason(status_code),
                     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"<Error><Code>InvalidArgument</Code></Error>")
+        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()