admin/getDomain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

d

Browse Source
This commit is contained in:
Your Name
2026-04-22 14:13:21 +08:00
parent e0406b5d0e
commit 7cbde2aa78
145 changed files with 23086 additions and 2243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
domainCheck/detect/chinaz.py
View File

@@ -11,15 +11,50 @@
import json # JSON处理模块，用于解析API响应数据
import os # 操作系统接口模块，用于获取脚本所在目录路径
import re # 正则表达式模块，用于提取网页中的标题和加密密钥
import threading
import time # 时间处理模块，用于生成时间戳
from typing import List, Optional # 类型提示
import quickjs # JavaScript执行引擎，用于执行JS加密代码生成签名
import requests # HTTP请求库，用于发送网络请求
from loguru import logger # 日志记录模块，用于记录程序运行日志
from requests.adapters import HTTPAdapter
def check_title(domain: str,sensitive_words: Optional[List[str]] = None, proxies: dict = None):
_CHINAZ_JS_CACHE = None
_SESSION_LOCAL = threading.local()
def _resolve_chinaz_timeout(proxies: dict = None, budget_seconds: Optional[float] = None) -> float:
timeout = float(1.8 if proxies else 2.8)
if budget_seconds not in (None, "", 0, "0"):
timeout = min(timeout, max(0.6, float(budget_seconds or 0.0)))
return max(0.6, timeout)
def _get_session():
session = getattr(_SESSION_LOCAL, "session", None)
if session is not None:
return session
session = requests.Session()
session.mount("http://", HTTPAdapter(pool_connections=32, pool_maxsize=64, max_retries=0))
session.mount("https://", HTTPAdapter(pool_connections=32, pool_maxsize=64, max_retries=0))
_SESSION_LOCAL.session = session
return session
def _load_chinaz_js():
global _CHINAZ_JS_CACHE
if _CHINAZ_JS_CACHE is not None:
return _CHINAZ_JS_CACHE
current_dir = os.path.dirname(__file__)
js_path = os.path.join(current_dir, "chinaz.js")
with open(js_path, "r", encoding="utf-8") as f:
_CHINAZ_JS_CACHE = f.read()
return _CHINAZ_JS_CACHE
def check_title(domain: str, sensitive_words: Optional[List[str]] = None, proxies: dict = None, budget_seconds: Optional[float] = None):
'''
检测域名标题并获取SEO数据
@@ -31,105 +66,133 @@ def check_title(domain: str,sensitive_words: Optional[List[str]] = None, proxies
if sensitive_words is None:
sensitive_words = []
# 添加重试机制
max_retries = 3
for retry in range(max_retries):
try:
# 构建SEO查询URL（站长工具网站）
url = f'https://seo.chinaz.com/{domain}'
# 发送GET请求获取页面内容，设置超时10秒
response = requests.get(url, timeout=10, proxies=proxies)
# 判断请求是否成功（HTTP 200表示成功）
if response.status_code == 200:
# 使用正则表达式提取网站标题（从id="site_title"的div标签中提取文本内容）
# 正则解释：匹配 id="site_title"> 开头，</div> 结尾，中间的内容
match = re.search(r'id="site_title">([^<]+)</div>', response.text)
# 使用正则表达式提取加密密钥enkey（用于后续API请求的身份验证）
# 正则解释：匹配 var enkey = '...' 中的单引号内容
pattern = r"var enkey = '([^']+)''"
try:
step_started_at = time.perf_counter()
print(
f"[chinaz-entry] domain={domain} proxy={'yes' if proxies else 'no'}",
flush=True,
)
session_started_at = time.perf_counter()
logger.info(f"站长之家阶段: domain={domain} | stage=session_init_start")
session = _get_session()
logger.info(
f"站长之家阶段: domain={domain} | stage=session_init_done "
f"| elapsed_ms={int((time.perf_counter() - session_started_at) * 1000)}"
)
# 构建SEO查询URL（站长工具网站）
url = f'https://seo.chinaz.com/{domain}'
page_started_at = time.perf_counter()
logger.info(
f"站长之家阶段: domain={domain} | stage=page_request_start | proxy={'yes' if proxies else 'no'}"
)
# 发送GET请求获取页面内容，设置超时并收口到步骤剩余预算内
response = session.get(
url,
timeout=_resolve_chinaz_timeout(proxies, budget_seconds=budget_seconds),
proxies=proxies,
)
logger.info(
f"站长之家阶段: domain={domain} | stage=page_request_done | status={response.status_code} "
f"| elapsed_ms={int((time.perf_counter() - page_started_at) * 1000)}"
)
# 判断请求是否成功（HTTP 200表示成功）
if response.status_code == 200:
# 使用正则表达式提取网站标题（从id="site_title"的div标签中提取文本内容）
# 正则解释：匹配 id="site_title"> 开头，</div> 结尾，中间的内容
match = re.search(r'id="site_title">([^<]+)</div>', response.text)
# 使用正则表达式提取加密密钥enkey（用于后续API请求的身份验证）
# 正则解释：匹配 var enkey = '...' 中的单引号内容
pattern = r"var enkey = '([^']+)''"
match_enkey = re.search(pattern, response.text)
# 如果上面的正则匹配失败，尝试匹配正确的单引号版本
if not match_enkey:
pattern = r"var enkey = '([^']+)" # 匹配 var enkey = '...' 格式
match_enkey = re.search(pattern, response.text)
# 如果上面的正则匹配失败，尝试匹配正确的单引号版本
if not match_enkey:
pattern = r"var enkey = '([^']+)" # 匹配 var enkey = '...' 格式
match_enkey = re.search(pattern, response.text)
# 如果找到enkey则提取并去除首尾空格，否则为空字符串
enkey = match_enkey.group(1).strip() if match_enkey else ''
# 提取网站标题，如果找到则返回，否则返回空字符串
title = match.group(1).strip() if match else ''
# 如果找到enkey则提取并去除首尾空格，否则为空字符串
enkey = match_enkey.group(1).strip() if match_enkey else ''
if title:
for sensitive_word in sensitive_words:
# 使用正则表达式检查是否包含敏感词
if re.search(sensitive_word, title, re.IGNORECASE):
logger.warning(f'检测到敏感词: {sensitive_word} 在结果: {title}')
return False, f'检测到敏感词: {sensitive_word} 在结果: {title}', ()
# 提取网站标题，如果找到则返回，否则返回空字符串
title = match.group(1).strip() if match else ''
logger.info(
f"站长之家阶段: domain={domain} | stage=page_parsed | title={'yes' if bool(title) else 'no'} "
f"| enkey={'yes' if bool(enkey) else 'no'}"
)
# 调用get_site_data获取详细SEO数据
success, result = get_site_data(domain, enkey, proxies=proxies)
# 检查网站分类
if success and result:
# 定义需要拉黑的分类
blacklist_categories = ['视频电影', '体育运动', '常用查询', '游戏网站', '游戏', '视频', '体育']
# 检查分类是否在黑名单中
if isinstance(result, dict):
# 检查可能的分类字段
category_fields = ['Category', 'category', '分类', '网站分类']
for field in category_fields:
if field in result:
category = result[field]
if isinstance(category, str):
for blacklist_category in blacklist_categories:
if blacklist_category in category:
logger.warning(f'检测到黑名单分类: {category} 在域名: {domain}')
return False, f'网站分类: {category} 属于黑名单分类', ()
# 检查Result是否为列表
elif isinstance(result, list):
for item in result:
if isinstance(item, dict):
category_fields = ['Category', 'category', '分类', '网站分类']
for field in category_fields:
if field in item:
category = item[field]
if isinstance(category, str):
for blacklist_category in blacklist_categories:
if blacklist_category in category:
logger.warning(f'检测到黑名单分类: {category} 在域名: {domain}')
return False, f'网站分类: {category} 属于黑名单分类', ()
return True, 'success', (success, result)
# 请求失败返回空标题和空元组
return True, '', ()
except Exception as e: # 捕获所有异常
error_str = str(e)
# 检查是否是需要重试的错误类型
if any(error_type in error_str for error_type in ['Too many open files', 'EOF occurred in violation of protocol', 'Max retries exceeded', 'Read timed out']):
logger.warning(f"站长工具检测第{retry+1}次失败: {domain}, 错误: {e}")
if retry < max_retries - 1:
time.sleep(2) # 等待2秒后重试
continue
else:
# 达到最大重试次数，返回失败但不拉黑
logger.error(f"站长工具检测失败，已达到最大重试次数: {domain}, 错误: {e}")
return True, 'failure', (False, str(e))
else:
# 其他错误直接返回
logger.error(f"站长工具检测失败: {domain}, 错误: {e}")
return False, str(e), ()
if title:
for sensitive_word in sensitive_words:
# 使用正则表达式检查是否包含敏感词
if re.search(sensitive_word, title, re.IGNORECASE):
logger.warning(f'检测到敏感词: {sensitive_word} 在结果: {title}')
return False, f'检测到敏感词: {sensitive_word} 在结果: {title}', ()
# 调用get_site_data获取详细SEO数据
seo_started_at = time.perf_counter()
logger.info(f"站长之家阶段: domain={domain} | stage=seo_request_start")
remaining_budget_seconds = None
if budget_seconds not in (None, "", 0, "0"):
remaining_budget_seconds = max(0.6, float(budget_seconds or 0.0) - (time.perf_counter() - step_started_at))
success, result = get_site_data(domain, enkey, proxies=proxies, budget_seconds=remaining_budget_seconds)
logger.info(
f"站长之家阶段: domain={domain} | stage=seo_request_done | success={success} "
f"| elapsed_ms={int((time.perf_counter() - seo_started_at) * 1000)}"
)
if not success:
return False, str(result or '站长工具 SEO 数据获取失败'), ()
# 检查网站分类
if success and result:
# 定义需要拉黑的分类
blacklist_categories = ['视频电影', '体育运动', '常用查询', '游戏网站', '游戏', '视频', '体育']
# 检查分类是否在黑名单中
if isinstance(result, dict):
# 检查可能的分类字段
category_fields = ['Category', 'category', '分类', '网站分类']
for field in category_fields:
if field in result:
category = result[field]
if isinstance(category, str):
for blacklist_category in blacklist_categories:
if blacklist_category in category:
logger.warning(f'检测到黑名单分类: {category} 在域名: {domain}')
return False, f'网站分类: {category} 属于黑名单分类', ()
# 检查Result是否为列表
elif isinstance(result, list):
for item in result:
if isinstance(item, dict):
category_fields = ['Category', 'category', '分类', '网站分类']
for field in category_fields:
if field in item:
category = item[field]
if isinstance(category, str):
for blacklist_category in blacklist_categories:
if blacklist_category in category:
logger.warning(f'检测到黑名单分类: {category} 在域名: {domain}')
return False, f'网站分类: {category} 属于黑名单分类', ()
return True, 'success', result
logger.debug(f"站长工具检测失败: {domain}, 状态码: {response.status_code}")
return False, f"站长工具页面返回状态码: {response.status_code}", ()
except Exception as e: # 捕获所有异常
error_str = str(e)
# 检查是否是需要重试的错误类型
if any(error_type in error_str for error_type in ['Too many open files', 'EOF occurred in violation of protocol', 'Max retries exceeded', 'Read timed out']):
logger.error(f"站长工具检测失败，重试链路已耗尽: {domain}, 错误: {e}")
return True, 'failure', (False, str(e))
logger.error(f"站长工具检测失败: {domain}, 错误: {e}")
return False, str(e), ()
# ========================= 【最终请求】 =========================
def get_site_data(domain, enkey: str, proxies: dict = None):
def get_site_data(domain, enkey: str, proxies: dict = None, budget_seconds: Optional[float] = None):
'''
获取站点的详细SEO数据
@@ -143,15 +206,10 @@ def get_site_data(domain, enkey: str, proxies: dict = None):
- (False, 错误信息): 查询失败
'''
try:
# 获取当前脚本所在目录（用于构建JS文件的完整路径）
current_dir = os.path.dirname(__file__)
# 构建JS加密代码文件的完整路径
js_path = os.path.join(current_dir, "chinaz.js")
# 读取JS加密代码文件（包含生成签名的函数）
with open(js_path, "r", encoding="utf-8") as f:
js_code = f.read()
session = _get_session()
js_code = _load_chinaz_js()
js_started_at = time.perf_counter()
logger.info(f"站长之家阶段: domain={domain} | stage=js_context_start")
# 创建JavaScript执行上下文（QuickJS引擎）
ctx = quickjs.Context()
@@ -166,6 +224,10 @@ def get_site_data(domain, enkey: str, proxies: dict = None):
# 获取当前时间戳（毫秒），用于防止缓存和重放攻击
ts = str(int(time.time() * 1000))
logger.info(
f"站长之家阶段: domain={domain} | stage=js_context_done "
f"| elapsed_ms={int((time.perf_counter() - js_started_at) * 1000)}"
)
# 构建请求参数字典（包含API所需的全部参数）
params = {
@@ -198,28 +260,43 @@ def get_site_data(domain, enkey: str, proxies: dict = None):
url = 'https://othertool.chinaz.com/GetTopRanked.ashx'
# 发送GET请求获取SEO数据
resp = requests.get(
api_started_at = time.perf_counter()
logger.info(
f"站长之家阶段: domain={domain} | stage=seo_api_request_start | proxy={'yes' if proxies else 'no'}"
)
resp = session.get(
url, # 请求URL
params=params, # 查询参数（自动拼接为URL参数）
headers=headers, # 请求头
timeout=10, # 超时时间10秒
timeout=_resolve_chinaz_timeout(proxies, budget_seconds=budget_seconds), # 进一步缩短超时，避免长时间卡住线程
proxies=proxies # 代理配置
)
logger.info(
f"站长之家阶段: domain={domain} | stage=seo_api_request_done | status={resp.status_code} "
f"| elapsed_ms={int((time.perf_counter() - api_started_at) * 1000)}"
)
# 判断请求是否成功
if resp.status_code == 200:
# 使用正则表达式提取JSON数据（API返回格式：callback(json_data)）
# 正则解释：匹配括号内的JSON内容
match = re.search(r'\((.*?)\)', resp.text)
if not match:
return False, "站长工具接口返回格式异常"
# 提取JSON字符串
json_str = match.group(1)
# 解析JSON数据为Python字典
data = json.loads(json_str)
logger.info(
f"站长之家阶段: domain={domain} | stage=seo_api_parsed "
f"| state_code={data.get('StateCode')}"
)
# 返回状态码和结果（StateCode为1表示成功）
return data['StateCode'] == 1, data['Result']
return False, f"站长工具接口返回状态码: {resp.status_code}"
except Exception as e: # 捕获所有异常
logger.error(e) # 记录错误日志
@@ -239,4 +316,4 @@ def get_site_data(domain, enkey: str, proxies: dict = None):
#
# title_3, seo_data_3 = check_title(test_domain_3)
# logger.info(f"域名标题: {title_3}")
# logger.info(f"SEO数据: {seo_data_3}")
# logger.info(f"SEO数据: {seo_data_3}")