196 lines
5.5 KiB
Python
196 lines
5.5 KiB
Python
# -*- coding: UTF-8 -*-
|
||
'''
|
||
@Project :domainScanDemo
|
||
@File :config.py
|
||
@IDE :PyCharm
|
||
@Author :梦伴
|
||
@Date :2026/4/9 0:08
|
||
@explain : 系统配置
|
||
'''
|
||
|
||
import os
|
||
import sys
|
||
from dotenv import load_dotenv
|
||
|
||
# 确定基础目录
|
||
if hasattr(sys, '_MEIPASS'):
|
||
# PyInstaller 打包后的临时目录
|
||
BASE_DIR = sys._MEIPASS
|
||
else:
|
||
# 开发环境目录
|
||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
# 加载环境变量
|
||
env_path = os.path.join(BASE_DIR, '.env')
|
||
if os.path.exists(env_path):
|
||
load_dotenv(env_path)
|
||
print(f"成功加载环境变量文件: {env_path}")
|
||
else:
|
||
load_dotenv()
|
||
print(f"环境变量文件不存在: {env_path}，使用默认环境变量")
|
||
|
||
|
||
class Config:
|
||
"""
|
||
系统配置
|
||
"""
|
||
|
||
# 数据库配置
|
||
DB_HOST = os.getenv('DB_HOST', 'localhost')
|
||
DB_PORT = int(os.getenv('DB_PORT', 5432))
|
||
DB_DATABASE = os.getenv('DB_DATABASE', 'domain_scan_db')
|
||
DB_USER = os.getenv('DB_USER', 'postgres')
|
||
DB_PASSWORD = os.getenv('DB_PASSWORD', 'postgres')
|
||
|
||
# 消息队列配置
|
||
RABBITMQ_HOST = os.getenv('RABBITMQ_HOST', 'localhost')
|
||
RABBITMQ_PORT = int(os.getenv('RABBITMQ_PORT', 5672))
|
||
RABBITMQ_USER = os.getenv('RABBITMQ_USER', 'guest')
|
||
RABBITMQ_PASSWORD = os.getenv('RABBITMQ_PASSWORD', 'guest')
|
||
RABBITMQ_VHOST = os.getenv('RABBITMQ_VHOST', '/')
|
||
|
||
# Redis配置
|
||
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
|
||
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
|
||
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', '')
|
||
REDIS_DB = int(os.getenv('REDIS_DB', 0))
|
||
|
||
# 聚名网配置
|
||
JUMING_COOKIE = os.getenv('JUMING_COOKIE', '')
|
||
JUMING_REFERER = os.getenv('JUMING_REFERER', 'https://www.juming.com/')
|
||
JUMING_USER_AGENT = os.getenv('JUMING_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')
|
||
|
||
# 代理配置
|
||
PROXY_ENABLED = os.getenv('PROXY_ENABLED', 'false').lower() == 'true'
|
||
PROXY_URL = os.getenv('PROXY_URL', '')
|
||
|
||
# 检测配置
|
||
DETECT_TIMEOUT = int(os.getenv('DETECT_TIMEOUT', 30))
|
||
DETECT_RETRY_COUNT = int(os.getenv('DETECT_RETRY_COUNT', 3))
|
||
DETECT_CONCURRENCY = int(os.getenv('DETECT_CONCURRENCY', 10))
|
||
|
||
# 域名配置
|
||
DOMAIN_TLDS = ['com', 'net']
|
||
DOMAIN_BATCH_SIZE = int(os.getenv('DOMAIN_BATCH_SIZE', 1000))
|
||
|
||
# 日志配置
|
||
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
|
||
LOG_FILE = os.getenv('LOG_FILE', 'app.log')
|
||
|
||
# 任务配置
|
||
TASK_PRIORITY = {
|
||
'rdap': 10,
|
||
'wayback': 8,
|
||
'baidu': 6,
|
||
'qihu360': 5,
|
||
'google': 5,
|
||
'chinaz': 4,
|
||
'aizhan': 4,
|
||
'juziseo': 3,
|
||
'jucha': 3
|
||
}
|
||
|
||
# 敏感词配置 - 现在从数据库加载
|
||
SENSITIVE_WORDS = []
|
||
|
||
# 检测项配置
|
||
DETECT_ITEMS = {
|
||
'rdap': True,
|
||
'wayback': True,
|
||
'baidu': True,
|
||
'qihu360': True,
|
||
'google': True,
|
||
'chinaz': True,
|
||
'aizhan': True,
|
||
'juziseo': True,
|
||
'jucha': True
|
||
}
|
||
|
||
# 目录配置
|
||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
DATA_DIR = os.path.join(BASE_DIR, 'data')
|
||
LOG_DIR = os.path.join(BASE_DIR, 'logs')
|
||
|
||
# 确保目录存在
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
|
||
|
||
# 从数据库加载敏感词
|
||
def load_sensitive_words():
|
||
"""
|
||
从数据库加载敏感词
|
||
"""
|
||
try:
|
||
from app.utils.database import Database
|
||
db = Database()
|
||
sensitive_words = db.get_sensitive_words()
|
||
words = [word['word'] for word in sensitive_words]
|
||
db.close()
|
||
return words
|
||
except Exception as e:
|
||
print(f"加载敏感词失败: {e}")
|
||
return []
|
||
|
||
# 从文件加载检测选项
|
||
def load_detect_options():
|
||
"""
|
||
从文件加载检测选项
|
||
"""
|
||
try:
|
||
import json
|
||
import os
|
||
if os.path.exists('detect_options.json'):
|
||
with open('detect_options.json', 'r', encoding='utf-8') as f:
|
||
detect_options = json.load(f)
|
||
return detect_options
|
||
return {}
|
||
except Exception as e:
|
||
print(f"加载检测选项失败: {e}")
|
||
return {}
|
||
|
||
# 导出配置
|
||
config = Config()
|
||
# 加载检测选项
|
||
config.DETECT_OPTIONS = load_detect_options()
|
||
|
||
# 延迟加载敏感词，避免循环导入
|
||
def load_sensitive_words_lazy():
|
||
"""
|
||
延迟加载敏感词
|
||
"""
|
||
return load_sensitive_words()
|
||
|
||
# 设置敏感词属性为延迟加载
|
||
config.load_sensitive_words = load_sensitive_words_lazy
|
||
|
||
# 检查检测类型是否应该执行
|
||
def should_detect(detect_type):
|
||
"""
|
||
检查检测类型是否应该执行
|
||
|
||
:param detect_type: 检测类型
|
||
:return: bool - 是否应该执行
|
||
"""
|
||
# 检测类型映射
|
||
type_mapping = {
|
||
'register': 'detect_register',
|
||
'chinaz': 'detect_chinaz',
|
||
'aizhan': 'detect_aizhan',
|
||
'baidu_site': 'detect_baidu_site',
|
||
'360_site': 'detect_360_site',
|
||
'baidu_security': 'detect_baidu_security',
|
||
'whois': 'detect_whois',
|
||
'beian': 'detect_beian',
|
||
'intercept': 'detect_intercept',
|
||
'juziseo': 'detect_juziseo',
|
||
'juziseo_outlink': 'detect_juziseo_outlink'
|
||
}
|
||
|
||
# 获取对应的配置键
|
||
config_key = type_mapping.get(detect_type)
|
||
if not config_key:
|
||
return True # 默认执行
|
||
|
||
# 检查是否在配置中，默认为True
|
||
return config.DETECT_OPTIONS.get(config_key, True) |