MoviePilot/app/modules/indexer/parser/bitpt.py

#
# 极速之星 https://bitpt.cn/
# 作者： ThedoRap
# 日期： 2025-10-02
#
# -*- coding: utf-8 -*-
"""
解析极速之星站点的用户资料、流量和做种信息。
"""

import re
from typing import Optional, Tuple
from urllib.parse import urlencode, urljoin

from bs4 import BeautifulSoup

from app.foundation import size as size_tools
from app.foundation import temporal as time_tools
from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema


class BitptSiteUserInfo(SiteParserBase):
    """
    解析 Bitpt 站点的用户资料、流量与消息。
    """
    schema = SiteSchema.Bitpt

    def _parse_site_page(self, html_text: SitePage) -> None:
        """
        识别并配置站点数据页面与请求参数。
        """
        self._user_basic_page = "userdetails.php?uid={uid}"
        self._user_detail_page = None
        self._user_basic_params = {}
        self._user_traffic_page = None
        self._sys_mail_unread_page = None
        self._user_mail_unread_page = None
        self._mail_unread_params = {}
        self._torrent_seeding_base = "browse.php"
        self._torrent_seeding_params = {"t": "myseed", "st": "2", "d": "desc"}
        self._torrent_seeding_headers = {}
        self._addition_headers = {}

    def _parse_logged_in(self, html_text: str) -> bool:
        """
        根据当前页面的用户入口判断登录状态。
        """
        soup = BeautifulSoup(html_text, 'html.parser')
        return bool(soup.find(id='userinfotop'))

    def _parse_user_base_info(self, html_text: SitePage) -> None:
        """
        解析站点用户身份和页面提供的基础统计。
        """
        if not html_text:
            return None
        soup = BeautifulSoup(html_text, 'html.parser')
        table = soup.find('table', class_='frmtable')
        if not table:
            return

        rows = table.find_all('tr')
        info_dict: dict[str, str] = {}
        for row in rows:
            cells = row.find_all('td')
            if len(cells) == 2:
                key = cells[0].text.strip()
                value = cells[1].text.strip()
                info_dict[key] = value

        self.userid = info_dict.get('UID')
        self.username = info_dict['用户名'].split('\xa0')[0] if '用户名' in info_dict else None
        self.user_level = info_dict['用户级别'] if '用户级别' in info_dict else None
        self.join_at = time_tools.normalize_datetime(info_dict['注册时间']) if '注册时间' in info_dict else None

        self.upload = size_tools.parse_size(info_dict['上传流量']) if '上传流量' in info_dict else 0
        self.download = size_tools.parse_size(info_dict['下载流量']) if '下载流量' in info_dict else 0
        self.ratio = float(info_dict['共享率']) if '共享率' in info_dict else 0
        bonus_str = info_dict.get('星辰', '')
        bonus_match = re.search(r'累计([\d\.]+)', bonus_str)
        self.bonus = float(bonus_match.group(1)) if bonus_match else 0
        self.message_unread = 0

        if hasattr(self, '_torrent_seeding_base') and self._torrent_seeding_base:
            self.seeding = 0
            self.seeding_size = 0
        else:
            seeding_info = soup.find('div', style="margin:0 auto;width:90%;font-size:14px;margin-top:10px;margin-bottom:10px;text-align:center;")
            if seeding_info:
                seeding_link = seeding_info.find_all('a')[1].text if len(seeding_info.find_all('a')) > 1 else ''
                match = re.search(r'当前上传的种子\((\d+)个, 共([\d\.]+ [KMGT]B)\)', seeding_link)
                if match:
                    self.seeding = int(match.group(1))
                    self.seeding_size = size_tools.parse_size(match.group(2))
                else:
                    self.seeding = 0
                    self.seeding_size = 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_page_info(self, html_text: str) -> Tuple[int, int]:
        """
        统计当前页做种条目数及其真实容量。
        """
        if not html_text:
            return 0, 0
        soup = BeautifulSoup(html_text, 'html.parser')
        torrent_table = soup.find('table', class_='torrenttable')
        if not torrent_table:
            return 0, 0
        rows = torrent_table.find_all('tr')
        if len(rows) <= 1:
            return 0, 0
        torrents = [row for row in rows[1:] if 'btr' in row.get('class', [])]
        page_seeding = 0
        page_seeding_size = 0
        for torrent in torrents:
            size_td = torrent.find('td', class_='r')
            if size_td:
                size_a = size_td.find('a')
                size_text = size_a.text.strip() if size_a else size_td.text.strip()
                if size_text:
                    page_seeding += 1
                    page_seeding_size += size_tools.parse_size(size_text)
        return page_seeding, page_seeding_size

    def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]:
        """
        该站点在其他解析入口处理此数据，本入口不追加操作。
        """
        pass

    def _parse_message_content(self, html_text: str) -> None:
        """
        该站点在其他解析入口处理此数据，本入口不追加操作。
        """
        pass

    def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: bool = False) -> Optional[str]:
        """
        该站点在其他解析入口处理此数据，本入口不追加操作。
        """
        pass

    def parse(self) -> None:
        """
        执行通用解析后补齐本站页面数据，保留原有对外字段格式。
        """
        super().parse()
        if self._index_html:
            soup = BeautifulSoup(self._index_html, 'html.parser')
            user_link = soup.find('a', href=re.compile(r'userdetails\.php\?uid=\d+'))
            if user_link:
                uid_match = re.search(r'uid=(\d+)', user_link['href'])
                if uid_match:
                    self.userid = uid_match.group(1)

        if self.userid and self._user_basic_page:
            basic_url = self._user_basic_page.format(uid=self.userid)
            basic_html = self._get_page_content(url=urljoin(self._base_url, basic_url))
            self._parse_user_base_info(basic_html)

        if hasattr(self, '_torrent_seeding_base') and self._torrent_seeding_base:
            seeding_base_url = urljoin(self._base_url, self._torrent_seeding_base)
            if self._torrent_seeding_params is None:
                raise AttributeError("'NoneType' object has no attribute 'copy'")
            params = self._torrent_seeding_params.copy()
            page_num = 1
            while True:
                params['p'] = page_num
                query_string = urlencode(params)
                full_url = f"{seeding_base_url}?{query_string}"
                seeding_html = self._get_page_content(url=full_url)
                page_seeding, page_seeding_size = self._parse_user_torrent_seeding_page_info(seeding_html)
                self.seeding += page_seeding
                self.seeding_size += page_seeding_size
                if page_seeding == 0:
                    break
                page_num += 1

        # 文本字段归一为空字符串；流量、数量和比例保留基类约定的数值类型。
        self.userid = str(self.userid or "")
        self.username = str(self.username or "")
        self.user_level = str(self.user_level or "")
        self.join_at = str(self.join_at or "")