对整个代码库进行了全面的重构与界面现代化改造，显著提升了代码质量、用户体验和可维护性。

### ✨ 主要变更
- **GUI Modernization**
  - 移除 `qdarkstyle`，改用 `gui/style.qss` 中的自定义浅色主题，带来更统一、清新的视觉体验。
  - 完全重构「映射文件编辑器」：采用左右分栏布局，支持键搜索、动态增删键值对，并支持未保存状态追踪。
  - 「批处理工具」区域引入 `FlowLayout`，实现响应式按钮布局。

- **Code Quality & Style**
  - 对几乎所有 `.py` 文件进行了格式化与 lint 修复。
  - 移除了未使用的导入与多余空行，统一了代码风格。

- **Development & Testing**
  - 新增 `pyproject.toml`，为构建和包管理打下基础。
  - 添加 `TagManager` 的单元测试（`tests/test_tag_manager.py`）。
  - 在 `requirements.txt` 中新增开发依赖：`pytest`, `ruff`, `mypy`。

- **Refactor**
  - 重构 `TagManager` 构造函数，支持路径注入以便于测试。

Author: baiheyyds

.gitignore

@@ -12,6 +12,8 @@ venv/
 .venv/
 
 # Application-generated files
+.mypy_cache/
+.pytest_cache
 cache/
 *.tmp
 *.bak_*

batch_updater.py

@@ -2,12 +2,13 @@
 import asyncio
 import logging
 import re
+from typing import Any, Dict, List
+
 from tqdm import tqdm
-from typing import List, Dict, Any
 
-from core.init import init_context, close_context
-from config.config_token import GAME_DB_ID, BRAND_DB_ID, CHARACTER_DB_ID
 from config.config_fields import FIELDS
+from config.config_token import BRAND_DB_ID, CHARACTER_DB_ID, GAME_DB_ID
+from core.init import close_context, init_context
 
 # --- 可配置项 ---
 # 这现在是每一批次并发处理的数量
@@ -146,20 +147,20 @@ async def write_item_to_notion(context, item_data: Dict[str, Any], db_key: str):
     try:
         if db_key == "games":
             schema = context["schema_manager"].get_schema(config["id"])
-            
+
             # [关键修复] 在提交通知前，对需要分割的字段进行处理
             fields_to_split = ["剧本", "原画", "声优", "音乐", "作品形式"]
             for field in fields_to_split:
                 if field in bangumi_data:
                     raw_values = bangumi_data[field]
                     if not isinstance(raw_values, list):
                         raw_values = [raw_values]
-                    
+
                     processed_names = set()
                     for raw_item in raw_values:
                         split_results = await name_splitter.smart_split(raw_item, interaction_provider)
                         processed_names.update(split_results)
-                    
+
                     bangumi_data[field] = sorted(list(processed_names))
 
             await notion_client.create_or_update_game(
@@ -293,4 +294,4 @@ def get_user_choice():
 if __name__ == "__main__":
     from utils.logger import setup_logging_for_cli
     setup_logging_for_cli()
-    asyncio.run(main())
+    asyncio.run(main())

clients/bangumi_client.py

@@ -2,16 +2,12 @@
 # 该模块用于与 Bangumi API 交互，获取游戏和角色信息
 import asyncio
 import logging
-from rapidfuzz import fuzz
-
-import json
-import os
 import re
-import time
 import unicodedata
-from typing import Set
+from typing import Any, Dict, Set
 
 import httpx
+from rapidfuzz import fuzz
 
 from clients.notion_client import NotionClient
 from config.config_fields import FIELDS
@@ -134,9 +130,9 @@ async def search_and_select_bangumi_id(self, keyword: str) -> str | None:
                     f"🔍 [Bangumi] 模糊匹配成功（放宽判定）: {best['name']} (相似度 {candidates[0][0]:.2f})"
                 )
                 return str(best["id"])
-        
+
         logging.warning("⚠️ Bangumi自动匹配相似度不足，请手动选择:")
-        
+
         # Format candidates for display in GUI
         gui_candidates = []
         for idx, (ratio, item) in enumerate(candidates[:10]):
@@ -145,7 +141,7 @@ async def search_and_select_bangumi_id(self, keyword: str) -> str | None:
 
         # Use the interaction provider to get the user's choice
         selected_id = await self.interaction_provider.get_bangumi_game_choice(keyword, gui_candidates)
