roll back site test

This commit is contained in:
jxxghp
2024-03-28 13:20:48 +08:00
parent 828fc08362
commit ff07841dd6

View File

@@ -5,7 +5,6 @@ from typing import Union
from urllib.parse import urljoin from urllib.parse import urljoin
from lxml import etree from lxml import etree
from ruamel.yaml import CommentedMap
from app.chain import ChainBase from app.chain import ChainBase
from app.core.config import settings from app.core.config import settings
@@ -327,19 +326,24 @@ class SiteChain(ChainBase):
logger.info(f"清理站点配置:{key}") logger.info(f"清理站点配置:{key}")
self.systemconfig.delete(key) self.systemconfig.delete(key)
def test(self, site_info: Union[str, CommentedMap, dict]) -> Tuple[bool, str]: def test(self, url: str) -> Tuple[bool, str]:
""" """
测试站点是否可用 测试站点是否可用
:param site_info: 站点域名或者站点的数据对象 :param url: 站点域名
:return: (是否可用, 错误信息) :return: (是否可用, 错误信息)
""" """
# 检查域名是否可用 # 检查域名是否可用
if isinstance(site_info, str):
url = site_info
domain = StringUtils.get_url_domain(url) domain = StringUtils.get_url_domain(url)
site_info = self.siteoper.get_by_domain(domain) site_info = self.siteoper.get_by_domain(domain)
if not site_info: if not site_info:
return False, f"站点【{url}】不存在" return False, f"站点【{url}】不存在"
# 模拟登录
try:
# 特殊站点测试
if self.special_site_test.get(domain):
return self.special_site_test[domain](site_info)
# 通用站点测试 # 通用站点测试
site_url = site_info.url site_url = site_info.url
site_cookie = site_info.cookie site_cookie = site_info.cookie
@@ -348,21 +352,7 @@ class SiteChain(ChainBase):
public = site_info.public public = site_info.public
proxies = settings.PROXY if site_info.proxy else None proxies = settings.PROXY if site_info.proxy else None
proxy_server = settings.PROXY_SERVER if site_info.proxy else None proxy_server = settings.PROXY_SERVER if site_info.proxy else None
else:
# 外部站点测试
site_url = site_info.get("url")
domain = StringUtils.get_url_domain(site_url)
site_cookie = site_info.get("cookie")
ua = site_info.get("ua")
render = site_info.get("render")
public = site_info.get("public")
proxies = settings.PROXY if site_info.get("proxy") else None
proxy_server = settings.PROXY_SERVER if site_info.get("proxy") else None
# 模拟登录
try:
# 特殊站点测试
if self.special_site_test.get(domain):
return self.special_site_test[domain](site_info)
# 访问链接 # 访问链接
if render: if render:
page_source = PlaywrightHelper().get_page_source(url=site_url, page_source = PlaywrightHelper().get_page_source(url=site_url,