654 lines
28 KiB
Python
654 lines
28 KiB
Python
# -*- 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 re
|
||
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('<', '<') # 替换小于号
|
||
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 # 返回最终加密结果
|
||
|
||
|
||
def decode_a0j_strings(js_content):
|
||
"""
|
||
从 JS 代码中提取 a0i 字符串数组，并模拟 a0j 解码函数还原所有字符串。
|
||
|
||
JS 中的 a0j 函数使用自定义 Base64 alphabet 进行字符串解码：
|
||
- 原始 alphabet: abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=
|
||
- 标准 alphabet: ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=
|
||
"""
|
||
# 兼容多行和单行压缩格式
|
||
match = re.search(r"var N\s*=\s*\[(.*?)\];\s*a0i\s*=\s*function", js_content, re.DOTALL)
|
||
if not match:
|
||
# 尝试更宽松的正则
|
||
match = re.search(r"var N=\[(.*?)\];function a0i\(\)", js_content, re.DOTALL)
|
||
if not match:
|
||
raise ValueError("无法找到 a0i 数组")
|
||
|
||
arr_str = match.group(1)
|
||
strings = re.findall(r"'([^']*?)'", arr_str)
|
||
|
||
def a0j_decode(index):
|
||
d = index - 0xfb
|
||
if d < 0 or d >= len(strings):
|
||
return None
|
||
encoded = strings[d]
|
||
|
||
# 转换自定义 Base64 到标准 Base64
|
||
std_b64 = encoded.translate(str.maketrans(
|
||
'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/=',
|
||
'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/='
|
||
))
|
||
|
||
padding = 4 - len(std_b64) % 4
|
||
if padding != 4:
|
||
std_b64 += '=' * padding
|
||
|
||
try:
|
||
return base64.b64decode(std_b64).decode('utf-8')
|
||
except Exception:
|
||
return encoded
|
||
|
||
decoded_strings = {}
|
||
for i in range(0xfb, 0xfb + len(strings)):
|
||
decoded_strings[i] = a0j_decode(i)
|
||
|
||
return decoded_strings, strings
|
||
|
||
|
||
def decrypt_arg1(arg1, m_arr, key_str):
|
||
"""
|
||
核心解密逻辑：
|
||
1. 按 m 数组重排 arg1 字符
|
||
2. 与密钥 key_str 进行 XOR 运算
|
||
"""
|
||
# 步骤1：按 m 数组重排
|
||
q = [''] * len(m_arr)
|
||
for i, ch in enumerate(arg1):
|
||
for j, idx in enumerate(m_arr):
|
||
if idx == i + 1:
|
||
q[j] = ch
|
||
u = ''.join(q)
|
||
|
||
# 步骤2：与密钥进行 XOR
|
||
v = ''
|
||
max_len = min(len(u), len(key_str))
|
||
for x in range(0, max_len, 2):
|
||
u_val = int(u[x:x + 2], 16)
|
||
p_val = int(key_str[x:x + 2], 16)
|
||
xor_result = u_val ^ p_val
|
||
v += f"{xor_result:02x}"
|
||
|
||
return v
|
||
|
||
|
||
def extract_params(js_content):
|
||
"""从 JS 代码中提取所有关键参数"""
|
||
|
||
# 解码字符串表
|
||
decoded_strings, raw_strings = decode_a0j_strings(js_content)
|
||
|
||
# 提取 arg1（兼容空格和压缩格式）
|
||
arg1_match = re.search(r"var arg1\s*=\s*'([A-Fa-f0-9]+)'", js_content)
|
||
arg1 = arg1_match.group(1) if arg1_match else ''
|
||
|
||
# 提取 m 数组（兼容空格和压缩格式）
|
||
m_match = re.search(r"var m\s*=\s*\[(.*?)\]", js_content)
|
||
m_arr = []
|
||
if m_match:
|
||
m_str = m_match.group(1)
|
||
m_arr = [int(x.strip(), 16) for x in m_str.split(',')]
|
||
|
||
# 查找密钥
|
||
# 注意：a0j 函数有自修改机制，实际密钥可能不在 0x115
|
||
# 通过长度判断哪个是十六进制密钥（应该是 40 位数字字符串）
|
||
key_p = ''
|
||
for idx, val in decoded_strings.items():
|
||
if val and len(val) >= 30 and all(c in '0123456789' for c in val):
|
||
key_p = val
|
||
break
|
||
|
||
return {
|
||
'arg1': arg1,
|
||
'm_arr': m_arr,
|
||
'key': key_p,
|
||
'decoded_strings': decoded_strings
|
||
}
|
||
|
||
|
||
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.acw_sc__v2 = None
|
||
self.session = requests.Session() # 创建会话对象
|
||
self.session.timeout = REQUEST_TIMEOUT # 设置超时时间
|
||
self.session.proxies = proxies # 设置代理
|
||
self.base_url = BASE_URL # 设置基础URL
|
||
|
||
def captcha_init(self, retry_count=0) -> Tuple[bool, str]: # 初始化滑块验证码
|
||
"""
|
||
初始化滑块验证码
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 消息)
|
||
"""
|
||
if retry_count > 3: # 限制递归深度
|
||
return False, "验证码初始化失败：重试次数过多"
|
||
|
||
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': '69e9cfd94b5ec072ae8cd67b6f06ff5a0c0e70c8'})
|
||
# logger.info(self.cookie)
|
||
res = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie) # 发送请求
|
||
try:
|
||
response = res.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'] # 返回结果
|
||
except:
|
||
js_content = res.content.decode('utf8')
|
||
params = extract_params(js_content)
|
||
logger.info(params)
|
||
self.acw_sc__v2 = decrypt_arg1(params['arg1'], params['m_arr'], params['key'])
|
||
self.cookie.update({'acw_sc__v2': self.acw_sc__v2})
|
||
logger.info(self.cookie)
|
||
return self.captcha_init(retry_count + 1)
|
||
|
||
def captcha_verify(self, retry_count=0) -> Tuple[bool, str]: # 验证滑块验证码
|
||
"""
|
||
验证滑块验证码
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 消息)
|
||
"""
|
||
if retry_count > 3: # 限制递归深度
|
||
return False, "验证码验证失败：重试次数过多"
|
||
|
||
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' # 设置内容类型
|
||
res = self.session.post(url, headers=self.headers, data=verify_data,
|
||
cookies=self.cookie) # 发送验证请求
|
||
response = res.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脚本执行超时") # 记录错误
|
||
return False, "Node.js脚本执行超时" # 返回失败
|
||
except requests.exceptions.JSONDecodeError: # JSON解析错误
|
||
js_content = res.content.decode('utf8')
|
||
params = extract_params(js_content)
|
||
logger.info(params)
|
||
self.acw_sc__v2 = decrypt_arg1(params['arg1'], params['m_arr'], params['key'])
|
||
self.cookie.update({'acw_sc__v2': self.acw_sc__v2})
|
||
logger.info(self.cookie)
|
||
return self.captcha_verify(retry_count + 1)
|
||
|
||
except Exception as e: # 其他异常
|
||
logger.info(type(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.clear()
|
||
self.cookie = response.cookies # 保存Cookie
|
||
self.cookie.update({'acw_sc__v2': self.acw_sc__v2})
|
||
logger.info(self.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() # 解析响应
|
||
logger.info(response)
|
||
|
||
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.update(res.cookies)
|
||
# self.save_cookies()
|
||
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() # 按行分割
|
||
|
||
# 过滤掉可能包含的日期和非域名行
|
||
domains = []
|
||
for line in lines:
|
||
line = line.strip()
|
||
# 过滤掉空行
|
||
if not line:
|
||
continue
|
||
# 过滤掉日期格式的行（YYYY-MM-DD）
|
||
if re.match(r'^\d{4}-\d{2}-\d{2}$', line):
|
||
continue
|
||
# 过滤掉可能的标题行
|
||
if '域名' in line or 'domain' in line.lower():
|
||
continue
|
||
domains.append(line)
|
||
|
||
return domains # 返回过滤后的域名列表
|
||
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}") # 记录结果
|
||
#
|
||
# m.save_cookies() # 保存Cookie
|
||
#
|
||
# if not login_result[0]: # 登录失败
|
||
# logger.error("登录失败，程序退出") # 记录错误
|
||
# exit(1) # 退出程序
|
||
# #
|
||
# # m.load_cookies() # 加载Cookie
|
||
# logger.info(m.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
|