mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
205 lines
7.0 KiB
Python
205 lines
7.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
解析 HDDolby 站点的用户资料、流量和做种信息。
|
||
"""
|
||
|
||
import json
|
||
from typing import Optional, Tuple
|
||
|
||
from app.domain import site as site_rules
|
||
from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema
|
||
from app.modules.indexer.parser.nexus_php import NexusPhpSiteUserInfo
|
||
from app.runtime.log import logger
|
||
|
||
|
||
class HDDolbySiteUserInfo(SiteParserBase):
|
||
"""
|
||
解析 HDDolby 站点的用户资料、流量与消息。
|
||
"""
|
||
schema = SiteSchema.HDDolby
|
||
request_mode = "apikey"
|
||
|
||
# 用户级别字典
|
||
HDDolby_sysRoleList = {
|
||
"0": "Peasant",
|
||
"1": "User",
|
||
"2": "Power User",
|
||
"3": "Elite User",
|
||
"4": "Crazy User",
|
||
"5": "Insane User",
|
||
"6": "Veteran User",
|
||
"7": "Extreme User",
|
||
"8": "Ultimate User",
|
||
"9": "Nexus Master",
|
||
"10": "VIP",
|
||
"11": "Retiree",
|
||
"12": "Helper",
|
||
"13": "Seeder",
|
||
"14": "Transferrer",
|
||
"15": "Uploader",
|
||
"16": "Torrent Manager",
|
||
"17": "Forum Moderator",
|
||
"18": "Coder",
|
||
"19": "Moderator",
|
||
"20": "Administrator",
|
||
"21": "Sysop",
|
||
"22": "Staff Leader",
|
||
}
|
||
|
||
def _parse_site_page(self, html_text: SitePage) -> None:
|
||
"""
|
||
使用资源声明的 API 基础地址配置用户资料和做种接口。
|
||
|
||
未配置 api_url 时保留旧站的 api 子域，前端地址继续用于读取站点消息。
|
||
|
||
:param html_text: API 模式不使用首页内容
|
||
"""
|
||
api_url = self._api_url or f"https://api.{site_rules.extract_domain(self._base_url)}/api/v1"
|
||
self._base_url = f"{api_url.rstrip('/')}/"
|
||
self._user_traffic_page = None
|
||
self._user_detail_page = None
|
||
self._user_basic_page = "user/data"
|
||
self._user_basic_params = {}
|
||
self._user_basic_headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json, text/plain, */*"
|
||
}
|
||
self._sys_mail_unread_page = None
|
||
self._user_mail_unread_page = None
|
||
self._mail_unread_params = {}
|
||
self._torrent_seeding_page = "user/peers"
|
||
self._torrent_seeding_params = {}
|
||
self._torrent_seeding_headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json, text/plain, */*"
|
||
}
|
||
self._addition_headers = {
|
||
"x-api-key": self.apikey,
|
||
}
|
||
|
||
def _parse_logged_in(self, html_text: str) -> bool:
|
||
"""
|
||
API 认证结果由请求响应判定，此入口不检查页面登录标记。
|
||
|
||
:param html_text: 未使用的页面文本
|
||
:return: 始终返回 True
|
||
"""
|
||
return True
|
||
|
||
def _parse_user_base_info(self, html_text: SitePage) -> None:
|
||
"""
|
||
解析用户基本信息，这里把_parse_user_traffic_info和_parse_user_detail_info合并到这里。
|
||
"""
|
||
if not html_text:
|
||
return None
|
||
if isinstance(html_text, dict):
|
||
raise TypeError("the JSON object must be str, bytes or bytearray, not dict")
|
||
detail = json.loads(html_text)
|
||
if not detail or detail.get("status") != 0:
|
||
return
|
||
user_infos = detail.get("data")
|
||
"""
|
||
{
|
||
"id": "1",
|
||
"added": "2019-03-03 15:30:36",
|
||
"last_access": "2025-02-18 19:48:04",
|
||
"class": "22",
|
||
"uploaded": "852071699418375",
|
||
"downloaded": "1885536536176",
|
||
"seedbonus": "99774808.0",
|
||
"sebonus": "3739023.7",
|
||
"unread_messages": "0",
|
||
}
|
||
"""
|
||
if not user_infos:
|
||
return
|
||
user_info = user_infos[0]
|
||
self.userid = user_info.get("id")
|
||
self.username = user_info.get("username")
|
||
self.user_level = self.HDDolby_sysRoleList.get(user_info.get("class") or "1")
|
||
self.join_at = user_info.get("added")
|
||
self.upload = int(user_info.get("uploaded") or '0')
|
||
self.download = int(user_info.get("downloaded") or '0')
|
||
self.ratio = round(self.upload / self.download, 2) if self.download else 0
|
||
self.bonus = float(user_info.get("seedbonus") or "0")
|
||
self.message_unread = int(user_info.get("unread_messages") or '0')
|
||
|
||
def _parse_user_traffic_info(self, html_text: str) -> None:
|
||
"""
|
||
解析用户流量信息。
|
||
"""
|
||
pass
|
||
|
||
def _parse_user_detail_info(self, html_text: str) -> None:
|
||
"""
|
||
解析用户详细信息。
|
||
"""
|
||
pass
|
||
|
||
def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: Optional[bool] = False) -> Optional[str]:
|
||
"""
|
||
解析用户做种信息。
|
||
"""
|
||
if not html_text:
|
||
return None
|
||
seeding_info = json.loads(html_text)
|
||
if not seeding_info or seeding_info.get("status") != 0:
|
||
return None
|
||
torrents = seeding_info.get("data", [])
|
||
page_seeding_size = 0
|
||
page_seeding_info = []
|
||
for info in torrents:
|
||
size = info.get("size")
|
||
seeder = info.get("seeders") or 1
|
||
page_seeding_size += size
|
||
page_seeding_info.append([seeder, size])
|
||
self.seeding += len(torrents)
|
||
self.seeding_size += page_seeding_size
|
||
self.seeding_info.extend(page_seeding_info)
|
||
|
||
return None
|
||
|
||
def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]:
|
||
"""
|
||
解析未读消息链接。
|
||
HDDolby 使用 API 模式，消息正文通过 _pase_unread_msgs 走网页接口读取。
|
||
"""
|
||
return None
|
||
|
||
def _parse_message_content(self, html_text: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
|
||
"""
|
||
解析消息内容。
|
||
HDDolby 使用 API 模式，消息正文通过 _pase_unread_msgs 走网页接口读取。
|
||
"""
|
||
return None, None, None
|
||
|
||
def _pase_unread_msgs(self) -> None:
|
||
"""
|
||
HDDolby API 仅返回未读消息数量，正文需通过 NexusPHP 网页接口读取。
|
||
"""
|
||
if not self._site_cookie:
|
||
if self.message_unread:
|
||
logger.warn(
|
||
f"{self._site_name} 未配置 Cookie，无法读取站点消息正文"
|
||
f"（API 仅提供未读数量：{self.message_unread}）"
|
||
)
|
||
return
|
||
|
||
nexus = NexusPhpSiteUserInfo(
|
||
site_name=self._site_name,
|
||
url=self._site_url,
|
||
site_cookie=self._site_cookie,
|
||
apikey=self.apikey,
|
||
token=self.token,
|
||
request_utils=self._request_utils,
|
||
ua=self._ua,
|
||
emulate=self._emulate,
|
||
proxy=self._proxy,
|
||
)
|
||
nexus.message_unread = self.message_unread
|
||
nexus.message_read_force = self.message_unread > 0
|
||
nexus._pase_unread_msgs()
|
||
self.message_unread_contents = nexus.message_unread_contents.copy()
|
||
if self.message_unread_contents and not self.message_unread:
|
||
self.message_unread = len(self.message_unread_contents)
|