# -*- coding: utf-8 -*- """ 解析 NexusPHP 站点的用户资料、流量和做种信息。 """ import re from typing import Any, Optional, Union from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from lxml import etree from app.foundation import temporal as time_tools from app.foundation import text as text_tools from app.foundation.dom import DomUtils from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema from app.runtime.log import logger class NexusPhpSiteUserInfo(SiteParserBase): """ NexusPHP 站点用户信息解析器。 """ schema = SiteSchema.NexusPhp _summary_seeding: Optional[int] = None _summary_seeding_size: Optional[int] = None def _parse_site_page(self, html_text: SitePage) -> None: """ 识别用户详情入口，并为定制站点选择已知的做种接口。 """ html_text = self._prepare_html_text(html_text) # NexusPHP v1.10+ 使用 UUID 标识用户，详情链接参数为 uuid user_detail = re.search(r"userdetails\.php\?(?:id|uuid)=([0-9A-Za-z_-]+)", html_text) if user_detail and user_detail.group(1).strip(): self._user_detail_page = user_detail.group().strip().lstrip('/') self.userid = user_detail.group(1).strip() self._torrent_seeding_page = self._build_torrent_seeding_page() else: user_detail = re.search(r"(userdetails)", html_text) if user_detail and user_detail.group().strip(): self._user_detail_page = user_detail.group().strip().lstrip('/') self.userid = None self._torrent_seeding_page = None @staticmethod def _is_uuid(user_id: Union[str, int, None]) -> bool: """ 判断用户标识是否为 UUID 形式。 使用 UUID 的 NexusPHP 站点将做种列表接口参数改名为 useruuid。 """ return bool(re.fullmatch( r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", str(user_id or "").strip() )) def _build_torrent_seeding_page(self) -> str: """ 构造做种列表接口地址。 :return: 做种列表地址，无有效用户标识时返回空字符串 """ if not self._has_valid_userid(): return "" site_domain = self._site_domain.lower().removeprefix("www.") if site_domain in {"pterclub.net", "pterclub.com"}: # 猫站的列表入口要求完整的浏览器导航请求头，仅作用于该站用户数据。 headers: dict[str, Optional[str]] = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "same-origin", } self._torrent_seeding_headers = headers self._user_detail_headers = dict(headers) return f"getusertorrentlist.php?userid={self.userid}&type=seeding" if site_domain == "keepfrds.com": return f"torrents.php?option-torrents=3&userid={self.userid}" if site_domain == "pt.sjtu.edu.cn": return f"viewtorrentsdetail.php?id={self.userid}" user_id_param = "useruuid" if self._is_uuid(self.userid) else "userid" return f"getusertorrentlistajax.php?{user_id_param}={self.userid}&type=seeding" def _parse_message_unread(self, html_text: str) -> None: """ 解析未读短消息数量。 :param html_text: 当前页面或 API 响应文本 """ html = etree.HTML(html_text) try: if not DomUtils.has_child_elements(html): return message_labels = html.xpath('//a[@href="messages.php"]/..') message_labels.extend(html.xpath('//a[contains(@href, "messages.php")]/..')) if message_labels: message_text = message_labels[0].xpath("string(.)") logger.debug(f"{self._site_name} 消息原始信息 {message_text}") message_unread_match = re.findall(r"[^Date](信息箱\s*|\((?![^)]*:)|你有\xa0)(\d+)", message_text) if message_unread_match and len(message_unread_match[-1]) == 2: self.message_unread = text_tools.parse_int(message_unread_match[-1][1]) elif message_text.isdigit(): self.message_unread = text_tools.parse_int(message_text) finally: if html is not None: del html def _parse_user_base_info(self, html_text: SitePage) -> None: """ 解析用户基本信息。 """ if isinstance(html_text, dict): raise TypeError("expected string or bytes-like object, got 'dict'") # 合并解析，减少额外请求调用 self._parse_user_traffic_info(html_text) self._user_traffic_page = None self._parse_message_unread(html_text) html = etree.HTML(html_text) try: if not DomUtils.has_child_elements(html): return user_level = self._parse_user_level_marker(html.xpath("string(.)"), self.userid) if user_level: self.user_level = user_level ret = html.xpath(f'//a[contains(@href, "userdetails") and contains(@href, "{self.userid}")]//b//text()') if ret: self.username = str(ret[0]) return ret = html.xpath(f'//a[contains(@href, "userdetails") and contains(@href, "{self.userid}")]//text()') if ret: self.username = str(ret[0]) ret = html.xpath('//a[contains(@href, "userdetails")]//strong//text()') finally: if html is not None: del html if ret: self.username = str(ret[0]) return def _parse_user_traffic_info(self, html_text: str) -> None: """ 解析用户流量信息。 """ html_text = self._prepare_html_text(html_text) upload_match = re.search(r"[^总]上[传傳]量?[:：_<>/a-zA-Z-=\"'\s#;]+([\d,.\s]+[KMGTPI]*B)", html_text, re.IGNORECASE) self.upload = self.num_filesize(upload_match.group(1).strip()) if upload_match else 0 download_match = re.search(r"[^总子影力]下[载載]量?[:：_<>/a-zA-Z-=\"'\s#;]+([\d,.\s]+[KMGTPI]*B)", html_text, re.IGNORECASE) self.download = self.num_filesize(download_match.group(1).strip()) if download_match else 0 ratio_match = re.search(r"分享率[:：_<>/a-zA-Z-=\"'\s#;]+([\d,.\s]+)", html_text) # 计算分享率 calc_ratio = 0.0 if self.download <= 0.0 else round(self.upload / self.download, 3) # 优先使用页面上的分享率 self.ratio = text_tools.parse_float(ratio_match.group(1)) if ( ratio_match and ratio_match.group(1).strip()) else calc_ratio leeching_match = re.search(r"(Torrents leeching|下载中)[\u4E00-\u9FA5\D\s]+(\d+)[\s\S]+<", html_text) self.leeching = text_tools.parse_int(leeching_match.group(2)) if leeching_match and leeching_match.group( 2).strip() else 0 html = etree.HTML(html_text) try: has_ucoin, self.bonus = self._parse_ucoin(html) if has_ucoin: return bonus_links = html.xpath('//a[contains(@href,"mybonus")]') if html is not None else [] for bonus_link in bonus_links: # 魔力值直接写在链接文本中 for bonus_text in bonus_link.xpath('text()'): bonus_match = re.search(r"([\d,.]+)", str(bonus_text).strip()) if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1)) return # 数值嵌在链接内部的元素中（憨憨：
13,853,582
） bonus_text = bonus_link.xpath("string(.)").strip() bonus_match = re.fullmatch(r"[\s\[(（【]*([\d][\d,.]*)[\s\])）】]*", bonus_text) if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1)) return # PTT-NP 等站点的链接文本只有“使用&说明”，数值紧跟在链接之后 bonus_match = re.search(r"[：:]\s*([\d,.]+)", bonus_link.tail or "") if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1)) return for bonus_link in bonus_links: # 兜底解析链接附近容器中的魔力值，容器必须临近且文本紧凑， # 避免把公告、新闻等大段文本里的无关数字当作魔力值 bonus_containers = bonus_link.xpath('ancestor::*[position() <= 4][contains(., "魔力值")]') for bonus_container in bonus_containers: bonus_text = bonus_container.xpath("string(.)") if len(bonus_text) > 200: continue bonus_match = re.search(r"魔力值.*?[：:]\s*([\d,.]+)", bonus_text, flags=re.S) if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1)) return bonus_match = re.search(r"mybonus.[\[\]:：<>/a-zA-Z_\-=\"'\s#;.(使用魔力值豆]+\s*([\d,.]+)[<()&\s]", html_text) try: if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1)) return bonus_match = re.search(r"(?:魔力值|\])[\[\]:：<>/a-zA-Z_\-=\"'\s#;]+\s*([\d,.]+|\"[\d,.]+\")[<>()&\s]", html_text, flags=re.S) if bonus_match and bonus_match.group(1).strip(): self.bonus = text_tools.parse_float(bonus_match.group(1).strip('"')) except Exception as err: logger.error(f"{self._site_name} 解析魔力值出错, 错误信息: {str(err)}") finally: if html is not None: del html @staticmethod def _parse_ucoin(html: etree._Element) -> tuple[bool, float]: """ 解析站点金币、银币和铜币，并统一换算为铜币。 :param html: 包含货币标记的页面节点 :return: 是否取得非零货币值及换算结果；没有有效值时返回 False 和 0.0 """ if DomUtils.has_child_elements(html): gold, silver, copper = None, None, None golds = html.xpath('//span[@class = "ucoin-symbol ucoin-gold"]//text()') if golds: gold = text_tools.parse_float(str(golds[-1])) silvers = html.xpath('//span[@class = "ucoin-symbol ucoin-silver"]//text()') if silvers: silver = text_tools.parse_float(str(silvers[-1])) coppers = html.xpath('//span[@class = "ucoin-symbol ucoin-copper"]//text()') if coppers: copper = text_tools.parse_float(str(coppers[-1])) if gold or silver or copper: gold = gold if gold else 0 silver = silver if silver else 0 copper = copper if copper else 0 return True, gold * 100 * 100 + silver * 100 + copper return False, 0.0 def _parse_seeding_pages(self) -> None: """ 优先使用完整做种列表统计，列表为空或解析不完整时回退到用户详情页统计。 """ profile_seeding = self.seeding profile_seeding_size = self.seeding_size profile_seeding_info = list(self.seeding_info) # 用户详情页可能只显示部分做种记录，先暂存，避免与完整列表相加造成重复。 self.seeding = 0 self.seeding_size = 0 self.seeding_info.clear() try: super()._parse_seeding_pages() except Exception: self.seeding = profile_seeding self.seeding_size = profile_seeding_size self.seeding_info = profile_seeding_info raise if (not self.seeding or not self.seeding_size) and (profile_seeding or profile_seeding_size): self.seeding = profile_seeding self.seeding_size = profile_seeding_size if profile_seeding_info: self.seeding_info = profile_seeding_info # 定制站点明细可能只返回部分记录，明确的汇总值优先，明细仍用于做种人数分布。 if self._summary_seeding is not None: self.seeding = self._summary_seeding if self._summary_seeding_size is not None: self.seeding_size = self._summary_seeding_size def _parse_seeding_summary(self, html: Any) -> None: """ 读取已知定制站点的明确汇总，不把缺失体积推算为真实值。 """ if self._site_domain.lower().removeprefix("www.") not in { "pterclub.net", "pterclub.com", "keepfrds.com", "pt.sjtu.edu.cn" }: return page_text = " ".join(html.xpath('//text()[not(ancestor::script) and not(ancestor::style)]')) count_match = re.search(r"当前做种\s*[:：]?\s*([\d,]+)", page_text) if not count_match and self._site_domain.lower().removeprefix("www.") in {"pterclub.net", "pterclub.com"}: count_match = re.search(r"正在做种的种子\s*[（(]\s*共\s*([\d,]+)\s*个", page_text) size_match = re.search(r"做种大小\s*[:：]?\s*([\d,.]+\s*[KMGTPI]i?B)", page_text, re.I) if count_match: self._summary_seeding = text_tools.parse_int(count_match.group(1).replace(",", "")) self.seeding = self._summary_seeding if size_match: self._summary_seeding_size = self.num_filesize(size_match.group(1)) self.seeding_size = self._summary_seeding_size def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: Optional[bool] = False) -> Optional[str]: """ 做种相关信息。 :param html_text: 当前页面或 API 响应文本 :param multi_page: 是否多页数据 :return: 下页地址 """ if not html_text: logger.warning(f"{self._site_name} 做种接口未返回内容，请检查站点接口和访问权限") return None html = etree.HTML(str(html_text).replace(r'\/', '/')) try: if not DomUtils.has_child_elements(html): return None self._parse_seeding_summary(html) if self._site_domain.lower().removeprefix("www.") == "pt.sjtu.edu.cn": if self._summary_seeding is None: logger.warning(f"{self._site_name} 做种汇总为空或无法解析，请检查站点接口和访问权限") return None # 首页存在扩展链接，使用扩展链接 seeding_url_text = html.xpath('//a[contains(@href,"torrents.php") ' 'and contains(@href,"seeding")]/@href') if multi_page is False and seeding_url_text and seeding_url_text[0].strip(): self._torrent_seeding_page = seeding_url_text[0].strip() return self._torrent_seeding_page size_col, seeders_col = self._seeding_columns(html) page_seeding = 0 page_seeding_size = 0 page_seeding_info = [] site_domain = self._site_domain.lower().removeprefix("www.") if site_domain in {"pterclub.net", "pterclub.com"}: # 猫站完整做种列表按 td[2]/td[4]/td[5] 展示种子、体积和做种人数。 pterclub_rows = html.xpath( '//*[@id="outer"]/table/tbody/tr | //*[@id="outer"]/table/tr' '|//table[contains(concat(" ", normalize-space(@class), " "), " torrents ")]/tbody/tr' '|//table[contains(concat(" ", normalize-space(@class), " "), " torrents ")]/tr' ) for row in pterclub_rows: columns = row.xpath('./td | ./th') if len(columns) < 5 or not columns[1].xpath('.//a'): continue size = self.num_filesize(columns[3].xpath("string(.)").strip()) if not size: continue seeders = text_tools.parse_int(columns[4].xpath("string(.)").strip()) page_seeding += 1 page_seeding_size += size page_seeding_info.append([seeders, size]) else: # 如果 table class="torrents"，则限制在种子表中，避免解析页面其他表格。 table_class = '//table[@class="torrents"]' if html.xpath('//table[@class="torrents"]') else '' seeding_sizes = html.xpath(f'{table_class}//tr[position()>1]/td[{size_col}]') seeding_seeders = html.xpath(f'{table_class}//tr[position()>1]/td[{seeders_col}]/b/a/text()') if not seeding_seeders: seeding_seeders = html.xpath(f'{table_class}//tr[position()>1]/td[{seeders_col}]//text()') if seeding_sizes and seeding_seeders: page_seeding = len(seeding_sizes) for i in range(0, len(seeding_sizes)): size = self.num_filesize(seeding_sizes[i].xpath("string(.)").strip()) seeders = text_tools.parse_int(seeding_seeders[i]) page_seeding_size += size page_seeding_info.append([seeders, size]) if not page_seeding and not multi_page and self._summary_seeding is None: logger.warning(f"{self._site_name} 做种列表为空或无法解析，请检查站点接口和访问权限") self.seeding += page_seeding self.seeding_size += page_seeding_size self.seeding_info.extend(page_seeding_info) next_page = self._seeding_next_page(html) finally: if html is not None: del html return next_page def _seeding_columns(self, html: Any) -> tuple[int, int]: """ 优先使用表头标记，朋友的无标记模板使用已确认的固定列位。 """ size_col = 3 seeders_col = 4 # 搜索size列 size_col_xpath = '//tr[position()=1]/' \ 'td[(img[@class="size"] and img[@alt="size"])' \ ' or (text() = "大小")' \ ' or (a/img[@class="size" and @alt="size"])]' if html.xpath(size_col_xpath): size_col = len(html.xpath(f'{size_col_xpath}/preceding-sibling::td')) + 1 # 搜索seeders列 seeders_col_xpath = '//tr[position()=1]/' \ 'td[(img[@class="seeders"] and img[@alt="seeders"])' \ ' or (text() = "在做种")' \ ' or (a/img[@class="seeders" and @alt="seeders"])]' if html.xpath(seeders_col_xpath): seeders_col = len(html.xpath(f'{seeders_col_xpath}/preceding-sibling::td')) + 1 if self._site_domain.lower().removeprefix("www.") == "keepfrds.com": # 朋友没有列标记，第四列为做种时长，第五、六列才是体积和做种人数。 size_col, seeders_col = 5, 6 return size_col, seeders_col def _seeding_next_page(self, html: Any) -> Optional[str]: """ 筛除详情链接并补齐下一页的用户标识。 """ # 是否存在下页数据 next_page = None next_page_text = html.xpath( '//a[contains(.//text(), "下一页") or contains(.//text(), "下一頁") or contains(.//text(), ">")]/@href') # 防止识别到详情页 while next_page_text: next_page = next_page_text.pop().strip() if not next_page.startswith('details.php'): break next_page = None # 补齐分页地址所需的用户标识，避免请求错误的做种列表。 if next_page: next_page = self._fixup_next_page_url(next_page, self.userid) return next_page def _fixup_next_page_url(self, next_page: str, userid: Union[str, int, None]) -> Optional[str]: """ 修正做种下一页地址，无法补齐用户 ID 时停止翻页。 :param next_page: 页面中解析出的下一页地址 :param userid: 当前站点用户 ID :return: 修正后的下一页地址，无法构造时返回 None """ parsed_url = urlsplit(next_page) query_params: dict[str, Union[str, int]] = dict(parse_qsl(parsed_url.query, keep_blank_values=True)) if any(query_params.get(key) for key in ("userid", "useruuid")): return next_page if not userid: return None query_params["useruuid" if self._is_uuid(userid) else "userid"] = userid query_params.setdefault("type", "seeding") return urlunsplit(parsed_url._replace(query=urlencode(query_params))) def _parse_user_detail_info(self, html_text: str) -> None: """ 解析用户额外信息，加入时间，等级。 :param html_text: 当前页面或 API 响应文本 """ html = etree.HTML(html_text) try: if not DomUtils.has_child_elements(html): return self._parse_seeding_summary(html) self._get_user_level(html) self._fixup_traffic_info(html) # 加入日期 join_at_text = html.xpath( '//tr/td[text()="加入日期" or text()="注册日期" or *[text()="加入日期"]]/following-sibling::td[1]//text()' '|//div/b[text()="加入日期"]/../text()' '|//*[@id="outer"]/table/tr/td/div/div[1]/div[2]/div[3]/span[1]/span/@title') if join_at_text: self.join_at = time_tools.normalize_datetime(join_at_text[0].split(' (')[0].strip()) # 做种体积 & 做种数 # seeding 页面获取不到的话，此处再获取一次 seeding_tables = html.xpath( '//tr/td[normalize-space(.)="当前上传" or normalize-space(.)="当前做种"]' '/following-sibling::td[1]//table' ) tmp_seeding = 0 tmp_seeding_size = 0 tmp_seeding_info = [] for seeding_table in seeding_tables: # 猫站使用“当前做种”作为面板行名；该表第 4、5 列分别是体积和做种人数。 seeding_rows = seeding_table.xpath('./tr | ./tbody/tr') for seeding_row in seeding_rows: columns = seeding_row.xpath('./td | ./th') if len(columns) < 5: continue size = self.num_filesize(columns[3].xpath("string(.)").strip()) # 无表头的做种表以有效体积列识别数据行，避免把表头计入数量。 if not size: continue seeders = text_tools.parse_int(columns[4].xpath("string(.)").strip()) tmp_seeding += 1 tmp_seeding_size += size tmp_seeding_info.append([seeders, size]) if not self.seeding_size: self.seeding_size = tmp_seeding_size if not self.seeding: self.seeding = tmp_seeding if not self.seeding_info: self.seeding_info = tmp_seeding_info seeding_sizes = html.xpath('//tr/td[text()="做种统计"]/following-sibling::td[1]//text()') if seeding_sizes: seeding_match = re.search(r"总做种数:\s+(\d+)", seeding_sizes[0], re.IGNORECASE) seeding_size_match = re.search(r"总做种体积:\s+([\d,.\s]+[KMGTPI]*B)", seeding_sizes[0], re.IGNORECASE) tmp_seeding = text_tools.parse_int(seeding_match.group(1)) if ( seeding_match and seeding_match.group(1)) else 0 tmp_seeding_size = self.num_filesize( seeding_size_match.group(1).strip()) if seeding_size_match else 0 if not self.seeding_size: self.seeding_size = tmp_seeding_size if not self.seeding: self.seeding = tmp_seeding self._fixup_torrent_seeding_page(html) finally: if html is not None: del html def _fixup_torrent_seeding_page(self, html: etree._Element) -> None: """ 从页面入口或脚本参数补齐做种列表地址和用户标识。 :param html: 用户详情页面节点，缺失时保留当前地址 """ # 单独的种子页面 seeding_url_text = html.xpath('//a[contains(@href,"getusertorrentlist.php") ' 'and contains(@href,"seeding")]/@href') if seeding_url_text: self._torrent_seeding_page = seeding_url_text[0].strip() # 从 JavaScript 调用中获取用户 ID。 seeding_url_text = html.xpath('//a[contains(@href, "javascript: getusertorrentlistajax") ' 'and contains(@href,"seeding")]/@href') csrf_text = html.xpath('//meta[@name="x-csrf"]/@content') if not self._torrent_seeding_page and seeding_url_text: user_js = re.search(r"javascript:\s*getusertorrentlistajax\(\s*'([^']+)'", seeding_url_text[0]) if user_js and user_js.group(1).strip(): self.userid = user_js.group(1).strip() self._torrent_seeding_page = self._build_torrent_seeding_page() elif seeding_url_text and csrf_text: if csrf_text[0].strip(): self._torrent_seeding_page \ = "ajax_getusertorrentlist.php" self._torrent_seeding_params = {'userid': self.userid, 'type': 'seeding', 'csrf': csrf_text[0].strip()} @staticmethod def _parse_user_level_marker(page_text: str, userid: Union[str, int, None]) -> Optional[str]: """ 解析 PTT-NP 首页中 UID 标记后的通用用户等级。 """ user_id_pattern = re.escape(str(userid)) if userid else r"\d+" user_level_match = re.search( rf"\[\s*UID\s*=\s*{user_id_pattern}\s*\]\s*\[\s*\([^\)\]]+\)\s*([^\]]+?)\s*\]", page_text, flags=re.IGNORECASE, ) return user_level_match.group(1).strip() if user_level_match else None def _get_user_level(self, html: etree._Element) -> None: """ 从用户详情页或首页中解析用户等级。 """ # 等级获取同一行等级数据，图片格式等级取 title 信息，否则取文本信息。 user_levels_text = html.xpath('//tr/td[text()="等級" or text()="等级" or *[text()="等级"]]/' 'following-sibling::td[1]/img[1]/@title') if user_levels_text: self.user_level = user_levels_text[0].strip() return user_levels_text = html.xpath('//tr/td[text()="等級" or text()="等级"]/' 'following-sibling::td[1 and not(img)]' '|//tr/td[text()="等級" or text()="等级"]/' 'following-sibling::td[1 and img[not(@title)]]') if user_levels_text: self.user_level = user_levels_text[0].xpath("string(.)").strip() return user_levels_text = html.xpath('//tr/td[text()="等級" or text()="等级"]/' 'following-sibling::td[1]') if user_levels_text: self.user_level = user_levels_text[0].xpath("string(.)").strip() return user_levels_text = html.xpath('//a[contains(@href, "userdetails")]/text()') if not self.user_level and user_levels_text: for user_level_text in user_levels_text: user_level_match = re.search(r"\[(.*)]", user_level_text) if user_level_match and user_level_match.group(1).strip(): self.user_level = user_level_match.group(1).strip() break user_level = self._parse_user_level_marker(html.xpath("string(.)"), self.userid) if user_level: self.user_level = user_level def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]: """ 解析未读消息链接及下一页入口。 """ html = etree.HTML(html_text) try: if not DomUtils.has_child_elements(html): return None message_links = html.xpath('//tr[not(./td/img[@alt="Read"])]/td/a[contains(@href, "viewmessage")]/@href') msg_links.extend(message_links) # 是否存在下页数据 next_page = None next_page_text = html.xpath('//a[contains(.//text(), "下一页") or contains(.//text(), "下一頁")]/@href') if next_page_text: next_page = next_page_text[-1].strip() finally: if html is not None: del html return next_page def _parse_message_content(self, html_text: str) -> tuple[Optional[str], Optional[str], Optional[str]]: """ 解析消息标题、时间和正文。 """ html = etree.HTML(html_text) try: if not DomUtils.has_child_elements(html): return None, None, None # 标题 message_head_text = None message_head = html.xpath('//h1/text()' '|//div[@class="layui-card-header"]/span[1]/text()') if message_head: message_head_text = message_head[-1].strip() # 消息时间 message_date_text = None message_date = html.xpath('//h1/following-sibling::table[.//tr/td[@class="colhead"]]//tr[2]/td[2]' '|//div[@class="layui-card-header"]/span[2]/span[2]') if message_date: message_date_text = message_date[0].xpath("string(.)").strip() # 消息内容 message_content_text = None message_content = html.xpath('//h1/following-sibling::table[.//tr/td[@class="colhead"]]//tr[3]/td' '|//div[contains(@class,"layui-card-body")]') if message_content: message_content_text = message_content[0].xpath("string(.)").strip() finally: if html is not None: del html return message_head_text, message_date_text, message_content_text def _fixup_traffic_info(self, html: etree._Element) -> None: """ 从详情表格补读为零的流量、分享率和魔力值，非零值保持不变。 """ if not self.upload: upload_text = self._traffic_table_value(html, r"上[传傳]量?") if upload_text is not None: self.upload = self.num_filesize(upload_text) if not self.download: download_text = self._traffic_table_value(html, r"下[载載]量?") if download_text is not None: self.download = self.num_filesize(download_text) if not self.ratio: ratio_text = self._traffic_table_value(html, "分享率") if ratio_text is not None: ratio = text_tools.parse_float(ratio_text) else: ratio = round(self.upload / self.download, 3) if self.download else 0.0 self.ratio = ratio # 用户详情表中的魔力值可补齐首页缺失的数据。 if not self.bonus: bonus_text = html.xpath('//tr/td[text()="魔力值" or text()="猫粮"]/following-sibling::td[1]/text()') if bonus_text: self.bonus = text_tools.parse_float(bonus_text[0].strip()) @staticmethod def _traffic_table_value(html: Any, field: str) -> Optional[str]: """ 按单元格字段名读取数值，支持同格或相邻列布局，排除“实际”流量。 """ for cell in html.xpath('//tr/td[not(.//td)]'): cell_text = " ".join(cell.xpath("string(.)").split()) field_match = re.fullmatch(rf"{field}\s*[:：]?\s*(.*)", cell_text) if not field_match: continue value = field_match.group(1).strip() if not value: siblings = cell.xpath('following-sibling::td[1]') value = " ".join(siblings[0].xpath("string(.)").split()) if siblings else "" value_match = re.fullmatch(r"([\d,.]+(?:\s*[KMGTPI]i?B)?)(?:\s*[（(].*)?", value, re.I) if value_match: return value_match.group(1) return None