-        
+
         return selected_id
 
     async def fetch_game(self, subject_id: str) -> dict:
@@ -170,9 +166,9 @@ async def fetch_game(self, subject_id: str) -> dict:
         game_data.update(infobox_data)
         return game_data
 
-    async def _process_infobox(self, infobox: list, target_db_id: str, bangumi_url: str) -> dict:
-        processed = {}
-        if not infobox:
+    async def _process_infobox(self, infobox: list, target_db_id: str | None, bangumi_url: str) -> dict:
+        processed: Dict[str, Any] = {}
+        if not infobox or not target_db_id:
             return processed
 
         async def _map_and_set_prop(key, value):
@@ -265,7 +261,7 @@ async def fetch_characters(self, subject_id: str) -> list:
 
         characters = []
         for char_summary, detail_resp in zip(char_list_with_actors, responses):
-            if isinstance(detail_resp, Exception) or detail_resp.status_code != 200:
+            if not isinstance(detail_resp, httpx.Response) or detail_resp.status_code != 200:
                 continue
 
             detail = detail_resp.json()
@@ -327,7 +323,7 @@ async def create_or_update_character(self, char: dict, warned_keys: Set[str]) ->
             "BWH": FIELDS["character_bwh"],
             "身高": FIELDS["character_height"],
         }
-        props = {}
+        props: Dict[str, Any] = {}
         for internal_key, value in char.items():
             if not value:
                 continue
@@ -383,7 +379,7 @@ async def create_or_link_characters(self, game_page_id: str, subject_id: str):
                 "PATCH", f"https://api.notion.com/v1/pages/{game_page_id}", patch
             )
             return
-        warned_keys_for_this_game = set()
+        warned_keys_for_this_game: Set[str] = set()
         tasks = [
             self.create_or_update_character(ch, warned_keys_for_this_game) for ch in characters
         ]

clients/base_client.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import os
+
 import httpx
 
 
@@ -45,20 +46,20 @@ async def _request(self, method: str, url: str, **kwargs) -> httpx.Response | No
         try:
             # 确保URL是绝对路径
             full_url = url if url.startswith("http") else f"{self.base_url}{url}"
-            
+
             # 合并默认headers和调用时传入的headers
             request_headers = self.headers.copy()
             if "headers" in kwargs:
                 request_headers.update(kwargs.pop("headers"))
 
             logging.info(f"🔍 [{self.__class__.__name__}] {method.upper()} {full_url}")
-            
+
             response = await self.client.request(method, full_url, headers=request_headers, **kwargs)
             response.raise_for_status()
-            
+
             logging.debug(f"✅ [{self.__class__.__name__}] 请求成功: {response.status_code} {response.reason_phrase}")
             return response
-            
+
         except httpx.HTTPStatusError as e:
             logging.error(f"❌ [{self.__class__.__name__}] 请求失败: {e.response.status_code} for url: {e.request.url}")
             logging.error(f"    -> 响应: {e.response.text[:300]}") # 打印部分响应内容

clients/brand_cache.py

@@ -95,4 +95,4 @@ def _hash_content(self, data: dict) -> str:
             serialized = json.dumps(data, sort_keys=True)
             return hashlib.md5(serialized.encode("utf-8")).hexdigest()
         except Exception:
-            return ""
+            return ""

clients/dlsite_client.py

@@ -12,8 +12,8 @@
 from selenium.webdriver.support.ui import WebDriverWait
 from selenium_stealth import stealth
 
-from utils.driver import create_driver
 from utils.tag_logger import append_new_tags
+
 from .base_client import BaseClient
 
 TAG_JP_PATH = os.path.join(os.path.dirname(__file__), "..", "mapping", "tag_jp_to_cn.json")
@@ -48,7 +48,7 @@ async def search(self, keyword, limit=30):
         logging.info(f"🔍 [Dlsite] 正在搜索关键词: {keyword}")
         query = urllib.parse.quote_plus(keyword)
         url = f"/maniax/fsr/=/language/jp/sex_category%5B0%5D/male/keyword/{query}/work_category%5B0%5D/doujin/work_category%5B1%5D/books/work_category%5B2%5D/pc/work_category%5B3%5D/app/order%5B0%5D/trend/options_and_or/and/per_page/30/page/1/from/fs.header"
-        
+
         resp = await self.get(url, timeout=15)
         if not resp:
             return []
