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

# -*- coding: utf-8 -*-
"""
解析 Unit3D 站点的用户资料、流量和做种信息。
"""

import re
from typing import Optional
from urllib.parse import urlsplit

from lxml import etree

from app.foundation import size as size_tools
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


class Unit3dSiteUserInfo(SiteParserBase):
    """
    Unit3D 架构站点的用户信息解析器。
    """

    schema = SiteSchema.Unit3d

    def _parse_user_base_info(self, html_text: SitePage) -> None:
        """
        解析用户名并将 Unit3D 的用户名作为用户标识供后续请求使用。
        """
        html_text = self._prepare_html_text(html_text)
        html = etree.HTML(html_text)
        try:
            tmps = html.xpath('//a[contains(@href, "/users/") and contains(@href, "settings")]/@href')
            if tmps:
                user_name_match = re.search(r"/users/(.+)/settings", tmps[0])
                if user_name_match and user_name_match.group().strip():
                    self.username = user_name_match.group(1)
                    self.userid = self.username
                    self._torrent_seeding_page = f"/users/{self.username}/active?perPage=100&client=&seeding=include"
                    self._user_detail_page = f"/users/{self.username}"

            tmps = html.xpath('//a[contains(@href, "bonus/earnings")]')
            if not tmps and self.username:
                tmps = html.xpath('//a[contains(@href, "/users/") and contains(@href, "/bonus/")]')
                tmps = [link for link in tmps
                        if urlsplit(link.get("href")).path.startswith(f"/users/{self.username}/bonus/")]
            if tmps:
                bonus_text = tmps[0].xpath("string(.)")
                bonus_match = re.search(r"([\d,.]+)", bonus_text)
                if bonus_match and bonus_match.group(1).strip():
                    self.bonus = text_tools.parse_float(bonus_match.group(1))
        finally:
            if html is not None:
                del html

    def _parse_site_page(self, html_text: SitePage) -> None:
        """
        Unit3D 不需要从站点首页解析额外的站点级数据。
        """
        pass

    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 None

            # 用户等级
            user_levels_text = html.xpath('//div[contains(@class, "content")]//span[contains(@class, "badge-user")]/text()')
            if user_levels_text:
                self.user_level = user_levels_text[0].strip()

            # 加入日期
            join_at_text = html.xpath('//div[contains(@class, "content")]//h4[contains(text(), "注册日期") '
                                      'or contains(text(), "註冊日期") '
                                      'or contains(text(), "Registration date")]/text()')
            if join_at_text:
                self.join_at = time_tools.normalize_datetime(
                    join_at_text[0].replace('注册日期', '').replace('註冊日期', '').replace('Registration date', ''))
        finally:
            if html is not None:
                del html

    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: 下页地址
        """
        html = etree.HTML(html_text)
        try:
            if not DomUtils.has_child_elements(html):
                return None

            size_col = 9
            seeders_col = 2
            # 搜索size列
            if html.xpath('//thead//th[contains(@class,"size")]'):
                size_col = len(html.xpath('//thead//th[contains(@class,"size")][1]/preceding-sibling::th')) + 1
            # 搜索seeders列
            if html.xpath('//thead//th[contains(@class,"seeders")]'):
                seeders_col = len(html.xpath('//thead//th[contains(@class,"seeders")]/preceding-sibling::th')) + 1

            page_seeding = 0
            page_seeding_size = 0
            page_seeding_info = []
            seeding_sizes = html.xpath(f'//tr[position()]/td[{size_col}]')
            seeding_seeders = html.xpath(f'//tr[position()]/td[{seeders_col}]')
            if seeding_sizes and seeding_seeders:
                page_seeding = len(seeding_sizes)

                for i in range(0, len(seeding_sizes)):
                    size = size_tools.parse_size(seeding_sizes[i].xpath("string(.)").strip())
                    seeders = text_tools.parse_int(seeding_seeders[i].xpath("string(.)").strip())

                    page_seeding_size += size
                    page_seeding_info.append([seeders, size])

            self.seeding += page_seeding
            self.seeding_size += page_seeding_size
            self.seeding_info.extend(page_seeding_info)

            # 是否存在下页数据
            next_page = None
            next_pages = html.xpath('//ul[@class="pagination"]/li[contains(@class,"active")]/following-sibling::li')
            if next_pages and len(next_pages) > 1:
                page_num = next_pages[0].xpath("string(.)").strip()
                if page_num.isdigit():
                    next_page = f"{self._torrent_seeding_page}&page={page_num}"
        finally:
            if html is not None:
                del html

        return next_page

    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 = size_tools.parse_size(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 = size_tools.parse_size(download_match.group(1).strip()) if download_match else 0
        ratio_match = re.search(r"分享率[:：_<>/a-zA-Z-=\"'\s#;]+([\d,.\s]+)", html_text)
        self.ratio = text_tools.parse_float(ratio_match.group(1)) if (
            ratio_match and ratio_match.group(1).strip()) else 0.0

    def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]:
        """
        Unit3D 没有兼容的未读消息列表解析。
        """
        return None

    def _parse_message_content(self, html_text: str) -> tuple[Optional[str], Optional[str], Optional[str]]:
        """
        Unit3D 没有兼容的未读消息正文解析。
        """
        return None, None, None