From 1eaef2aee82647f5af97003a7abad69414fa887f Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 17 Oct 2026 01:18:26 +0800 Subject: [PATCH] refactor(types): govern nexus_audiences site parser --- app/modules/indexer/parser/nexus_audiences.py | 87 +++++++++++-------- 1 file changed, 49 insertions(+), 38 deletions(-) diff --git a/app/modules/indexer/parser/nexus_audiences.py b/app/modules/indexer/parser/nexus_audiences.py index 1c2ee2209..c85401c1f 100644 --- a/app/modules/indexer/parser/nexus_audiences.py +++ b/app/modules/indexer/parser/nexus_audiences.py @@ -1,34 +1,41 @@ # -*- coding: utf-8 -*- import json import re +from typing import Optional from urllib.parse import urljoin from lxml import etree +from app.adapters.network.http import RequestUtils from app.foundation import size as size_tools from app.foundation import text as text_tools from app.foundation.dom import DomUtils -from app.modules.indexer.parser import SiteSchema +from app.modules.indexer.parser import MessageContent, SiteSchema from app.modules.indexer.parser.nexus_php import NexusPhpSiteUserInfo from app.runtime.log import logger class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): + """解析 NexusAudiences 站点的用户资料、流量与消息。""" schema = SiteSchema.NexusAudiences __UNKNOWN_UNREAD_COUNT = 99999 - def __init__(self, *args, **kwargs): + def __init__(self, site_name: str, url: str, site_cookie: Optional[str], apikey: Optional[str], + token: Optional[str], request_utils: Optional[RequestUtils] = None, + ua: Optional[str] = None, emulate: bool = False, + proxy: Optional[bool] = None, api_url: Optional[str] = None) -> None: """ 初始化 Audiences 未读私信列表地址，第一页不能携带 page 参数。 """ - super().__init__(*args, **kwargs) + super().__init__(site_name, url, site_cookie, apikey, token, request_utils, + ua, emulate, proxy, api_url) self._user_mail_unread_page = self.__build_unread_mailbox_page(box=1) self._sys_mail_unread_page = None self.__next_mail_page = 1 - self.__seen_unread_message_links = set() - self.__message_list_previews = {} + self.__seen_unread_message_links: set[str] = set() + self.__message_list_previews: dict[str, MessageContent] = {} - def _parse_message_unread(self, html_text): + def _parse_message_unread(self, html_text: str) -> None: """ 解析 Audiences 新版顶部用户栏中的未读消息数。 """ @@ -58,7 +65,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): super()._parse_message_unread(html_text) - def _parse_message_unread_links(self, html_text: str, msg_links: list): + def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]: """ 解析 Audiences 未读消息链接。 """ @@ -82,7 +89,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return next_page - def _parse_message_content(self, html_text): + def _parse_message_content(self, html_text: str) -> tuple[Optional[str], Optional[str], Optional[str]]: """ 解析 Audiences 新版短消息详情页。 """ @@ -106,12 +113,12 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return super()._parse_message_content(html_text) - def _pase_unread_msgs(self): + def _pase_unread_msgs(self) -> None: """ 解析 Audiences 未读消息，避免异常分页重复通知和空详情通知。 """ self.__reset_unread_message_parse_state() - unread_msg_links = [] + unread_msg_links: list[str] = [] if self.message_unread > 0 or self.message_read_force: next_page = self.__parse_unread_message_list_page( link=self._user_mail_unread_page, @@ -142,7 +149,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): continue self.message_unread_contents.append((head, date, content)) - def __parse_unread_message_list_page(self, link: str, unread_msg_links: list): + def __parse_unread_message_list_page(self, link: Optional[str], unread_msg_links: list[str]) -> Optional[str]: """ 读取并解析一页 Audiences 未读消息列表。 """ @@ -157,7 +164,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): unread_msg_links ) - def __reset_unread_message_parse_state(self): + def __reset_unread_message_parse_state(self) -> None: """ 重置 Audiences 未读消息分页状态，避免复用解析器时沿用上次页码和去重集合。 """ @@ -165,7 +172,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): self.__seen_unread_message_links.clear() self.__message_list_previews.clear() - def __filter_new_message_links(self, message_links: list) -> list: + def __filter_new_message_links(self, message_links: list[str]) -> list[str]: """ 过滤 Audiences 异常分页重复返回的消息详情链接。 """ @@ -179,16 +186,17 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return new_message_links @staticmethod - def __parse_table_unread_message_links(html) -> list: + def __parse_table_unread_message_links(html: etree._Element) -> list[str]: """ 解析 Audiences 旧版表格消息列表中的未读消息链接。 """ - return html.xpath( + links: list[str] = html.xpath( '//tr[.//img[contains(concat(" ", normalize-space(@class), " "), " unreadpm ") ' 'or @alt="Unread" or @title="未读"]]/td/a[contains(@href, "viewmessage")]/@href' ) + return links - def __parse_pm_item_unread_message_links(self, html) -> list: + def __parse_pm_item_unread_message_links(self, html: etree._Element) -> list[str]: """ 解析 Audiences 新版 pm-item 私信列表中的未读消息链接。 """ @@ -215,7 +223,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): self.__cache_pm_item_preview(message_link, row) return message_links - def __cache_pm_item_preview(self, message_link: str, row): + def __cache_pm_item_preview(self, message_link: str, row: etree._Element) -> None: """ 缓存新版列表页预览，用于详情页结构变化时兜底生成站点消息。 """ @@ -234,7 +242,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): self.__message_list_previews[urljoin(self._base_url, message_link)] = (head, date, content) @staticmethod - def __extract_pm_item_text(row, xpath: str): + def __extract_pm_item_text(row: etree._Element, xpath: str) -> Optional[str]: """ 提取新版私信列表节点文本并规整空白字符。 """ @@ -246,7 +254,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return text or None @staticmethod - def __extract_first_text(html, xpath: str): + def __extract_first_text(html: etree._Element, xpath: str) -> Optional[str]: """ 提取第一个匹配节点的规整文本。 """ @@ -256,7 +264,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return NexusAudiencesSiteUserInfo.__normalize_text(nodes[0].xpath("string(.)")) @staticmethod - def __extract_pm_view_meta(html, label: str): + def __extract_pm_view_meta(html: etree._Element, label: str) -> Optional[str]: """ 按标签提取 Audiences 新版短消息详情页中的元信息。 """ @@ -271,7 +279,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return NexusAudiencesSiteUserInfo.__normalize_text(values[0].xpath("string(.)")) @staticmethod - def __normalize_text(text: str): + def __normalize_text(text: str) -> Optional[str]: """ 规整 Audiences 新版消息页文本空白字符。 """ @@ -280,7 +288,9 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): text = re.sub(r"\s+", " ", text.replace("\xa0", " ")).strip() return text or None - def __fill_empty_message_content_from_list(self, msg_link: str, head, date, content): + def __fill_empty_message_content_from_list( + self, msg_link: str, head: Optional[str], date: Optional[str], content: Optional[str] + ) -> MessageContent: """ 使用列表页预览填补详情页解析不到的字段。 """ @@ -290,7 +300,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): preview_head, preview_date, preview_content = preview return head or preview_head, date or preview_date, content or preview_content - def __should_fetch_next_unread_page(self, new_message_links: list) -> bool: + def __should_fetch_next_unread_page(self, new_message_links: list[str]) -> bool: """ 判断是否还需要继续请求 Audiences 下一页未读消息列表。 """ @@ -308,7 +318,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): and len(self.__seen_unread_message_links) >= self.message_unread @staticmethod - def __is_empty_message_content(head, date, content) -> bool: + def __is_empty_message_content(head: object, date: object, content: object) -> bool: """ 判断消息详情是否完全为空，避免把解析失败页包装成 None 通知。 """ @@ -321,7 +331,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): """ return f"messages.php?action=viewmailbox&box={box}&unread=yes" - def __build_next_unread_mailbox_page(self, has_unread: bool) -> str: + def __build_next_unread_mailbox_page(self, has_unread: bool) -> Optional[str]: """ 当前页存在未读消息时按 Audiences 的 page 参数规则生成下一页地址。 """ @@ -332,21 +342,21 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): self.__next_mail_page += 1 return f"{self._user_mail_unread_page}&page={next_page}" - def _parse_user_traffic_info(self, html_text): + def _parse_user_traffic_info(self, html_text: str) -> None: """ 解析用户流量信息 """ super()._parse_user_traffic_info(html_text) self.__parse_userbar_info(html_text) - def _parse_user_detail_info(self, html_text: str): + def _parse_user_detail_info(self, html_text: str) -> None: """ 解析用户额外信息 """ super()._parse_user_detail_info(html_text) self.__parse_userbar_info(html_text) - def __parse_userbar_info(self, html_text: str): + def __parse_userbar_info(self, html_text: str) -> None: """ 解析 Audiences 新版顶部用户栏，覆盖 NexusPHP 通用正则的误判。 """ @@ -369,7 +379,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): if html is not None: del html - def __parse_user_identity(self, user_node): + def __parse_user_identity(self, user_node: etree._Element) -> None: """ 从新版用户卡属性中提取用户 ID、用户名和等级。 """ @@ -387,9 +397,9 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): # 站点组名仅用于展示，保留通用等级以与其他 NexusPHP 站点一致。 user_level = re.sub(r"^\([^)]*\)\s*", "", user_level.strip()).strip() if user_level: - self.user_level = user_level # type: ignore[assignment] + self.user_level = user_level - def __parse_uploader_stats(self, stats_text: str): + def __parse_uploader_stats(self, stats_text: str) -> None: """ 解析 data-uploader-stats 中的结构化流量数据。 """ @@ -412,7 +422,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): value = str(item.get("value") or "").strip() self.__set_metric_value(label=label, tone=tone, value=value) - def __parse_compact_metric(self, html, metric: str, field: str): + def __parse_compact_metric(self, html: etree._Element, metric: str, field: str) -> None: """ 按 compact metric 的 class 读取新版用户栏中的单项数据。 """ @@ -428,7 +438,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): if values: self.__set_metric_value(field=field, value=values[-1].strip()) - def __set_metric_value(self, value: str, label: str = None, tone: str = None, field: str = None): + def __set_metric_value(self, value: str, label: Optional[str] = None, tone: Optional[str] = None, field: Optional[str] = None) -> None: """ 将 Audiences 用户栏指标写入通用用户数据字段。 """ @@ -450,7 +460,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): self.seeding = text_tools.parse_int(active_match.group(1)) self.leeching = text_tools.parse_int(active_match.group(2)) - def __parse_inbox_unread(self, message_link): + def __parse_inbox_unread(self, message_link: etree._Element) -> Optional[int]: """ 从 Audiences 收件箱入口提取未读数。 """ @@ -474,7 +484,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return None @staticmethod - def __extract_inbox_unread_pair(text: str): + def __extract_inbox_unread_pair(text: Optional[str]) -> Optional[int]: """ 从 Audiences 总数/未读数格式中提取未读数，例如 1749/172。 """ @@ -492,7 +502,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return None @staticmethod - def __extract_inbox_unread_badge(text: str): + def __extract_inbox_unread_badge(text: Optional[str]) -> Optional[int]: """ 从明确的未读角标中提取未读数，避免把普通收件箱总数误作未读。 """ @@ -509,7 +519,7 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return None @staticmethod - def __has_inbox_unread_marker(message_link) -> bool: + def __has_inbox_unread_marker(message_link: etree._Element) -> bool: """ 判断收件箱入口是否只有未读状态但没有可靠数量。 """ @@ -518,7 +528,8 @@ class NexusAudiencesSiteUserInfo(NexusPhpSiteUserInfo): return True return bool(message_link.xpath('.//*[contains(@class, "site-userbar__compact-tool-badge--unread")]')) - def _parse_seeding_pages(self): + def _parse_seeding_pages(self) -> None: + """按站点页面规则读取做种列表并保留明确的汇总数据。""" if not self._torrent_seeding_page: return self._torrent_seeding_headers = {"Referer": urljoin(self._base_url, self._user_detail_page)}