# -*- coding: UTF-8 -*- ''' @Project :domainScanDemo @File :jucha.py @IDE :PyCharm @Author :梦伴 @Date :2026/4/2 22:49 @explain : 聚查网API封装类 - 优化版，减少冗余代码，添加详细注释 ''' # 导入标准库 import base64 # Base64编码解码 import hashlib # 哈希加密 import json # JSON数据处理 import os # 操作系统接口 import pickle # 序列化反序列化 import random # 随机数生成 import shutil import subprocess # 子进程管理 import time # 时间处理 # 导入第三方库 import requests # HTTP请求库 from loguru import logger # 日志记录 from requests.cookies import RequestsCookieJar # Cookie管理 # # 移除自定义子进程类的替换，避免影响其他模块 # class MySubprocessPopen(subprocess.Popen): # 自定义子进程类 # def __init__(self, *args, **kwargs): # 初始化方法 # kwargs['encoding'] = "UTF-8" # 设置默认编码为UTF-8 # super().__init__(*args, **kwargs) # 调用父类初始化方法 # # os.environ["EXECJS_RUNTIME"] = "Node" # 设置JavaScript运行时环境为Node.js _ORIGINAL_POPEN = subprocess.Popen def create_subprocess(*args, **kwargs): kwargs.setdefault("encoding", "utf-8") if hasattr(subprocess, "STARTUPINFO") and hasattr(subprocess, "STARTF_USESHOWWINDOW"): startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW kwargs.setdefault("startupinfo", startupinfo) return _ORIGINAL_POPEN(*args, **kwargs) def resolve_node_executable() -> str: current_dir = os.path.dirname(__file__) project_dir = os.path.dirname(current_dir) bundled_node = os.path.join(project_dir, "tools", "node-v20.19.4-win-x64", "node.exe") if os.path.exists(bundled_node): return bundled_node detected = shutil.which("node") if detected: return detected fallback_candidates = [ "/usr/bin/node", "/usr/local/bin/node", os.path.expanduser("~/.nvm/versions/node/current/bin/node"), os.path.expanduser("~/.nvm/versions/node/v20.20.2/bin/node"), "/root/.nvm/versions/node/v20.20.2/bin/node", ] for candidate in fallback_candidates: if os.path.exists(candidate): return candidate return "node" def calculate_seed(t: int) -> str: # 计算验证码种子值 now = int(time.time()) # 获取当前时间戳 c = now - t # 计算时间差 seed = base64.b64encode(str(c).encode()).decode() # Base64编码 return seed # 返回种子字符串 def random_fingerprint() -> str: # 生成随机设备指纹 data = str(random.random()) + str(time.time()) # 组合随机数和时间戳 return hashlib.sha256(data.encode()).hexdigest() # 返回SHA256哈希值 class JC(object): # 聚查网API封装类 token: str # 验证码token session_id: str # 会话ID fingerprint: str # 设备指纹 captchaId: str # 验证码ID encryptionPublicKey: str # 加密公钥 cookie: RequestsCookieJar = {} # 聚查网Cookie juming_cookie: RequestsCookieJar = {} # 聚名网Cookie # 通用请求头配置 headers = { 'accept': 'application/json, text/javascript, */*; q=0.01', # 接受的内容类型 'accept-language': 'zh-CN,zh;q=0.9', # 接受的语言 'content-type': 'application/x-www-form-urlencoded', # 内容类型 'origin': 'https://www.jucha.com', # 请求源 'priority': 'u=1, i', # 请求优先级 'referer': 'https://www.jucha.com/login', # 来源页面 'sec-ch-ua': '"Chromium";v="146", "Not-A.Brand";v="24", "Google Chrome";v="146"', # 浏览器标识 'sec-ch-ua-mobile': '?0', # 是否移动端 'sec-ch-ua-platform': '"Windows"', # 操作系统平台 'sec-fetch-dest': 'empty', # 请求目标 'sec-fetch-mode': 'cors', # 请求模式 'sec-fetch-site': 'same-origin', # 请求站点 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36', # 用户代理 'x-requested-with': 'XMLHttpRequest', # AJAX请求标识 } def __init__(self, proxies: dict = None): # 初始化JC类 self.session = requests.Session() # 创建会话对象 self.session.timeout = 10 # 设置超时时间 self.session.proxies = proxies # 设置代理 self.base_url = "https://www.jucha.com" # 设置基础URL self.cookie = requests.cookies.RequestsCookieJar() # 初始化CookieJar self.juming_cookie = requests.cookies.RequestsCookieJar() # 初始化聚名网CookieJar def _get_headers(self, referer: str = None) -> dict: # 获取请求头（可自定义referer） headers = self.headers.copy() # 复制默认请求头 if referer: # 如果指定了referer headers['referer'] = referer # 更新referer return headers # 返回请求头 def _get_route_headers(self, route: str) -> dict: # 根据路由获取对应的请求头 referer_map = { # 路由到referer的映射 'whois': f'{self.base_url}/whois/', # WHOIS查询的referer 'beian': f'{self.base_url}/baian/', # 备案查询的referer 'safe': f'{self.base_url}/safe/', # 安全检测的referer } return self._get_headers(referer_map.get(route)) # 返回对应的请求头 def _handle_captcha(self, retry_times: int = 5) -> tuple: # 处理滑块验证码 for _ in range(retry_times): # 循环重试 init_result = self.captcha_init() # 初始化验证码 if init_result[0]: # 初始化成功 verify_result = self.captcha_verify() # 验证验证码 if verify_result[0]: # 验证成功 return True, '验证码验证成功' # 返回成功 return False, f'滑块验证码失败，{retry_times}次内未成功' # 返回失败 def _check_request(self, url: str, data: dict, headers: dict, cookies=None) -> dict: # 检查请求并处理验证码 res = self.session.post(url=url, data=data, headers=headers, cookies=cookies) # 发送POST请求 try: response = res.json() # 尝试解析JSON except Exception as e: # JSON解析失败 logger.error(f"JSON解析失败: {str(e)}") # 记录错误 logger.error(f"原始响应内容（前500字符）: {res.text[:500]}") # 记录原始响应 return {'code': -1, 'msg': f'JSON解析失败: {str(e)}'} # 返回错误 return response # 返回响应 def _check_and_handle_captcha(self, response: dict, domain: str, route: str, xm_codes, data: dict = None) -> tuple: # 检查并处理验证码 # 检查response是否为字典 if not isinstance(response, dict): return False, None, response # 返回失败 if response.get('code') == 1001: # 需要验证码 captcha_result = self._handle_captcha() # 处理验证码 if captcha_result[0]: # 验证码验证成功 new_data = data.copy() if data else {} # 复制数据 new_data.update({ # 更新验证码参数 '_csrf': '', # CSRF令牌 'ymlb': domain, # 域名 'xm_codes[]': xm_codes, # 检测代码 'type': '2' if route in ['beian', 'safe'] else '1', # 类型：备案和安全检测为2，WHOIS为1 'route': route, # 路由 'captcha_verify_param': self.token, # 验证码token 'sessionId': self.session_id, # 会话ID }) return True, new_data, None # 返回需要重试 return False, None, captcha_result # 返回失败 return False, None, response # 不需要验证码 def captcha_init(self): # 初始化滑块验证码 self.fingerprint = random_fingerprint() # 生成随机指纹 data = { # 构建请求数据 "request_id": self.fingerprint, # 请求ID "scene": "default", # 场景 "seed": calculate_seed(286) # 种子值 } url = f"{self.base_url}/captcha/init" # 初始化URL response = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie).json() # 发送请求 if response['code'] == 1: # 初始化成功 self.captchaId = response['data']['captchaId'] # 保存验证码ID self.encryptionPublicKey = response['data']['encryptionPublicKey'] # 保存加密公钥 return response['code'] == 1, response['msg'] # 返回结果 def captcha_verify(self): # 验证滑块验证码 data = { # 构建验证数据 "offset": 290, # 滑块偏移量 "duration": 611, # 滑动持续时间 "trail": [ # 滑动轨迹 {"x": 0, "y": 0, "time": 0}, {"x": 0, "y": 0, "time": 17}, {"x": 1, "y": 0, "time": 89}, {"x": 5, "y": 0, "time": 106}, {"x": 14, "y": 0, "time": 123}, {"x": 28, "y": 0, "time": 139}, {"x": 47, "y": 0, "time": 156}, {"x": 74, "y": 0, "time": 173}, {"x": 102, "y": 0, "time": 189}, {"x": 131, "y": 0, "time": 206}, {"x": 159, "y": 0, "time": 223}, {"x": 184, "y": 0, "time": 239}, {"x": 205, "y": 0, "time": 256}, {"x": 227, "y": 1, "time": 273}, {"x": 248, "y": 0, "time": 289}, {"x": 263, "y": 0, "time": 306}, {"x": 276, "y": 0, "time": 323}, {"x": 286, "y": 0, "time": 339}, {"x": 290, "y": 0, "time": 356}, {"x": 290, "y": 0, "time": 373}, {"x": 290, "y": 0, "time": 389}, {"x": 290, "y": 0, "time": 406}, {"x": 290, "y": 0, "time": 423}, {"x": 290, "y": 0, "time": 478} ], "fingerprint": self.fingerprint, # 设备指纹 "captchaId": self.captchaId, # 验证码ID "serverPublicKey": self.encryptionPublicKey # 服务器公钥 } data_str = json.dumps(data, separators=(",", ":")) # 转换为JSON字符串 current_dir = os.path.dirname(__file__) jsFile_path = current_dir + "/sdk_leg_env.js" # JavaScript脚本路径 if not os.path.exists(jsFile_path): # 检查脚本文件是否存在 logger.error(f"Node.js脚本文件不存在: {jsFile_path}") # 记录错误 return False, f"缺少签名脚本: {jsFile_path}" # 返回失败 try: # 尝试执行Node.js脚本 with create_subprocess( # 创建子进程 [resolve_node_executable(), jsFile_path], # 执行node命令 stdin=subprocess.PIPE, # 标准输入管道 stdout=subprocess.PIPE, # 标准输出管道 stderr=subprocess.PIPE, # 标准错误管道 text=True, # 文本模式 encoding="utf-8", # UTF-8编码 errors="ignore" # 忽略编码错误 ) as proc: # 进程上下文 stdout, stderr = proc.communicate(data_str, timeout=10) # 传入参数并获取输出 verify_data = stdout.strip() # 去除空白字符 if stderr: # 如果有错误输出 logger.error(f'genData_stderr->[{stderr}]') # 记录错误 if proc.returncode != 0: # 检查退出码 return False, f"Node.js脚本执行失败，退出码: {proc.returncode}, 错误信息: {stderr}" # 返回失败 url = f"{self.base_url}/captcha/verify" # 验证URL verify_headers = self.headers.copy() # 复制请求头 verify_headers['Content-Type'] = 'application/json' # 设置内容类型（仅用于验证请求） verify_response = self.session.post(url, headers=verify_headers, data=verify_data, cookies=self.cookie) # 发送验证请求 if verify_response.cookies: # 如果响应中有新的Cookie self.cookie.update(verify_response.cookies) # 合并Cookie（而不是替换） response = verify_response.json() # 解析JSON响应 if response['code'] == 1: # 验证成功 self.token = response['data']['token'] # 保存token self.session_id = response['data']['session_id'] # 保存会话ID return response['code'] == 1, response['msg'] # 返回结果 except subprocess.TimeoutExpired: # 超时异常 logger.error("Node.js脚本执行超时（超过10秒）") # 记录错误 return False, "Node.js脚本执行超时" # 返回失败 except Exception as e: # 其他异常 logger.error(f"执行Node.js脚本异常: {str(e)}") # 记录错误 return False, f"执行Node.js脚本异常: {str(e)}" # 返回失败 if not verify_data: # 检查输出是否为空 return False, "Node.js脚本未输出任何内容" # 返回失败 def save_cookies(self, filepath="jucha_cookies.pkl"): # 保存Cookie到文件 with open(filepath, "wb") as f: # 以二进制写入模式打开文件 pickle.dump(self.cookie, f) # 序列化保存Cookie logger.info(f"已保存Cookie到文件: {filepath}") # 保存到Redis try: import redis from app.config import config redis_client = redis.Redis( host=config.REDIS_HOST, port=config.REDIS_PORT, password=config.REDIS_PASSWORD, db=config.REDIS_DB, decode_responses=True ) # 将cookie转换为字典 cookie_dict = {} # 检查self.cookie的类型 if isinstance(self.cookie, dict): # 如果是字典，直接使用 cookie_dict = self.cookie logger.info(f"Cookie是字典类型，直接使用") elif hasattr(self.cookie, '__iter__'): # 如果是可迭代对象，遍历处理 for cookie in self.cookie: # 检查cookie对象是否有name和value属性 if hasattr(cookie, 'name') and hasattr(cookie, 'value'): cookie_dict[cookie.name] = cookie.value logger.info(f"Cookie是可迭代对象，处理后得到: {cookie_dict}") else: logger.warning(f"Cookie类型不支持: {type(self.cookie)}") redis_client.set('domain_tool:jucha_cookies', str(cookie_dict)) except Exception as e: logger.error(f"保存Cookie到Redis异常: {str(e)}") pass def load_cookies(self, filepath="jucha_cookies.pkl"): # 从文件加载Cookie try: # 尝试加载 with open(filepath, "rb") as f: # 以二进制读取模式打开文件 loaded_cookie = pickle.load(f) # 反序列化加载Cookie # 检查加载的cookie类型 if isinstance(loaded_cookie, dict): # 如果是字典，转换为RequestsCookieJar cookie_jar = requests.cookies.RequestsCookieJar() for name, value in loaded_cookie.items(): cookie_jar.set(name, value) self.cookie = cookie_jar else: self.cookie = loaded_cookie except Exception as e: # 加载失败 logger.error(f"加载Cookie失败: {e}") self.cookie = requests.cookies.RequestsCookieJar() # 创建空的CookieJar def load_juming_cookies(self, filepath="juming_cookies.pkl"): # 从文件加载聚名网Cookie try: # 尝试加载 with open(filepath, "rb") as f: # 以二进制读取模式打开文件 self.juming_cookie = pickle.load(f) # 反序列化加载Cookie except: # 加载失败 self.juming_cookie = requests.cookies.RequestsCookieJar() # 创建空的CookieJar def auth_login(self): # 使用聚名网账号登录聚查网 params = { # 请求参数 'platform': 'juming', # 平台标识 } res = self.session.get( # 发送GET请求 url=f'{self.base_url}/home/login/get_auth_url', # 获取授权URL params=params, # 请求参数 headers=self.headers, # 请求头 cookies=self.juming_cookie # 聚名网Cookie ) if res.cookies: # 如果响应中有Cookie self.cookie.update(res.cookies) # 更新Cookie response = res.json() # 解析JSON响应 if response['code'] == 1: # 获取授权URL成功 url = f'%s&tiao=1' % response['data']['auth_url'] # 构建跳转URL # 合并聚名网Cookie和聚查网Cookie combined_cookies = requests.cookies.RequestsCookieJar() combined_cookies.update(self.juming_cookie) combined_cookies.update(self.cookie) response = self.session.get( # 发送GET请求 url=url, # 跳转URL headers=self.headers, # 请求头 cookies=combined_cookies, # 合并后的Cookie allow_redirects=False # 不自动重定向 ) if response.cookies: # 如果响应中有Cookie # 安全更新Cookie，避免冲突 for name, value in response.cookies.items(): self.cookie[name] = value # 直接赋值，覆盖同名Cookie if response.status_code == 302: # 重定向状态码 url = response.headers['location'] # 获取重定向URL response = self.session.get( # 发送GET请求 url=url, # 重定向URL headers=self.headers, # 请求头 allow_redirects=False # 不自动重定向 ) if response.cookies: # 如果响应中有Cookie # 安全更新Cookie，避免冲突 for name, value in response.cookies.items(): self.cookie[name] = value # 直接赋值，覆盖同名Cookie if response.status_code == 302: # 再次重定向 return True, '登录成功' # 返回成功 return False, '聚名登录过期' # 返回失败 return False, response['msg'] # 返回失败 def _build_check_data(self, domain: str, route: str, xm_codes, data: dict = None) -> dict: # 构建检测请求数据 if data is not None: # 如果data不为空 return data # 直接返回data return { # 构建新的请求数据 '_csrf': '', # CSRF令牌 'ymlb': domain, # 域名 'xm_codes[]': xm_codes, # 检测代码 'type': '2' if route in ['beian', 'safe'] else '1', # 类型：备案和安全检测为2，WHOIS为1 'route': route, # 路由 } def _make_check_request(self, domain: str, route: str, xm_codes, data: dict = None) -> tuple: # 发起检测请求 headers = self._get_route_headers(route) # 获取对应的请求头 data = self._build_check_data(domain, route, xm_codes, data) # 构建请求数据 url = f'{self.base_url}/home_item/check' # 检测URL response = self._check_request(url, data, headers, self.cookie) # 发送请求 return response, data # 返回响应和数据 def _handle_check_response(self, response: dict, domain: str, route: str, xm_codes, data: dict, callback) -> tuple: # 处理检测响应 need_retry, retry_data, final_response = self._check_and_handle_captcha( # 检查并处理验证码 response, domain, route, xm_codes, data ) if need_retry: # 需要重试 return callback(domain=domain, data=retry_data) # 递归调用回调 # 检查final_response是否为字典 if not isinstance(final_response, dict): return False, f'响应格式错误: {final_response}', '' # 返回失败 if final_response['code'] == 1: # 检测成功 return self._get_search_result(final_response, domain, route) # 获取查询结果（传入domain） return False, final_response.get('msg', '未知错误'), '' # 返回失败 def _get_search_result(self, response: dict, domain: str, route: str) -> tuple: # 获取查询结果 data = { # 构建查询数据 '_csrf': '', # CSRF令牌 'domain': domain, # 域名（注意：使用domain而不是ymlb） 'rwid': response['data']['rwid'], # 任务ID 'type': '2' if route in ['beian', 'safe'] else '1', # 类型 'route': 'baian' if route == 'beian' else route, # 路由 } headers = self._get_route_headers(route) # 获取对应的请求头 url = f'{self.base_url}/home/item/search' if route != 'whois' else f'{self.base_url}/home/item/search_one' # 查询URL search_response = self._check_request(url, data, headers, self.cookie) # 发送查询请求 # 检查search_response是否为字典 if not isinstance(search_response, dict): return False, f'响应格式错误: {search_response}', '' # 返回失败 if search_response.get('code', -1) != 1: # 查询失败 return False, search_response.get('msg', '查询失败'), '' # 返回失败 if route == 'whois': # WHOIS查询 # 安全获取WHOIS状态 whois_zt = '' try: data1 = search_response.get('data', {}) if isinstance(data1, dict): data2 = data1.get('data', {}) if isinstance(data2, dict): whois = data2.get('whois', {}) if isinstance(whois, dict): data3 = whois.get('data', {}) if isinstance(data3, dict): data4 = data3.get('data', {}) if isinstance(data4, dict): whois_zt = data4.get('zt', '') except Exception as e: logger.error(f"获取WHOIS状态失败: {e}") return ( # 返回WHOIS结果 True, # 成功 search_response['msg'], # 消息 whois_zt # WHOIS状态 ) elif route == 'beian': # 备案查询 # 安全获取备案数据 beian_data = {} beian_msg = '' try: data1 = search_response.get('data', {}) if isinstance(data1, dict): data2 = data1.get('data', {}) if isinstance(data2, dict): beian = data2.get('beian', {}) if isinstance(beian, dict): data3 = beian.get('data', {}) if isinstance(data3, dict): beian_data = data3.get('data', {}) beian_msg = data3.get('msg', '') except Exception as e: logger.error(f"获取备案数据失败: {e}") is_dict = isinstance(beian_data, dict) # 是否为字典 return ( # 返回备案结果 True, # 成功 search_response['msg'], # 消息 ( # 备案信息元组 beian_data.get('sj', '') if is_dict else '', # 备案时间 beian_data.get('lx', '') if is_dict else '', # 备案类型 beian_data.get('sy', '') if is_dict else '', # 备案首页地址 beian_msg, # 备案状态 ) ) elif route == 'safe': # 安全检测 # 安全获取安全检测数据 safe_data = {} try: data1 = search_response.get('data', {}) if isinstance(data1, dict): safe_data = data1.get('data', {}) except Exception as e: logger.error(f"获取安全检测数据失败: {e}") check_items = [ # 检测项配置列表 ('qqjc', 'QQ检测'), ('weixin', '微信检测'), ('qiang', '被墙检测'), ('dyjc', '抖音检测'), ('bdjc', '百度检测'), ('llqjcgg', '谷歌检测'), ('llqjchh', '火狐检测'), ] results = [] # 结果列表 for item_key, _ in check_items: # 遍历检测项 try: code = 1 msg = '' item_data = safe_data.get(item_key, {}) if isinstance(item_data, dict): data1 = item_data.get('data', {}) if isinstance(data1, dict): code = int(data1.get('data', 1)) # 获取检测码 msg = data1.get('msg', '') # 获取消息 if code == 3: # 如果检测码为3 msg = '拦截' # 设置为拦截 results.append((code, msg)) # 添加到结果列表 except Exception as e: logger.error(f"获取安全检测项 {item_key} 失败: {e}") results.append((1, '查询失败')) # 添加失败结果 return ( # 返回安全检测结果 True, # 成功 search_response['msg'], # 消息 tuple(results) # 安全检测结果元组 ) return False, '未知路由', '' # 返回失败 def check_whois_domain(self, domain: str, data=None): # 查询域名WHOIS信息 try: # 异常处理 response, check_data = self._make_check_request(domain, 'whois', 'whois', data) # 发起检测请求 return self._handle_check_response(response, domain, 'whois', 'whois', check_data, self.check_whois_domain) # 处理响应 except Exception as e: # 异常处理 logger.error(f"check_domain异常: {e}") # 记录错误 return False, str(e), '' # 返回失败 def beian_check_domain(self, domain: str, data=None): # 查询域名备案信息 try: # 异常处理 response, check_data = self._make_check_request(domain, 'beian', 'beian', data) # 发起检测请求 return self._handle_check_response(response, domain, 'beian', 'beian', check_data, self.beian_check_domain) # 处理响应 except Exception as e: # 异常处理 logger.error(f"beian_check_domain异常: {e}") # 记录错误 return False, str(e), [] # 返回失败 def safe_check_domain(self, domain: str, data=None): # 查询域名安全信息 try: # 异常处理 xm_codes = [ # 检测代码列表 'qqjc', # QQ检测 'weixin', # 微信检测 'dyjc', # 抖音检测 'qiang', # 被墙检测 'bdjc', # 百度检测 'llqjcgg', # 谷歌检测 'llqjchh', # 火狐检测 ] response, check_data = self._make_check_request(domain, 'safe', xm_codes, data) # 发起检测请求 return self._handle_check_response(response, domain, 'safe', xm_codes, check_data, self.safe_check_domain) # 处理响应 except Exception as e: # 异常处理 logger.error(f"safe_check_domain异常: {e}") # 记录错误 return False, str(e), [] # 返回失败 # # 测试代码 # if __name__ == '__main__': # j = JC() # 创建JC实例 # j.load_cookies() # j.load_juming_cookies() # 加载聚名网Cookie # logger.info("开始登录...") # login_result = j.auth_login() # 使用聚名网账号登录聚查网 # logger.info(f"登录结果: {login_result}") # if not login_result[0]: # logger.error("登录失败，无法继续测试") # exit(1) # # 测试WHOIS查询 # logger.info("测试WHOIS查询:") # result = j.check_whois_domain('921229.com') # logger.info(f"WHOIS查询结果: {result}") # logger.info(j.cookie) # # # 测试备案查询 # # logger.info("测试备案查询:") # # result = j.beian_check_domain('baidu.com') # # logger.info(f"备案查询结果: {result}") # # # # 测试安全检测 # logger.info("测试安全检测:") # result = j.safe_check_domain('576777.com') # logger.info(f"安全检测结果: {result}") # # # j.save_cookies() # 保存Cookie