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

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

import json
from operator import iadd
from typing import Optional
from urllib.parse import urljoin

from app.domain import site as site_rules
from app.modules.indexer.parser import SitePage, SiteParserBase, SiteSchema
from app.runtime.log import logger


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

    # 用户级别字典
    MTeam_sysRoleList = {
        "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": "Uploader",
        "13": "Moderator",
        "14": "Administrator",
        "15": "Sysop",
        "16": "Staff",
        "17": "Offer memberStaff",
        "18": "Bet memberStaff",
    }

    def _parse_site_page(self, html_text: SitePage) -> None:
        """
        获取站点页面地址。
        """
        # 更换api地址
        self._base_url = f"https://api.{site_rules.extract_domain(self._base_url)}"
        self._user_traffic_page = None
        self._user_detail_page = None
        self._user_basic_page = "api/member/profile"
        self._user_basic_params = {
            "uid": self.userid
        }
        self._sys_mail_unread_page = None
        self._user_mail_unread_page = "api/msg/search"
        self._mail_unread_params = {
            "keyword": "",
            "box": "-2",
            "type": "pageNumber",
            "pageSize": 100
        }
        self._torrent_seeding_page = "api/member/getUserTorrentList"
        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:
        """
        解析用户资料与流量，优先保留站点等级名及空分享率语义。

        :param html_text: MTorrent 用户资料接口的 JSON 文本
        """
        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("code") != "0":
            return
        user_info = detail.get("data", {})
        self.userid = user_info.get("id")
        self.username = user_info.get("username")
        self.user_level = user_info.get("roleName") or self.MTeam_sysRoleList.get(user_info.get("role") or "1")
        self.join_at = user_info.get("memberStatus", {}).get("createdDate")

        self.upload = int(user_info.get("memberCount", {}).get("uploaded") or '0')
        self.download = int(user_info.get("memberCount", {}).get("downloaded") or '0')
        # 空分享率表示站点未提供有效比值，不能与真实的 0 合并。
        self.ratio = user_info.get("memberCount", {}).get("shareRate")
        self.bonus = user_info.get("memberCount", {}).get("bonus") or 0
        self.message_read_force = True
        self._torrent_seeding_params = {
            "pageNumber": 1,
            "pageSize": 200,
            "type": "SEEDING",
            "userid": self.userid
        }

    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("code") != "0":
            return None
        seeding_data = seeding_info.get("data") or {}
        if not isinstance(seeding_data, dict):
            return None
        torrents = seeding_data.get("data") or []
        if not isinstance(torrents, list):
            return None
        page_seeding = 0
        page_seeding_size = 0
        page_seeding_info = []
        for info in torrents:
            if not isinstance(info, dict):
                continue
            torrent = info.get("torrent")
            if not isinstance(torrent, dict):
                continue
            size = int(torrent.get("size") or '0')
            seeders = int(torrent.get("source") or '0')
            page_seeding += 1
            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)

        # 空页或没有有效条目的页面不再推进，汇总可能滞后于实时做种明细。
        if not page_seeding:
            return None

        # 查询总做种数
        seeder_count = 0
        try:
            result = self._get_page_content(
                url=urljoin(self._base_url, "api/tracker/myPeerStatus"),
                params={"uid": self.userid},
            )
            if result:
                seeder_info = json.loads(result)
                seeder_count = int(seeder_info.get("data", {}).get("seeder") or 0)
        except Exception as e:
            logger.error(f"获取做种数失败: {str(e)}")
        if not seeder_count:
            return None
        if self.seeding >= seeder_count:
            return None
        # 还有下一页
        parameters = self._torrent_seeding_params
        if parameters is None:
            raise TypeError("'NoneType' object is not subscriptable")
        # 保留原有 += 运算及异常，不把动态请求参数强制转换为整数。
        parameters["pageNumber"] = iadd(parameters["pageNumber"], 1)
        return ""

    def _parse_message_unread_links(self, html_text: str, msg_links: list[str]) -> Optional[str]:
        """
        解析未读消息链接，这里直接读出详情。
        """
        if not html_text:
            return None
        messages_info = json.loads(html_text)
        if not messages_info or messages_info.get("code") != "0":
            return None
        messages = messages_info.get("data", {}).get("data", [])
        for message in messages:
            if not message.get("unread"):
                continue
            head = message.get("title")
            date = message.get("createdDate")
            content = message.get("context")
            if head and date and content:
                self.message_unread_contents.append((head, date, content))
                # 设置已读
                self._get_page_content(
                    url=urljoin(self._base_url, "api/msg/markRead"),
                    params={"msgId": message.get("id")}
                )
        # 是否存在下页数据
        return None

    def _parse_message_content(self, html_text: str) -> None:
        """
        解析消息内容。
        """
        pass