@@ -128,10 +128,10 @@ def extract_list_from_td(table_cell):
                         # Replace <br> tags with a common separator
                         for br in table_cell.find_all("br"):
                             br.replace_with(",")
-                        
+
                         # Get all text, using the common separator
                         all_text = table_cell.get_text(separator=",")
-                        
+
                         # Standardize all separators to the common one and then split
                         processed_text = all_text.replace('、', ',').replace('/', ',').replace('，', ',')
                         return [name.strip() for name in processed_text.split(',') if name.strip()]
@@ -234,4 +234,4 @@ def _blocking_task():
                 )
                 return {}
 
-        return await asyncio.to_thread(_blocking_task)
+        return await asyncio.to_thread(_blocking_task)

clients/fanza_client.py

@@ -1,6 +1,7 @@
 # clients/fanza_client.py
 import logging
 import re
+from typing import Any, Dict, List
 from urllib.parse import quote, urljoin
 
 from bs4 import BeautifulSoup, Tag
@@ -20,26 +21,32 @@ async def search(self, keyword: str, limit=30):
             encoded_keyword = quote(keyword.encode("utf-8", errors="ignore"))
             url = f"/search/?service=pcgame&searchstr={encoded_keyword}&sort=date"
             resp = await self.get(url, timeout=15, cookies=self.cookies)
-            
+
             results = []
             if resp:
                 soup = BeautifulSoup(resp.text, "lxml")
                 result_list = soup.select_one("ul.component-legacy-productTile")
                 if result_list:
                     for li in result_list.find_all("li", class_="component-legacy-productTile__item", limit=limit):
+                        if not isinstance(li, Tag):
+                            continue
                         title_tag = li.select_one(".component-legacy-productTile__title")
                         price_tag = li.select_one(".component-legacy-productTile__price")
                         url_tag = li.select_one("a.component-legacy-productTile__detailLink")
                         type_tag = li.select_one(".component-legacy-productTile__relatedInfo")
                         item_type = type_tag.get_text(strip=True) if type_tag else "未知"
 
-                        if not (title_tag and url_tag and url_tag.has_attr("href")):
+                        if not (title_tag and url_tag):
+                            continue
+
+                        href = url_tag.get("href")
+                        if not isinstance(href, str):
                             continue
 
                         title = title_tag.get_text(strip=True)
                         price_text = price_tag.get_text(strip=True) if price_tag else "未知"
                         price = price_text.split("円")[0].replace(",", "").strip()
