diff --git a/package.json b/package.json index 8fa30ac..434c879 100644 --- a/package.json +++ b/package.json @@ -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+版本", diff --git a/plugins/autosignin/__init__.py b/plugins/autosignin/__init__.py index 8fbaaf5..6701749 100644 --- a/plugins/autosignin/__init__.py +++ b/plugins/autosignin/__init__.py @@ -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: diff --git a/plugins/autosignin/sites/__init__.py b/plugins/autosignin/sites/__init__.py index 69b7d1e..b0e2ef2 100644 --- a/plugins/autosignin/sites/__init__.py +++ b/plugins/autosignin/sites/__init__.py @@ -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]: diff --git a/plugins/autosignin/sites/mteam.py b/plugins/autosignin/sites/mteam.py index 7c36176..eb13700 100644 --- a/plugins/autosignin/sites/mteam.py +++ b/plugins/autosignin/sites/mteam.py @@ -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) diff --git a/plugins/autosignin/sites/yema.py b/plugins/autosignin/sites/yema.py new file mode 100644 index 0000000..dc62cc9 --- /dev/null +++ b/plugins/autosignin/sites/yema.py @@ -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) diff --git a/plugins/sitestatistic/__init__.py b/plugins/sitestatistic/__init__.py index 99e95ee..d99a389 100644 --- a/plugins/sitestatistic/__init__.py +++ b/plugins/sitestatistic/__init__.py @@ -43,7 +43,7 @@ class SiteStatistic(_PluginBase): # 插件图标 plugin_icon = "statistic.png" # 插件版本 - plugin_version = "3.4" + plugin_version = "3.5" # 插件作者 plugin_author = "lightolly" # 作者主页 diff --git a/plugins/sitestatistic/siteuserinfo/__init__.py b/plugins/sitestatistic/siteuserinfo/__init__.py index 7be0b61..3a525cd 100644 --- a/plugins/sitestatistic/siteuserinfo/__init__.py +++ b/plugins/sitestatistic/siteuserinfo/__init__.py @@ -33,6 +33,7 @@ class SiteSchema(Enum): FileList = "FileList" TNode = "TNode" MTorrent = "MTorrent" + Yema = "Yema" class ISiteUserInfo(metaclass=ABCMeta): diff --git a/plugins/sitestatistic/siteuserinfo/mtorrent.py b/plugins/sitestatistic/siteuserinfo/mtorrent.py index 6dda16d..56fb6c2 100644 --- a/plugins/sitestatistic/siteuserinfo/mtorrent.py +++ b/plugins/sitestatistic/siteuserinfo/mtorrent.py @@ -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): diff --git a/plugins/sitestatistic/siteuserinfo/yema.py b/plugins/sitestatistic/siteuserinfo/yema.py new file mode 100644 index 0000000..1cddf6b --- /dev/null +++ b/plugins/sitestatistic/siteuserinfo/yema.py @@ -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 'YemaPT' 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