refactor(book): 重构图书服务以提高稳定性和可维护性

- 添加 Path 模块导入并定义允许的内容类型常量
- 改进设置查询逻辑并添加异常处理
- 将请求头配置改为使用 session 对象管理
- 为文件名合法性检查方法添加类型注解和文档字符串
- 更新内容类型检查使用类常量替代硬编码值
- 使用 session 对象替代直接 requests 调用
- 添加响应状态码检查确保请求成功
- 简化重定向处理逻辑并优化文件下载流程
- 移除重复的私有方法实现

app/services/book_service.py

@@ -1,5 +1,6 @@
 import os
 import requests
+from pathlib import Path
 from bs4 import BeautifulSoup
 from app.extensions import db
 from tenacity import retry, stop_after_attempt, RetryError
@@ -8,27 +9,49 @@ from app.models import Setting
 
 
 class BookService:
+    ALLOWED_CONTENT_TYPES = ['"application/octet-stream"', "application/octet-stream", "text/plain",
+                             "application/zip"]
+
     def __init__(self):
-        self.setting = Setting.query.filter_by(name="book_download").first().to_dict()
+        setting_record = Setting.query.filter_by(name="book_download").first()
-        self.setting["targetDate"] = timestamp_to_datetime(
+        if not setting_record:
-            datetime_to_timestamp(self.setting["targetDate"]) - 24 * 60 * 60)
+            raise RuntimeError("Setting 'book_download' not found in database")
+        self.setting = setting_record.to_dict()
+
+        # 目标日期减一天
+        target_ts = datetime_to_timestamp(self.setting["targetDate"]) - 24 * 60 * 60
+        self.setting["targetDate"] = timestamp_to_datetime(target_ts)
+
         app_dir = os.path.dirname(os.path.dirname(__file__))
         self.download_path = os.path.join(os.path.dirname(app_dir), "download")  # 下载目录
         self.today_timestamp = get_today_timestamp()
         self.one_date_loading = False
-        self.headers = {
+        self.session = requests.session()
+        self.session.headers.update({
             "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36",
             "cookie": "cPNj_2132_saltkey=CeJLs8Ed; "
                       "cPNj_2132_auth=44bdzFfx4TYjBMhuKf4AZdi2JM%2BXZ0Okxd0JEl1Hmwbz%2Fr3WiX0sLmATmolWkLlRgorFz%2BULizz5o6G%2FJ3YGWYPF0xw; "
-        }
+        })
+        self.session.timeout = 30
 
-    def legitimate_naming(self, name):  # 合法命名
+    # ------------------------------------------------------------
-        name = name.replace(":", "：")
+    # 数据库辅助方法
-        name = name.replace("<", "《")
+    # ------------------------------------------------------------
-        name = name.replace(">", "》")
+    def _update_setting(self, **kwargs):
-        name = name.replace("/", " ")
+        Setting.query.filter_by(name="book_download").update(kwargs)
-        name = name.replace("\\", " ")
+        db.session.commit()
-        name = name.replace("?", "？")
+        for key, value in kwargs.items():
+            self.setting[key] = value
+
+    # ------------------------------------------------------------
+    # 文件与目录处理
+    # ------------------------------------------------------------
+    @staticmethod
+    def legitimate_naming(name: str) -> str:
+        """将文件名中的非法字符替换为合法字符"""
+        replacements = {":": "：", "<": "《", ">": "》", "/": " ", "\\": " ", "?": "？"}
+        for old, new in replacements.items():
+            name = name.replace(old, new)
         return name
 
     def create_folder(self, target_dir, name):  # 创建文件夹
@@ -40,12 +63,12 @@ class BookService:
             print(f"文件夹：{name} 已存在")
 
     def isDownloadTarget(self, soup):  # 是否是可以下载的文件
-        return soup.headers.get('Content-Type') in ['"application/octet-stream"', "application/octet-stream",
+        return soup.headers.get('Content-Type') in self.ALLOWED_CONTENT_TYPES
-                                                    "text/plain", "application/zip"]
 
     def post_page(self, name, url):  # 帖子页面
         print(f"详情页面 {self.setting["baseUrl"]}{url}")
-        response = requests.get(f"{self.setting["baseUrl"]}{url}", headers=self.headers)
+        response = self.session.get(f"{self.setting["baseUrl"]}{url}")
+        response.raise_for_status()
         print(f"详情页面状态码：{str(response.status_code)}")
         soup = BeautifulSoup(response.text, 'lxml')
         resource_boxs = soup.select('ignore_js_op')
@@ -67,11 +90,8 @@ class BookService:
 
     @retry(stop=stop_after_attempt(3))
     def download_file(self, file_url, dir_name, file_name):  # 下载文件
-        file_res = requests.get(file_url, headers=self.headers, stream=True, allow_redirects=False)
+        file_res = self.session.get(file_url, stream=True)
-        if file_res.status_code in (301, 302, 307, 308):  # 非200状态码，重新请求
+        file_res.raise_for_status()
-            file_url = file_res.headers["location"]
-            print(f'跳转页面了：{file_url}')
-            file_res = requests.get(file_url, headers={"referer": file_url}, stream=True)
         print(f"下载状态码：{file_res.status_code} {file_url}")
         print(dir_name + "/" + file_name)
         if self.isDownloadTarget(file_res):
@@ -93,7 +113,8 @@ class BookService:
             return None
         # yield f"data: 当前页面 {page}\n\n"
         url = f"{self.setting["baseUrl"]}forum.php?mod={self.setting["mod"]}&fid={self.setting["fid"]}&page={page}"  # 版块页面地址
-        response = requests.get(url, headers=self.headers)  # 请求
+        response = self.session.get(url)  # 请求
+        response.raise_for_status()
         soup = BeautifulSoup(response.text, 'lxml')  # 解析
         total_page_element = soup.select_one("a.bm_h")  # 获取总页数元素
         if total_page_element is not None:  # 总页数元素存在 则更新总页数
@@ -162,9 +183,3 @@ class BookService:
         target_page = self.setting["totalPage"] - self.setting["targetCountdownPage"]  # 目标页
         yield from self.section_page(target_page)  # 前往目标页
         yield f"event: close\ndata: 已完成\n\n"
-
-    def _update_setting(self, **kwargs):
-        Setting.query.filter_by(name="book_download").update(kwargs)
-        db.session.commit()
-        for key, value in kwargs.items():
-            self.setting[key] = value