236 lines
9.5 KiB
Python
236 lines
9.5 KiB
Python
# -*- coding: UTF-8 -*-
|
||
'''
|
||
@Project :domainScanDemo
|
||
@File :juziseo.py
|
||
@IDE :PyCharm
|
||
@Author :梦伴
|
||
@Date :2026/4/10 15:00
|
||
@explain : 桔子SEO API封装类 - 提供登录、域名查询等功能
|
||
'''
|
||
|
||
# 导入标准库
|
||
import os # 操作系统接口
|
||
import pickle # 序列化反序列化
|
||
import re
|
||
import time # 时间处理
|
||
from Crypto.Cipher import AES
|
||
from Crypto.Util.Padding import pad
|
||
import base64
|
||
# 导入第三方库
|
||
import requests # HTTP请求库
|
||
from loguru import logger # 日志记录
|
||
from requests.cookies import RequestsCookieJar # Cookie管理
|
||
|
||
from detect.geetest2 import Geetest2
|
||
|
||
|
||
|
||
# AES-CBC 加密
|
||
def aes_cbc_encrypt(data, key=b'pvjxzjmzwawfscft', iv=b'qvibva1wg0uxwjeu'):
|
||
cipher = AES.new(key, AES.MODE_CBC, iv)
|
||
padded_data = pad(data, AES.block_size) # PKCS7填充
|
||
encrypted = cipher.encrypt(padded_data)
|
||
return base64.b64encode(encrypted).decode()
|
||
|
||
|
||
class Juziseo:
|
||
"""
|
||
桔子SEO API封装类
|
||
"""
|
||
cookie: RequestsCookieJar = {} # 桔子SEO 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://seo.juziseo.com', # 请求源
|
||
'priority': 'u=1, i', # 请求优先级
|
||
'referer': 'https://seo.juziseo.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):
|
||
"""
|
||
初始化桔子SEO类
|
||
|
||
:param proxies: 代理配置
|
||
"""
|
||
self.validate = None
|
||
self.challenge = None
|
||
self.gt = None
|
||
self.session = requests.Session() # 创建会话对象
|
||
self.session.proxies = proxies # 设置代理
|
||
self.session.timeout = 10 # 设置超时时间10秒
|
||
self.base_url = "https://seo.juziseo.com" # 设置基础URL
|
||
|
||
def save_cookies(self, filepath="juziseo_cookies.pkl"):
|
||
"""
|
||
保存Cookie到文件和Redis
|
||
|
||
:param filepath: 文件路径
|
||
"""
|
||
with open(filepath, "wb") as f: # 以二进制写入模式打开文件
|
||
pickle.dump(self.cookie, f) # 序列化保存Cookie
|
||
logger.info(f"已保存桔子SEO 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 = {}
|
||
for cookie in self.cookie:
|
||
cookie_dict[cookie.name] = cookie.value
|
||
redis_client.set('domain_tool:juziseo_cookies', str(cookie_dict))
|
||
except Exception as e:
|
||
pass
|
||
|
||
def load_cookies(self, filepath="juziseo_cookies.pkl"):
|
||
"""
|
||
从文件加载Cookie
|
||
|
||
:param filepath: 文件路径
|
||
"""
|
||
try: # 尝试加载
|
||
with open(filepath, "rb") as f: # 以二进制读取模式打开文件
|
||
self.cookie = pickle.load(f) # 反序列化加载Cookie
|
||
# logger.info(f"已从 {filepath} 加载桔子SEO Cookie")
|
||
except: # 加载失败
|
||
self.cookie = requests.cookies.RequestsCookieJar() # 创建空的CookieJar
|
||
logger.warning(f"加载桔子SEO Cookie失败，创建空Cookie")
|
||
|
||
def start_msg_captcha_servlet(self):
|
||
url = f"{self.base_url}/class/gtcode/msg/StartMsgCaptchaServlet.php"
|
||
response = requests.get(url, headers=self.headers).json()
|
||
logger.info(response)
|
||
self.gt = response['gt']
|
||
self.challenge = response['challenge']
|
||
return response['success'] == 1
|
||
|
||
def get_captcha(self):
|
||
GETT2 = Geetest2()
|
||
code = GETT2.get_tp(gt=self.gt, challenge=self.challenge, type_='auto')
|
||
logger.info(code)
|
||
if code['result'] == 'success':
|
||
self.challenge = code['challenge']
|
||
self.gt = code['gt']
|
||
self.validate = code['validate']
|
||
return True
|
||
else:
|
||
logger.error(f"获取桔子SEO验证码失败: {code.get('msg', '未知错误')}")
|
||
return False
|
||
|
||
def login(self, username, password):
|
||
"""
|
||
登录桔子SEO
|
||
|
||
:param username: 账号
|
||
:param password: 密码
|
||
:return: tuple - (是否成功, 消息)
|
||
"""
|
||
try:
|
||
for _ in range(3):
|
||
if self.start_msg_captcha_servlet():
|
||
if self.get_captcha():
|
||
break
|
||
data = {
|
||
'return_url': '/',
|
||
'user_name': aes_cbc_encrypt(bytes(username, 'utf-8')),
|
||
'password': aes_cbc_encrypt(bytes(password, 'utf-8')),
|
||
'geetest_challenge': self.challenge,
|
||
'geetest_validate': self.validate,
|
||
'geetest_seccode': self.validate + '|jordan',
|
||
'_post_type': 'ajax',
|
||
}
|
||
logger.debug(data)
|
||
# 发送登录请求
|
||
url = f"{self.base_url}/account/ajax/login_process/"
|
||
response = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie)
|
||
logger.info(response.text)
|
||
# 解析响应
|
||
result = response.json()
|
||
|
||
if result.get('errno') == 1:
|
||
# 登录成功，保存Cookie
|
||
self.cookie = response.cookies
|
||
self.save_cookies()
|
||
logger.info("桔子SEO登录成功")
|
||
return True, "登录成功"
|
||
else:
|
||
logger.error(f"桔子SEO登录失败: {result.get('msg', '未知错误')}")
|
||
return False, result.get('msg', '登录失败')
|
||
except Exception as e:
|
||
logger.error(f"桔子SEO登录异常: {e}")
|
||
return False, f"登录失败: {str(e)}"
|
||
|
||
def check_history(self, domain: str, sensitive_words: list = None):
|
||
# 检测域名历史是否存在敏感词
|
||
|
||
if sensitive_words is None:
|
||
sensitive_words = []
|
||
data = {
|
||
'qrtypeindex': '1',
|
||
'domains': domain,
|
||
'_post_type': 'ajax',
|
||
}
|
||
url = f"{self.base_url}/snapshot/save/"
|
||
response = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie).json()
|
||
# logger.info(response)
|
||
if response['errno'] == 1:
|
||
url = response['rsm']['url']
|
||
response_html = self.session.get(url, headers=self.headers, cookies=self.cookie).content.decode('utf-8')
|
||
title_sensitive_words_match = re.search(r'标题敏感词\D*(\d+)', response_html, re.S)
|
||
title_suspected_sensitive_words_match = re.search(r'标题有疑似敏感词\D*(\d+)', response_html, re.S)
|
||
content_sensitive_words_match = re.search(r'内容敏感词\D*(\d+)', response_html, re.S)
|
||
baidu_sensitive_words_match = re.search(r'百度历史收录敏感\D*(\d+)', response_html, re.S)
|
||
if bool(title_sensitive_words_match or title_suspected_sensitive_words_match or content_sensitive_words_match or baidu_sensitive_words_match):
|
||
return False, "存在敏感词"
|
||
subdomain_match = re.search(r'子域名：\D*(\d+)', response_html, re.S)
|
||
if subdomain_match:
|
||
return False, f"存在子域名: {subdomain_match.group(1)}"
|
||
|
||
for sensitive_word in sensitive_words:
|
||
if sensitive_word in response_html:
|
||
return False, f"存在敏感词: {sensitive_word}"
|
||
return True, 'success'
|
||
return False, str(response.get('err', '请求失败'))
|
||
|
||
def check_external_link(self, domain: str, sensitive_words: list = None):
|
||
# 外链查询域名是否存在敏感词
|
||
if sensitive_words is None:
|
||
sensitive_words = []
|
||
data = {
|
||
'qrtypeindex': '1',
|
||
'domains': domain,
|
||
'_post_type': 'ajax',
|
||
}
|
||
url = f"{self.base_url}/domain_rank/save_domain/"
|
||
response = self.session.post(url, headers=self.headers, data=data, cookies=self.cookie).json()
|
||
logger.info(response)
|
||
if response['errno'] == 1:
|
||
url = response['rsm']['url']
|
||
response_html = self.session.get(url, headers=self.headers, cookies=self.cookie).content.decode('utf-8')
|
||
subdomain_match = re.search(r'子域名：\D*(\d+)', response_html, re.S)
|
||
if subdomain_match:
|
||
return False, f"存在子域名: {subdomain_match.group(1)}"
|
||
for sensitive_word in sensitive_words:
|
||
if sensitive_word in response_html:
|
||
return False, f"存在敏感词: {sensitive_word}"
|
||
return True, 'success'
|
||
return False, str(response.get('err', '请求失败')) |