d0zingcat/MoviePilot-Plugins
1
0
Fork 0
mirror of https://github.com/d0zingcat/MoviePilot-Plugins.git synced 2026-05-21 07:26:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

add YemaPT

Browse Source
This commit is contained in:
jxxghp
2024-05-20 18:22:52 +08:00
parent c3460404e4
commit f89dbcda52
9 changed files with 205 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
package.json
View File

@@ -3,11 +3,12 @@
"name": "站点自动签到",
"description": "自动模拟登录、签到站点。",
"labels": "站点",
"version": "2.2",
"version": "2.3",
"icon": "signin.png",
"author": "thsrite",
"level": 2,
"history": {
"v2.3": "优化模拟登录逻辑，支持YemaPT模拟登录",
"v2.2": "适配馒头最新变化，需要升级至v1.8.5+版本且维护好Authorization",
"v2.1": "增强API安全性",
"v2.0": "站点签到时更新站点使用统计信息，需要主程序升级至v1.8.3+版本",
@@ -27,11 +28,12 @@
"name": "站点数据统计",
"description": "自动统计和展示站点数据。",
"labels": "站点",
"version": "3.4",
"version": "3.5",
"icon": "statistic.png",
"author": "lightolly",
"level": 2,
"history": {
"v3.5": "站点数据统计支持YemaPT",
"v3.4": "修复馒头站点数据统计",
"v3.3": "支持选择仪表板组件规格",
"v3.2": "支持在仪表板中显示站点统计信息，需要主程序升级至v1.8.7+版本",

55
plugins/autosignin/__init__.py
View File

@@ -38,7 +38,7 @@ class AutoSignIn(_PluginBase):
# 插件图标
plugin_icon = "signin.png"
# 插件版本
plugin_version = "2.2"
plugin_version = "2.3"
# 插件作者
plugin_author = "thsrite"
# 作者主页
@@ -76,14 +76,6 @@ class AutoSignIn(_PluginBase):
_end_time: int = None
_auto_cf: int = 0
def __init__(self):
super().__init__()
# 特殊模拟登录站点
self._special_login_sites = {
"m-team.io": self.__mteam_login,
"m-team.cc": self.__mteam_login,
}
def init_plugin(self, config: dict = None):
self.sites = SitesHelper()
self.siteoper = SiteOper()
@@ -978,34 +970,25 @@ class AutoSignIn(_PluginBase):
"""
模拟登录一个站点
"""
domain = StringUtils.get_url_domain(site_info.get("url"))
if domain in self._special_login_sites:
return site_info.get("name"), self._special_login_sites[domain](site_info)
return site_info.get("name"), self.__login_base(site_info)
@staticmethod
def __mteam_login(site: CommentedMap) -> str:
"""
mteam登录
"""
headers = {
"Content-Type": "application/json",
"User-Agent": site.get("ua"),
"Accept": "application/json, text/plain, */*",
"Authorization": site.get("token")
}
# 更新最后访问时间
res = RequestUtils(headers=headers,
timeout=60,
proxies=settings.PROXY if site.get("proxy") else None,
referer=f"{site.get('url')}index"
).post_res(url=urljoin(site.get('url'), "api/member/updateLastBrowse"))
if res:
return "模拟登录成功"
elif res is not None:
return f"模拟登录失败，状态码：{res.status_code}"
site_module = self.__build_class(site_info.get("url"))
# 开始记时
start_time = datetime.now()
if site_module and hasattr(site_module, "login"):
try:
state, message = site_module().login(site_info)
except Exception as e:
traceback.print_exc()
state, message = False, f"模拟登录失败：{str(e)}"
else:
return "模拟登录失败，无法打开网站"
state, message = self.__login_base(site_info)
# 统计
seconds = (datetime.now() - start_time).seconds
domain = StringUtils.get_url_domain(site_info.get('url'))
if state:
self.sitestatistic.success(domain=domain, seconds=seconds)
else:
self.sitestatistic.fail(domain)
return site_info.get("name"), message
@staticmethod
def __login_base(site_info: CommentedMap) -> str:

4
plugins/autosignin/sites/__init__.py
View File

@@ -28,7 +28,9 @@ class _ISiteSigninHandler(metaclass=ABCMeta):
:param url: 站点Url
:return: 是否匹配，如匹配则会调用该类的signin方法
"""
return True if StringUtils.url_equal(url, self.site_url) else False
if StringUtils.url_equal(url, self.site_url):
return True
return False
@abstractmethod
def signin(self, site_info: CommentedMap) -> Tuple[bool, str]:

63
plugins/autosignin/sites/mteam.py
View File

@@ -4,7 +4,6 @@ from urllib.parse import urljoin
from ruamel.yaml import CommentedMap
from app.core.config import settings
from app.log import logger
from app.plugins.autosignin.sites import _ISiteSigninHandler
from app.utils.http import RequestUtils
@@ -31,41 +30,29 @@ class MTorrent(_ISiteSigninHandler):
:param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息
:return: 签到结果信息
"""
site = site_info.get("name")
site_cookie = site_info.get("cookie")
ua = site_info.get("ua")
proxy = site_info.get("proxy")
render = site_info.get("render")
url = site_info.get("url")
token = site_info.get("token")
if render:
# 获取页面html
html_text = self.get_page_source(url=url,
cookie=site_cookie,
ua=ua,
proxy=proxy,
render=render)
if not html_text:
logger.error(f"{site} 模拟登录失败，请检查站点连通性")
return False, '模拟登录失败，请检查站点连通性'
if "登 錄" in html_text:
logger.error(f"{site} 模拟登录失败，Cookie已失效")
return False, '模拟登录失败，Cookie已失效'
return True, '模拟登录成功'
headers = {
"Content-Type": "application/json",
"User-Agent": site_info.get("ua"),
"Accept": "application/json, text/plain, */*",
"Authorization": site_info.get("token")
}
# 更新最后访问时间
res = RequestUtils(headers=headers,
timeout=60,
proxies=settings.PROXY if site_info.get("proxy") else None,
referer=f"{site_info.get('url')}index"
).post_res(url=urljoin(site_info.get('url'), "api/member/updateLastBrowse"))
if res:
return True, "模拟登录成功"
elif res is not None:
return False, f"模拟登录失败，状态码：{res.status_code}"
else:
headers = {
"Content-Type": "application/json",
"User-Agent": ua,
"Accept": "application/json, text/plain, */*",
"Authorization": token
}
res = RequestUtils(headers=headers,
timeout=60,
proxies=settings.PROXY if proxy else None
).post_res(url=urljoin(url, "api/member/updateLastBrowse"))
if res:
logger.info(f'【{site}】模拟登录成功')
return True, f'模拟登录成功'
else:
logger.error(f"{site} 模拟登录失败，{res.status_code if res else '网络错误'}")
return False, f"模拟登录失败，{res.status_code if res else '网络错误'}"
return False, "模拟登录失败，无法打开网站"
def login(self, site_info: CommentedMap) -> Tuple[bool, str]:
"""
执行登录操作
:param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息
:return: 登录结果信息
"""
return self.signin(site_info)

58
plugins/autosignin/sites/yema.py Normal file
View File

@@ -0,0 +1,58 @@
from typing import Tuple
from urllib.parse import urljoin
from ruamel.yaml import CommentedMap
from app.core.config import settings
from app.plugins.autosignin.sites import _ISiteSigninHandler
from app.utils.http import RequestUtils
class MTorrent(_ISiteSigninHandler):
"""
m-team签到
"""
# 匹配的站点Url，每一个实现类都需要设置为自己的站点Url
site_url = "yemapt.org"
@classmethod
def match(cls, url: str) -> bool:
"""
根据站点Url判断是否匹配当前站点签到类，大部分情况使用默认实现即可
:param url: 站点Url
:return: 是否匹配，如匹配则会调用该类的signin方法
"""
return True if cls.site_url in url else False
def signin(self, site_info: CommentedMap) -> Tuple[bool, str]:
"""
执行签到操作，馒头实际没有签到，非仿真模式下需要更新访问时间
:param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息
:return: 签到结果信息
"""
headers = {
"Content-Type": "application/json",
"User-Agent": site_info.get("ua"),
"Accept": "application/json, text/plain, */*",
}
# 更新最后访问时间
res = RequestUtils(headers=headers,
timeout=15,
cookies=site_info.get("cookie"),
proxies=settings.PROXY if site_info.get("proxy") else None,
referer=site_info.get('url')
).post_res(url=urljoin(site_info.get('url'), "api/user/profile"))
if res and res.json().get("success"):
return True, "模拟登录成功"
elif res is not None:
return False, f"模拟登录失败，状态码：{res.status_code}"
else:
return False, "模拟登录失败，无法打开网站"
def login(self, site_info: CommentedMap) -> Tuple[bool, str]:
"""
执行登录操作
:param site_info: 站点信息，含有站点Url、站点Cookie、UA等信息
:return: 登录结果信息
"""
return self.signin(site_info)

2
plugins/sitestatistic/__init__.py
View File

@@ -43,7 +43,7 @@ class SiteStatistic(_PluginBase):
# 插件图标
plugin_icon = "statistic.png"
# 插件版本
plugin_version = "3.4"
plugin_version = "3.5"
# 插件作者
plugin_author = "lightolly"
# 作者主页

1
plugins/sitestatistic/siteuserinfo/__init__.py
View File

@@ -33,6 +33,7 @@ class SiteSchema(Enum):
FileList = "FileList"
TNode = "TNode"
MTorrent = "MTorrent"
Yema = "Yema"
class ISiteUserInfo(metaclass=ABCMeta):

3
plugins/sitestatistic/siteuserinfo/mtorrent.py
View File

@@ -2,12 +2,11 @@
import json
from typing import Optional, Tuple
from urllib.parse import urljoin
from lxml import etree
from app.log import logger
from app.db.systemconfig_oper import SystemConfigOper
from app.plugins.sitestatistic.siteuserinfo import ISiteUserInfo, SITE_BASE_ORDER, SiteSchema
from app.utils.string import StringUtils
class MTorrentSiteUserInfo(ISiteUserInfo):

93
plugins/sitestatistic/siteuserinfo/yema.py Normal file
View File

@@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
import json
from typing import Optional, Tuple
from app.plugins.sitestatistic.siteuserinfo import ISiteUserInfo, SITE_BASE_ORDER, SiteSchema
from app.utils.string import StringUtils
class TYemaSiteUserInfo(ISiteUserInfo):
schema = SiteSchema.Yema
order = SITE_BASE_ORDER + 60
@classmethod
def match(cls, html_text: str) -> bool:
return '<title>YemaPT</title>' in html_text
def _parse_site_page(self, html_text: str):
"""
获取站点页面地址
"""
self._user_traffic_page = None
self._user_detail_page = None
self._user_basic_page = "api/consumer/fetchSelfDetail"
self._user_basic_params = {}
self._sys_mail_unread_page = None
self._user_mail_unread_page = None
self._mail_unread_params = {}
self._torrent_seeding_page = None
self._torrent_seeding_headers = {}
self._addition_headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/plain, */*",
}
def _parse_logged_in(self, html_text):
"""
判断是否登录成功, 通过判断是否存在用户信息
暂时跳过检测，待后续优化
:param html_text:
:return:
"""
return True
def _parse_user_base_info(self, html_text: str):
"""
解析用户基本信息，这里把_parse_user_traffic_info和_parse_user_detail_info合并到这里
"""
if not html_text:
return None
detail = json.loads(html_text)
if not detail or not detail.get("success"):
return
user_info = detail.get("data", {})
self.userid = user_info.get("id")
self.username = user_info.get("name")
self.user_level = user_info.get("level")
self.join_at = StringUtils.unify_datetime_str(user_info.get("registerTime"))
self.upload = user_info.get('uploadSize')
self.download = user_info.get('downloadSize')
self.ratio = round(self.upload / (self.download or 1), 2)
self.bonus = user_info.get("bonus")
self.message_unread = 0
def _parse_user_traffic_info(self, html_text: str):
"""
解析用户流量信息
"""
pass
def _parse_user_detail_info(self, html_text: str):
"""
解析用户详细信息
"""
pass
def _parse_user_torrent_seeding_info(self, html_text: str, multi_page: bool = False) -> Optional[str]:
"""
解析用户做种信息
"""
pass
def _parse_message_unread_links(self, html_text: str, msg_links: list) -> Optional[str]:
"""
解析未读消息链接，这里直接读出详情
"""
pass
def _parse_message_content(self, html_text) -> Tuple[Optional[str], Optional[str], Optional[str]]:
"""
解析消息内容
"""
pass