diff --git a/app/chain/site.py b/app/chain/site.py index cc608923..a772915c 100644 --- a/app/chain/site.py +++ b/app/chain/site.py @@ -49,6 +49,7 @@ class SiteChain(ChainBase): "zhuque.in": self.__zhuque_test, "m-team.io": self.__mteam_test, "m-team.cc": self.__mteam_test, + "ptlsp.com": self.__ptlsp_test, } def is_special_site(self, domain: str) -> bool: @@ -123,6 +124,14 @@ class SiteChain(ChainBase): return True, f"连接成功,但更新状态失败" return False, "Cookie已失效" + @staticmethod + def __ptlsp_test(site: Site) -> Tuple[bool, str]: + """ + 判断站点是否已经登陆:ptlsp + """ + site.url = f"{site.url}index.php" + return __test(site) + @staticmethod def __parse_favicon(url: str, cookie: str, ua: str) -> Tuple[str, Optional[str]]: """ @@ -347,47 +356,54 @@ class SiteChain(ChainBase): return self.special_site_test[domain](site_info) # 通用站点测试 - site_url = site_info.url - site_cookie = site_info.cookie - ua = site_info.ua or settings.USER_AGENT - render = site_info.render - public = site_info.public - proxies = settings.PROXY if site_info.proxy else None - proxy_server = settings.PROXY_SERVER if site_info.proxy else None - - # 访问链接 - if render: - page_source = PlaywrightHelper().get_page_source(url=site_url, - cookies=site_cookie, - ua=ua, - proxies=proxy_server) - if not public and not SiteUtils.is_logged_in(page_source): - if under_challenge(page_source): - return False, f"无法通过Cloudflare!" - return False, f"仿真登录失败,Cookie已失效!" - else: - res = RequestUtils(cookies=site_cookie, - ua=ua, - proxies=proxies - ).get_res(url=site_url) - # 判断登录状态 - if res and res.status_code in [200, 500, 403]: - if not public and not SiteUtils.is_logged_in(res.text): - if under_challenge(res.text): - msg = "站点被Cloudflare防护,请打开站点浏览器仿真" - elif res.status_code == 200: - msg = "Cookie已失效" - else: - msg = f"状态码:{res.status_code}" - return False, f"{msg}!" - elif public and res.status_code != 200: - return False, f"状态码:{res.status_code}!" - elif res is not None: - return False, f"状态码:{res.status_code}!" - else: - return False, f"无法打开网站!" + self.__test(site_info) except Exception as e: return False, f"{str(e)}!" + + @staticmethod + def __test(site_info: Site) -> Tuple[bool, str]: + """ + 通用站点测试 + """ + site_url = site_info.url + site_cookie = site_info.cookie + ua = site_info.ua or settings.USER_AGENT + render = site_info.render + public = site_info.public + proxies = settings.PROXY if site_info.proxy else None + proxy_server = settings.PROXY_SERVER if site_info.proxy else None + + # 访问链接 + if render: + page_source = PlaywrightHelper().get_page_source(url=site_url, + cookies=site_cookie, + ua=ua, + proxies=proxy_server) + if not public and not SiteUtils.is_logged_in(page_source): + if under_challenge(page_source): + return False, f"无法通过Cloudflare!" + return False, f"仿真登录失败,Cookie已失效!" + else: + res = RequestUtils(cookies=site_cookie, + ua=ua, + proxies=proxies + ).get_res(url=site_url) + # 判断登录状态 + if res and res.status_code in [200, 500, 403]: + if not public and not SiteUtils.is_logged_in(res.text): + if under_challenge(res.text): + msg = "站点被Cloudflare防护,请打开站点浏览器仿真" + elif res.status_code == 200: + msg = "Cookie已失效" + else: + msg = f"状态码:{res.status_code}" + return False, f"{msg}!" + elif public and res.status_code != 200: + return False, f"状态码:{res.status_code}!" + elif res is not None: + return False, f"状态码:{res.status_code}!" + else: + return False, f"无法打开网站!" return True, "连接成功" def remote_list(self, channel: MessageChannel, userid: Union[str, int] = None):