Files
domainCheck/detect/juming.py
2026-04-14 22:53:52 +08:00

489 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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 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('<', '&#60;') # 替换小于号
a_nr = a_nr.replace('>', '&#62;') # 替换大于号
a_nr = a_nr.replace('\\', '&#92;') # 替换反斜杠
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脚本
["node", 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"<a class='yda1 ydz' ym='([^']*)'" # 匹配域名
# results = re.findall(pattern_ym, html) # 查找所有域名
# logger.info(results)
# logger.info(f"第{page}页,找到{len(results)}个域名") # 记录结果
# if not results: # 如果没有域名
# logger.info("没有更多域名,停止获取") # 记录日志
# break # 退出循环
# time.sleep(5) # 等待5秒
#
# # 获取删除域名
# deleted_domains = m.new_cha_del("2026-03-11") # 获取删除域名
# logger.info(f"找到{len(deleted_domains)}个删除域名") # 记录结果
#
# m.save_cookies() # 保存Cookie