# -*- coding: UTF-8 -*- ''' @Project :domainScanDemo @File :juming.py @IDE :PyCharm @Author :梦伴 @Date :2026/3/19 15:51 @explain : 聚名网API封装类 - 提供登录、域名查询、删除域名列表等功能 ''' import base64 # Base64编码模块 import hashlib # 哈希算法模块 import json # JSON处理模块 import os # 操作系统接口模块 import pickle # 序列化模块 import random # 随机数生成模块 import subprocess # 子进程管理模块 import time # 时间处理模块 from functools import partial from typing import Optional, Tuple, Dict, List, Any # 类型提示 import requests # HTTP请求库 from loguru import logger # 日志记录库 from requests.cookies import RequestsCookieJar # Cookie处理 # 常量定义 BASE_URL = "https://www.juming.com" # 聚名网基础URL CAPTCHA_MAX_RETRY = 5 # 验证码最大重试次数 REQUEST_TIMEOUT = 10 # 请求超时时间（秒） DOWNLOAD_TIMEOUT = 60 # 下载超时时间（秒） SEED_TIME_OFFSET = 286 # 种子时间偏移量 SLIDE_OFFSET = 290 # 滑块偏移量 SLIDE_DURATION = 611 # 滑动持续时间 class MySubprocessPopen(subprocess.Popen): # 自定义子进程类 def __init__(self, *args, **kwargs): # 初始化方法 kwargs['encoding'] = "UTF-8" # 设置编码为UTF-8 super().__init__(*args, **kwargs) # 调用父类初始化 subprocess.Popen = MySubprocessPopen # 替换默认的Popen类 os.environ["EXECJS_RUNTIME"] = "Node" # 设置JavaScript运行时为Node.js startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW subprocess.Popen = partial(subprocess.Popen, encoding='utf-8', startupinfo=startupinfo) 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 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哈希值 def glwb(s: Optional[str]) -> str: # JS文本过滤函数 """ 过滤特殊字符，防止XSS攻击 Args: s: 待过滤的字符串 Returns: 过滤后的字符串 """ if s is None or not isinstance(s, str): # 检查输入是否有效 return '' # 返回空字符串 a_nr = s # 复制字符串 a_nr = a_nr.replace('"', '"') # 替换双引号 a_nr = a_nr.replace("'", ''') # 替换单引号 a_nr = a_nr.replace('<', '<') # 替换小于号 a_nr = a_nr.replace('>', '>') # 替换大于号 a_nr = a_nr.replace('\\', '\') # 替换反斜杠 return a_nr # 返回过滤后的字符串 def md5_19(text: str) -> str: # 计算MD5并返回前19位 """ 计算MD5哈希值并截取前19位 Args: text: 待哈希的文本 Returns: MD5哈希值的前19位 """ if not text: # 检查输入是否为空 return '' # 返回空字符串 md5 = hashlib.md5(text.encode('utf-8')).hexdigest() # 生成32位MD5 return md5[:19] # 返回前19位 def encrypt_password(loginToken: str, password: str) -> str: # 加密密码 """ 使用双重MD5加密密码 Args: loginToken: 登录令牌 password: 明文密码 Returns: 加密后的密码 """ filtered_pwd = glwb(password) # 过滤密码中的特殊字符 step1 = f'[jiami{filtered_pwd}mima]' # 拼接固定盐值 md5_step1 = md5_19(step1) # 第一次MD5加密 step2 = loginToken + md5_step1 # 拼接登录令牌 final_result = md5_19(step2) # 第二次MD5加密 return final_result # 返回最终加密结果 class JM(object): # 聚名网API封装类 token: str # 验证码token loginToken: str = '' # 登录令牌 session_id: str # 会话ID fingerprint: str # 设备指纹 captchaId: str # 验证码ID encryptionPublicKey: str # 加密公钥 cookie: RequestsCookieJar = {} # Cookie存储 # 默认请求头 headers = { 'Host': 'www.juming.com', # 主机名 'sec-ch-ua-platform': '"Windows"', # 平台标识 'x-requested-with': 'XMLHttpRequest', # AJAX请求标识 '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 Edg/146.0.0.0', # 用户代理 'accept': '*/*', # 接受所有类型 'sec-ch-ua': '"Chromium";v="146", "Not-A.Brand";v="24", "Microsoft Edge";v="146"', # 浏览器标识 'sec-ch-ua-mobile': '?0', # 是否移动端 'origin': 'https://www.juming.com', # 请求源 'sec-fetch-site': 'same-origin', # 请求站点 'sec-fetch-mode': 'cors', # 请求模式 'sec-fetch-dest': 'empty', # 请求目标 'referer': 'https://www.juming.com/', # 来源页面 'accept-language': 'zh-CN,zh;q=0.9', # 接受的语言 'priority': 'u=1, i', # 请求优先级 'Content-Type': 'application/x-www-form-urlencoded', # 内容类型 } def __init__(self, proxies: Optional[Dict] = None): # 初始化方法 self.session = requests.Session() # 创建会话对象 self.session.timeout = REQUEST_TIMEOUT # 设置超时时间 self.session.proxies = proxies # 设置代理 self.base_url = BASE_URL # 设置基础URL logger.debug(proxies) def captcha_init(self) -> Tuple[bool, str]: # 初始化滑块验证码 """ 初始化滑块验证码 Returns: Tuple[bool, str]: (是否成功, 消息) """ self.fingerprint = random_fingerprint() # 生成随机指纹 data = { # 构建请求数据 "request_id": self.fingerprint, # 请求ID "scene": "default", # 场景 "seed": calculate_seed(SEED_TIME_OFFSET) # 种子值 } logger.info(data) self.headers['Content-Type'] = 'application/x-www-form-urlencoded' # 设置内容类型 url = f"{self.base_url}/captcha/init" # 初始化URL # self.cookie.clear() # self.cookie.update({'acw_sc__v2': '69d948ed49117d89433d4316a79ada6f00d68f26'}) # logger.info(self.cookie) response = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie).json() # 发送请求 logger.info(response) 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) -> Tuple[bool, str]: # 验证滑块验证码 """ 验证滑块验证码 Returns: Tuple[bool, str]: (是否成功, 消息) """ data = { # 构建验证数据 "offset": SLIDE_OFFSET, # 滑块偏移量 "duration": SLIDE_DURATION, # 滑动持续时间 "trail": [ # 滑动轨迹 {"x": 0, "y": 0, "time": 0}, # 起点 {"x": 0, "y": 0, "time": 17}, # 第1个点 {"x": 1, "y": 0, "time": 89}, # 第2个点 {"x": 5, "y": 0, "time": 106}, # 第3个点 {"x": 14, "y": 0, "time": 123}, # 第4个点 {"x": 28, "y": 0, "time": 139}, # 第5个点 {"x": 47, "y": 0, "time": 156}, # 第6个点 {"x": 74, "y": 0, "time": 173}, # 第7个点 {"x": 102, "y": 0, "time": 189}, # 第8个点 {"x": 131, "y": 0, "time": 206}, # 第9个点 {"x": 159, "y": 0, "time": 223}, # 第10个点 {"x": 184, "y": 0, "time": 239}, # 第11个点 {"x": 205, "y": 0, "time": 256}, # 第12个点 {"x": 227, "y": 1, "time": 273}, # 第13个点 {"x": 248, "y": 0, "time": 289}, # 第14个点 {"x": 263, "y": 0, "time": 306}, # 第15个点 {"x": 276, "y": 0, "time": 323}, # 第16个点 {"x": 286, "y": 0, "time": 339}, # 第17个点 {"x": 290, "y": 0, "time": 356}, # 第18个点 {"x": 290, "y": 0, "time": 373}, # 第19个点 {"x": 290, "y": 0, "time": 389}, # 第20个点 {"x": 290, "y": 0, "time": 406}, # 第21个点 {"x": 290, "y": 0, "time": 423}, # 第22个点 {"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: # 异常处理 with subprocess.Popen( # 执行Node.js脚本 [resolve_node_executable(), jsFile_path], # 命令和参数 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=REQUEST_TIMEOUT) # 传入参数并获取输出 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 self.headers['Content-Type'] = 'application/json' # 设置内容类型 response = self.session.post(url, headers=self.headers, data=verify_data, cookies=self.cookie).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脚本执行超时") # 记录错误 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 user_zh_wxdl_ewm(self) -> Tuple[bool, str, Optional[str]]: # 获取微信登录二维码 """ 获取微信登录二维码 Returns: Tuple[bool, str, Optional[str]]: (是否成功, 消息, 二维码URL) """ url = f"{self.base_url}/user_zh/wxdl_ewm" # 二维码URL try: # 异常处理 response = self.session.post(url, headers=self.headers).json() # 发送请求 self.token = response['data']["token"] # 保存token return response['code'] == 1, response['msg'], response['data']['url'] if response['code'] == 1 else None # 返回结果 except Exception as e: # 异常处理 return False, f"请求失败: {str(e)}", None # 返回失败 def user_zh_p_login(self, email: str, password: str, data: Optional[Dict] = None) -> Tuple[bool, str]: # 账号密码登录 """ 使用账号密码登录 Args: email: 邮箱账号 password: 明文密码 data: 携带的请求数据（用于递归刷新token时传参） Returns: Tuple[bool, str]: (是否成功, 消息) """ try: # 异常处理 captcha_success = False # 验证码成功标志 captcha_msg = f"滑块验证码失败，{CAPTCHA_MAX_RETRY}次内未成功" # 默认失败消息 for _ in range(CAPTCHA_MAX_RETRY): # 循环尝试验证滑块 if data is None: # 如果没有传入data captcha_success, captcha_msg = self.captcha_init() # 初始化验证码 else: # 如果已有data captcha_success = True # 直接视为验证通过 if captcha_success: # 验证码初始化成功 if data is None: # 只有第一次需要验证滑块 captcha_success, captcha_msg = self.captcha_verify() # 验证滑块 if captcha_success: # 滑块验证成功 if data is None: # 首次登录 data = { # 构建登录数据 "re_mm": encrypt_password(self.loginToken, password), # 加密密码 "re_yx": email, # 邮箱 "fs": "tl", # 登录方式 "dltoken": self.loginToken, # 登录令牌 "token": json.dumps( # token（压缩格式JSON） {"code": self.token, "sessionId": self.session_id}, separators=(",", ":") ) } login_url = f"{self.base_url}/user_zh/p_login" # 登录URL response = self.session.post( # 发送登录请求 url=login_url, headers=self.headers, json=data, cookies=self.cookie, ) logger.info(response.text) result = response.json() # 解析响应 code = result.get("code") # 获取状态码 msg = result.get("msg", "登录接口未返回消息") # 获取消息 if code == 1: # 登录成功 self.cookie = response.cookies # 保存Cookie return True, msg # 返回成功 elif code == -118: # token过期 self.loginToken = result.get("token", "") # 更新登录令牌 data["re_mm"] = encrypt_password(self.loginToken, password) # 重新加密密码 data["dltoken"] = self.loginToken # 更新令牌 return self.user_zh_p_login(email, password, data) # 递归重新登录 else: # 其他错误 return False, msg # 返回失败 return captcha_success, captcha_msg # 返回验证码结果 except Exception as e: # 异常处理 return False, f"请求失败: {str(e)}" # 返回失败 def ykj_get_list(self, page: int = 1, page_size: int = 50, data: Optional[Dict] = None) -> Tuple[bool, Any]: # 获取一口价域名列表 """ 获取一口价域名列表 Args: page: 页码 page_size: 每页数量 data: 额外数据 Returns: Tuple[bool, Any]: (是否成功, HTML内容或错误消息) """ data = { # 构建请求数据 'psize': page_size, # 每页数量 'page': page, # 页码 } if data is None else data # 如果有data则使用data url = f"{self.base_url}/ykj/get_list" # 列表URL try: # 异常处理 self.headers['Content-Type'] = 'application/x-www-form-urlencoded' # 设置内容类型 res = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie) # 发送请求 response = res.json() # 解析响应 if response['code'] == -401: # 需要验证码 captcha_response = False, f'滑块验证码失败，{CAPTCHA_MAX_RETRY}次内未成功' # 默认失败 for _ in range(CAPTCHA_MAX_RETRY): # 循环尝试验证 captcha_response = self.captcha_init() # 初始化验证码 if captcha_response[0]: # 初始化成功 captcha_response = self.captcha_verify() # 验证验证码 if captcha_response[0]: # 验证成功 data = { 'psize': page_size, # 每页数量 'page': page, # 页码 # 构建token数据 'token': json.dumps({ # token（压缩格式JSON） 'code': self.token, 'sessionId': self.session_id, }, separators=(",", ":")) } return self.ykj_get_list(page, page_size,data=data) # 递归调用 return captcha_response[0], captcha_response[1] # 返回验证码结果 elif response['code'] == 1 and response.get('data') == 'yzmhuaok': # 验证码通过 self.cookie = res.cookies # 更新Cookie return self.ykj_get_list(page, page_size) # 递归调用 elif response['code'] == 1 and response.get('html'): # 成功获取HTML return response['code'] == 1, response['html'] # 返回HTML return False, response['msg'] # 返回失败 except Exception as e: # 异常处理 return False, f"请求失败: {str(e)}" # 返回失败 def new_cha_del(self, date: str) -> List[str]: # 获取删除域名列表 """ 获取指定日期的删除域名列表 Args: date: 日期（格式：YYYY-MM-DD） Returns: List[str]: 域名列表 """ url = f"{self.base_url}/newcha/del_down?scsj={date}" # 下载URL try: # 异常处理 response = self.session.get(url, headers=self.headers, cookies=self.cookie, allow_redirects=False) # 发送请求 url = response.headers['Location'] # 获取重定向URL response = self.session.get(url, timeout=DOWNLOAD_TIMEOUT) # 下载文件 response_text = response.content.decode('utf-8') # 解码内容 lines = response_text.strip().splitlines() # 按行分割 return lines # 返回域名列表 except Exception as e: # 异常处理 logger.error(f"<获取删除域名列表错误>: {str(e)}") # 记录错误 return [] # 返回空列表 def save_cookies(self, filepath: str = "juming_cookies.pkl") -> None: # 保存Cookie到文件 """ 保存Cookie到文件和Redis Args: filepath: 文件路径 """ with open(filepath, "wb") as f: # 以二进制写入模式打开文件 pickle.dump(self.cookie, f) # 序列化保存 # 保存到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 = {} for cookie in self.cookie: cookie_dict[cookie.name] = cookie.value redis_client.set('domain_tool:juming_cookies', str(cookie_dict)) except Exception as e: logger.error(f"保存Cookie到Redis失败: {str(e)}") pass def load_cookies(self, filepath: str = "juming_cookies.pkl") -> None: # 从文件加载Cookie """ 从文件加载Cookie Args: filepath: 文件路径 """ try: # 异常处理 with open(filepath, "rb") as f: # 以二进制读取模式打开文件 self.cookie = pickle.load(f) # 反序列化加载 except: # 异常处理 self.cookie = requests.cookies.RequestsCookieJar() # 创建空Cookie # if __name__ == '__main__': # 主程序入口 # m = JM() # 创建JM实例 # # # 账号密码登录（请替换为您的账号密码） # email = 'chaofanai1998@gmail.com' # 邮箱账号 # password = 'llzz123,./' # 密码 # # logger.info(f"开始登录: {email}") # 记录日志 # login_result = m.user_zh_p_login(email, password) # 执行登录 # logger.info(f"登录结果: {login_result}") # 记录结果 # # if not login_result[0]: # 登录失败 # logger.error("登录失败，程序退出") # 记录错误 # exit(1) # 退出程序 # # m.load_cookies() # 加载Cookie # # # 获取一口价域名 # import re # 导入正则表达式模块 # # for page in range(1, 2): # 循环获取页面 # success, html = m.ykj_get_list(page) # 获取列表 # if success: # 获取成功 # pattern_ym = r"