-                        full_url = urljoin(self.base_url, url_tag["href"])
+                        full_url = urljoin(self.base_url, href)
 
                         results.append({
                             "title": title, "url": full_url,
@@ -59,13 +66,13 @@ async def search(self, keyword: str, limit=30):
             if final_count > 0:
                 logging.info(f"✅ [Fanza] 主搜索成功，找到 {initial_count} 个原始结果，筛选后剩余 {final_count} 个游戏。")
                 return filtered_results
-            
+
             # --- 后备搜索逻辑 (如果主搜索无结果) ---
             logging.warning("⚠️ [Fanza] 主搜索 (dlsoft) 未找到结果，尝试后备搜索 (mono)...")
-            
+
             fallback_base_url = "https://www.dmm.co.jp"
             url_fallback = f"{fallback_base_url}/mono/-/search/=/searchstr={encoded_keyword}/sort=date/"
-            
+
             resp_fallback = await self.get(url_fallback, timeout=15, cookies=self.cookies)
             if not resp_fallback:
                 logging.error("❌ [Fanza] 后备搜索请求失败。")
@@ -79,24 +86,30 @@ async def search(self, keyword: str, limit=30):
                 return []
 
             for li in result_list_fallback.find_all("li", limit=limit):
+                if not isinstance(li, Tag):
+                    continue
                 url_tag = li.select_one(".tmb a")
                 if not url_tag: continue
-                
+
                 title_tag = url_tag.select_one(".txt")
                 price_tag = li.select_one(".price")
 
-                if not (title_tag and url_tag.has_attr("href")): continue
+                if not (title_tag and url_tag): continue
+
+                href = url_tag.get("href")
+                if not isinstance(href, str):
+                    continue
 
                 title = title_tag.get_text(strip=True)
                 price_text = price_tag.get_text(strip=True) if price_tag else "未知"
                 price = price_text.split("円")[0].replace(",", "").strip()
-                full_url = urljoin(fallback_base_url, url_tag["href"])
+                full_url = urljoin(fallback_base_url, href)
 
                 results_fallback.append({
                     "title": title, "url": full_url,
                     "价格": price or "未知", "类型": "未知(后备)",
                 })
-            
+
             initial_count_fallback = len(results_fallback)
             filtered_results_fallback = [
                 item for item in results_fallback
@@ -117,28 +130,32 @@ async def get_game_detail(self, url: str) -> dict:
 
         try:
             soup = BeautifulSoup(resp.text, "lxml")
-            details = {}
+            details: Dict[str, Any] = {}
+            game_types: List[str] = []
 
             # ==================================================================
             # 智能解析：根据URL判断使用哪套解析逻辑
             # ==================================================================
             if "/mono/" in url:
                 # --- 旧版/后备接口 (`/mono/`) 的解析逻辑 ---
                 logging.info("🔍 [Fanza] 检测到 /mono/ 链接，使用旧版表格解析器。")
-                
+
                 if title_tag := soup.select_one("h1#title"):
                     details["标题"] = title_tag.get_text(strip=True)
-                
+
                 if cover_tag := soup.select_one("#sample-video img, .area-img img"):
                      if src := cover_tag.get("src"):
-                        details["封面图链接"] = urljoin(self.base_url, src)
+                        if isinstance(src, str):
+                            details["封面图链接"] = urljoin(self.base_url, src)
 
                 if main_table := soup.select_one("table.mg-b20"):
                     rows = main_table.find_all("tr")
                     for row in rows:
+                        if not isinstance(row, Tag):
+                            continue
                         cells = row.find_all("td")
                         if len(cells) < 2: continue
-                        
+
                         key = cells[0].get_text(strip=True)
                         value_cell = cells[1]
 
@@ -153,11 +170,9 @@ async def get_game_detail(self, url: str) -> dict:
                         elif key.startswith("ジャンル"):
                             details["标签"] = [a.get_text(strip=True) for a in value_cell.find_all("a")]
                         elif "ゲームジャンル" in key:
-                            game_types = details.get("作品形式", [])
                             genre_text = value_cell.get_text(strip=True).upper()
                             for genre_key, genre_value in self._genre_reverse_mapping.items():
                                 if genre_key in genre_text: game_types.append(genre_value)
-                            if game_types: details["作品形式"] = list(dict.fromkeys(game_types))
                         elif "ボイス" in key:
                             if "あり" in value_cell.get_text(strip=True):
                                 game_types = details.get("作品形式", [])
@@ -168,6 +183,8 @@ async def get_game_detail(self, url: str) -> dict:
                 logging.info("🔍 [Fanza] 未检测到 /mono/ 链接，使用新版解析器。")
                 if top_table := soup.select_one(".contentsDetailTop__table"):
                     for row in top_table.find_all("div", class_="contentsDetailTop__tableRow"):
+                        if not isinstance(row, Tag):
+                            continue
                         key_tag = row.select_one(".contentsDetailTop__tableDataLeft p")
                         value_tag = row.select_one(".contentsDetailTop__tableDataRight")
                         if not (key_tag and value_tag): continue
@@ -200,7 +217,6 @@ def extract_list(value_div: Tag | None) -> list[str]:
                     for key in details:
                         if isinstance(details[key], list): details[key] = sorted(list(set(details[key])))
 
-                    game_types = []
                     if genre_div := find_row_value("ゲームジャンル"):
                         genre_text = genre_div.get_text(strip=True).upper()
                         for key, value in self._genre_reverse_mapping.items():
@@ -220,13 +236,16 @@ def extract_list(value_div: Tag | None) -> list[str]:
                     cover_selector = (".productPreview__mainImage img, #fn-main_image, .main-visual img")
                     if cover_img_tag := soup.select_one(cover_selector):
                         if src := cover_img_tag.get("src"): details["封面图链接"] = urljoin(self.base_url, src)
-                
+
                 if title_tag := soup.select_one("h1.productTitle__txt"):
                     details["标题"] = title_tag.get_text(strip=True)
                 if price_tag := soup.select_one(".priceInformation__price"):
                     details["价格"] = price_tag.get_text(strip=True).replace("円", "").replace(",", "")
 
+            if game_types:
+                details["作品形式"] = sorted(list(dict.fromkeys(game_types)))
+
             return details
         except Exception as e:
             logging.error(f"❌ [Fanza] 解析详情页失败: {e}")
-            return {}
+            return {}