204 lines
7.3 KiB
Python
204 lines
7.3 KiB
Python
# -*- coding: UTF-8 -*-
|
||
'''
|
||
@Project :domainScanDemo
|
||
@File :c360.py
|
||
@IDE :PyCharm
|
||
@Author :梦伴
|
||
@Date :2026/4/2 15:57
|
||
@explain : 360搜索敏感词检测工具 - 检测域名是否包含敏感词
|
||
'''
|
||
|
||
import re # 正则表达式模块
|
||
import os
|
||
import threading
|
||
import time
|
||
from collections import OrderedDict
|
||
import requests # HTTP请求库
|
||
from loguru import logger # 日志记录
|
||
from requests.adapters import HTTPAdapter
|
||
from typing import List, Tuple, Optional, Dict # 类型提示
|
||
|
||
|
||
# 常量定义
|
||
SO_SEARCH_URL = 'https://www.so.com/s' # 360搜索URL
|
||
SO_REFERER_TEMPLATE = 'https://www.so.com/s?ie=utf-8&q=site%3A{domain}' # Referer模板
|
||
SEARCH_PATTERN = r'target="_blank">([^<]+)</a></h3>' # 搜索结果匹配模式
|
||
BLOCKED_CODE = 3 # 拦截状态码
|
||
_DIRECT_HTTP = requests.Session()
|
||
_DIRECT_HTTP.trust_env = False
|
||
_DIRECT_HTTP.mount("http://", HTTPAdapter(pool_connections=128, pool_maxsize=256, max_retries=0))
|
||
_DIRECT_HTTP.mount("https://", HTTPAdapter(pool_connections=128, pool_maxsize=256, max_retries=0))
|
||
_PROXY_SESSIONS = OrderedDict()
|
||
_PROXY_SESSION_LOCK = threading.Lock()
|
||
SO_TIMEOUT_PROXY = max(
|
||
1.0,
|
||
float(os.getenv("DOMAINCHECK_360_TIMEOUT_PROXY", "1.4") or 1.4),
|
||
)
|
||
SO_TIMEOUT_DIRECT = max(
|
||
1.2,
|
||
float(os.getenv("DOMAINCHECK_360_TIMEOUT_DIRECT", "1.8") or 1.8),
|
||
)
|
||
|
||
|
||
def _proxy_session_cache_limit() -> int:
|
||
try:
|
||
return max(1, int(os.getenv("DOMAINCHECK_360_PROXY_SESSION_CACHE_SIZE", "128") or 128))
|
||
except Exception:
|
||
return 128
|
||
|
||
|
||
def _close_session_quietly(session) -> None:
|
||
try:
|
||
session.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _resolve_360_timeout(proxies: Optional[Dict] = None, budget_seconds: Optional[float] = None) -> float:
|
||
timeout = float(SO_TIMEOUT_PROXY if proxies else SO_TIMEOUT_DIRECT)
|
||
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(proxies: Optional[Dict] = None):
|
||
proxy_url = ""
|
||
if proxies:
|
||
proxy_url = str(proxies.get("https") or proxies.get("http") or "").strip()
|
||
if not proxy_url:
|
||
return _DIRECT_HTTP
|
||
|
||
with _PROXY_SESSION_LOCK:
|
||
session = _PROXY_SESSIONS.get(proxy_url)
|
||
if session is not None:
|
||
_PROXY_SESSIONS.move_to_end(proxy_url)
|
||
return session
|
||
session = requests.Session()
|
||
session.trust_env = False
|
||
session.mount("http://", HTTPAdapter(pool_connections=64, pool_maxsize=128, max_retries=0))
|
||
session.mount("https://", HTTPAdapter(pool_connections=64, pool_maxsize=128, max_retries=0))
|
||
_PROXY_SESSIONS[proxy_url] = session
|
||
while len(_PROXY_SESSIONS) > _proxy_session_cache_limit():
|
||
_, stale_session = _PROXY_SESSIONS.popitem(last=False)
|
||
_close_session_quietly(stale_session)
|
||
return session
|
||
|
||
|
||
def check_domain(
|
||
domain: str,
|
||
sensitive_words: Optional[List[str]] = None,
|
||
proxies: Optional[Dict] = None,
|
||
budget_seconds: Optional[float] = None,
|
||
) -> Tuple[bool, str]:
|
||
"""
|
||
检查域名是否包含敏感词
|
||
|
||
通过360搜索获取域名的搜索结果，然后检查搜索结果中是否包含敏感词
|
||
|
||
Args:
|
||
domain: 要检测的域名
|
||
sensitive_words: 敏感词列表，如果为None则初始化为空列表
|
||
proxies: 代理配置
|
||
|
||
Returns:
|
||
Tuple[bool, str]: (是否通过, 错误信息)
|
||
- True, '': 通过检测，不包含敏感词
|
||
- False, '敏感词：xxx': 包含敏感词，返回具体的敏感词
|
||
"""
|
||
# 初始化敏感词列表
|
||
if sensitive_words is None:
|
||
sensitive_words = []
|
||
|
||
# 构建请求头
|
||
headers = {
|
||
'Accept': '*/*', # 接受所有类型
|
||
'Accept-Language': 'zh-cn', # 中文语言
|
||
'Referer': SO_REFERER_TEMPLATE.format(domain=domain), # 来源页面
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36', # 用户代理
|
||
'Host': 'www.so.com', # 主机名
|
||
}
|
||
|
||
# 构建查询参数
|
||
params = {
|
||
'ie': 'utf-8', # 编码格式
|
||
'q': f'site:{domain}', # 搜索查询
|
||
}
|
||
|
||
try:
|
||
session_started_at = time.perf_counter()
|
||
logger.info(f'360阶段: domain={domain} | stage=session_init_start')
|
||
session = _get_session(proxies)
|
||
logger.info(
|
||
f'360阶段: domain={domain} | stage=session_init_done '
|
||
f'| elapsed_ms={int((time.perf_counter() - session_started_at) * 1000)}'
|
||
)
|
||
# 发送GET请求获取搜索结果
|
||
request_started_at = time.perf_counter()
|
||
logger.info(f'360阶段: domain={domain} | stage=request_start | proxy={"yes" if proxies else "no"}')
|
||
response = None
|
||
try:
|
||
response = session.get(
|
||
SO_SEARCH_URL,
|
||
params=params,
|
||
headers=headers,
|
||
proxies=proxies,
|
||
timeout=_resolve_360_timeout(proxies, budget_seconds=budget_seconds),
|
||
)
|
||
logger.info(
|
||
f'360阶段: domain={domain} | stage=request_done | status={response.status_code} '
|
||
f'| elapsed_ms={int((time.perf_counter() - request_started_at) * 1000)}'
|
||
)
|
||
|
||
# 解析响应内容
|
||
response_html = response.content.decode('utf-8', errors='ignore')
|
||
finally:
|
||
try:
|
||
if response is not None:
|
||
response.close()
|
||
except Exception:
|
||
pass
|
||
|
||
# 使用正则表达式提取搜索结果
|
||
search_results = re.findall(SEARCH_PATTERN, response_html)
|
||
logger.info(f'360阶段: domain={domain} | stage=parsed | result_count={len(search_results)}')
|
||
|
||
# 记录搜索结果数量
|
||
logger.info(f'360搜索结果数量: {len(search_results)}')
|
||
|
||
# 如果没有搜索结果，直接返回通过
|
||
if not search_results:
|
||
return True, ''
|
||
|
||
# 检查每个搜索结果是否包含敏感词
|
||
for result in search_results:
|
||
for sensitive_word in sensitive_words:
|
||
# 使用正则表达式检查是否包含敏感词
|
||
if re.search(sensitive_word, result, re.IGNORECASE):
|
||
logger.warning(f'检测到敏感词: {sensitive_word} 在结果: {result}')
|
||
return False, f'检测到敏感词: {sensitive_word} 在结果: {result}'
|
||
|
||
# 所有搜索结果都不包含敏感词，返回通过
|
||
return True, ''
|
||
|
||
except requests.exceptions.Timeout:
|
||
return False, '360搜索请求超时'
|
||
except requests.exceptions.RequestException as e:
|
||
error_str = str(e)
|
||
logger.debug(f'360搜索请求失败: {error_str}')
|
||
# 检查是否是"Too many open files"错误
|
||
if 'Too many open files' in error_str:
|
||
return True, 'failure'
|
||
return False, f'360搜索请求失败: {error_str}'
|
||
except Exception as e:
|
||
return False, f'未知错误: {str(e)}'
|
||
|
||
|
||
# if __name__ == '__main__':
|
||
# # 测试用例
|
||
# test_domain = 'niuniushouka.com'
|
||
# test_sensitive_words = ['赌博', '色情', '暴力','收卡']
|
||
#
|
||
# logger.info(f'开始检测域名: {test_domain}')
|
||
# result = check_domain(test_domain, test_sensitive_words)
|
||
# logger.info(f'检测结果: {result}')
|