mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
706 lines
33 KiB
Python
706 lines
33 KiB
Python
# -*- 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
|
||
# 数值嵌在链接内部的元素中（憨憨：<a href="mybonus.php"><div>13,853,582</div></a>）
|
||
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
|