560 lines
18 KiB
Python
560 lines
18 KiB
Python
"""
|
||
配置管理模块
|
||
|
||
使用 Pydantic Settings 从环境变量和 .env 文件加载配置
|
||
|
||
支持的配置项:
|
||
- 代理设置（代理池、轮换策略）
|
||
- 并发控制（最大并发数、重试次数）
|
||
- 日志级别
|
||
- 邮箱配置（可选）
|
||
- Sentinel 配置（可选）
|
||
"""
|
||
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||
from typing import List, Optional, Literal, Dict, Any
|
||
from dotenv import load_dotenv
|
||
from pathlib import Path
|
||
import random
|
||
|
||
# 加载 .env 文件
|
||
load_dotenv()
|
||
|
||
|
||
class ProxyConfig(BaseModel):
|
||
"""代理配置"""
|
||
|
||
enabled: bool = Field(default=False, description="是否启用代理")
|
||
pool: List[str] = Field(default_factory=list, description="代理池列表")
|
||
rotation: Literal["random", "round_robin"] = Field(
|
||
default="random",
|
||
description="代理轮换策略"
|
||
)
|
||
|
||
_current_index: int = 0 # 用于 round_robin 策略
|
||
|
||
def get_next_proxy(self) -> Optional[str]:
|
||
"""
|
||
获取下一个代理地址
|
||
|
||
根据配置的轮换策略返回代理
|
||
- random: 随机选择
|
||
- round_robin: 轮流使用
|
||
|
||
返回:
|
||
代理地址，如果代理池为空则返回 None
|
||
"""
|
||
if not self.enabled or not self.pool:
|
||
return None
|
||
|
||
if self.rotation == "random":
|
||
return random.choice(self.pool)
|
||
else: # round_robin
|
||
proxy = self.pool[self._current_index % len(self.pool)]
|
||
self._current_index += 1
|
||
return proxy
|
||
|
||
def validate_proxy_format(self, proxy: str) -> bool:
|
||
"""
|
||
验证代理格式是否正确
|
||
|
||
支持格式:
|
||
- http://ip:port
|
||
- http://user:pass@ip:port
|
||
- https://ip:port
|
||
- socks5://ip:port
|
||
|
||
参数:
|
||
proxy: 代理地址
|
||
|
||
返回:
|
||
True 如果格式正确，否则 False
|
||
"""
|
||
import re
|
||
pattern = r'^(http|https|socks5)://([^:]+:[^@]+@)?[\w.-]+:\d+$'
|
||
return bool(re.match(pattern, proxy))
|
||
|
||
|
||
class MailConfig(BaseModel):
|
||
"""邮箱配置"""
|
||
|
||
enabled: bool = Field(default=False, description="是否启用邮箱功能")
|
||
type: Literal["imap", "tempmail", "api", "manual", "cloudmail"] = Field(
|
||
default="manual",
|
||
description="邮箱类型"
|
||
)
|
||
|
||
# IMAP 配置
|
||
imap_host: Optional[str] = Field(default=None, description="IMAP 服务器地址")
|
||
imap_port: int = Field(default=993, description="IMAP 端口")
|
||
imap_username: Optional[str] = Field(default=None, description="邮箱用户名")
|
||
imap_password: Optional[str] = Field(default=None, description="邮箱密码")
|
||
|
||
# 临时邮箱 API 配置
|
||
api_key: Optional[str] = Field(default=None, description="临时邮箱 API Key")
|
||
api_endpoint: Optional[str] = Field(default=None, description="临时邮箱 API 端点")
|
||
|
||
# CloudMail 配置
|
||
cloudmail_api_url: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail API 基础 URL"
|
||
)
|
||
cloudmail_token: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail 身份令牌（预先生成）"
|
||
)
|
||
cloudmail_domain: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail 邮箱域名（例如 mygoband.com）"
|
||
)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典格式（供 MailHandler 使用）"""
|
||
config = {"type": self.type}
|
||
|
||
if self.type == "imap":
|
||
config.update({
|
||
"host": self.imap_host,
|
||
"port": self.imap_port,
|
||
"username": self.imap_username,
|
||
"password": self.imap_password,
|
||
})
|
||
elif self.type in ["tempmail", "api"]:
|
||
config.update({
|
||
"api_key": self.api_key,
|
||
"api_endpoint": self.api_endpoint,
|
||
})
|
||
elif self.type == "cloudmail":
|
||
config.update({
|
||
"api_base_url": self.cloudmail_api_url,
|
||
"token": self.cloudmail_token,
|
||
"domain": self.cloudmail_domain,
|
||
})
|
||
|
||
return config
|
||
|
||
|
||
class SentinelConfig(BaseModel):
|
||
"""Sentinel 配置"""
|
||
|
||
enabled: bool = Field(default=False, description="是否启用 Sentinel 解决器")
|
||
solver_type: Literal["external_script", "api", "module"] = Field(
|
||
default="external_script",
|
||
description="解决器类型"
|
||
)
|
||
|
||
# 外部脚本配置
|
||
script_path: Optional[str] = Field(
|
||
default=None,
|
||
description="外部脚本路径（Node.js 或其他）"
|
||
)
|
||
|
||
# API 配置
|
||
api_endpoint: Optional[str] = Field(
|
||
default=None,
|
||
description="Sentinel solver API 端点"
|
||
)
|
||
api_key: Optional[str] = Field(default=None, description="API Key")
|
||
|
||
# Python 模块配置
|
||
module_name: Optional[str] = Field(
|
||
default=None,
|
||
description="Python 模块名称（例如 'my_sentinel_solver'）"
|
||
)
|
||
|
||
|
||
class AppConfig(BaseSettings):
|
||
"""
|
||
应用配置（从环境变量加载）
|
||
|
||
环境变量优先级:
|
||
1. 系统环境变量
|
||
2. .env 文件
|
||
3. 默认值
|
||
"""
|
||
|
||
model_config = SettingsConfigDict(
|
||
env_file=".env",
|
||
env_file_encoding="utf-8",
|
||
case_sensitive=False,
|
||
extra="ignore" # 忽略未定义的环境变量
|
||
)
|
||
|
||
# ========== 代理配置 ==========
|
||
proxy_enabled: bool = Field(
|
||
default=False,
|
||
description="是否启用代理"
|
||
)
|
||
proxy_pool: str = Field(
|
||
default="",
|
||
description="代理池（逗号分隔），例如: http://ip1:port,socks5://ip2:port"
|
||
)
|
||
proxy_rotation: Literal["random", "round_robin"] = Field(
|
||
default="random",
|
||
description="代理轮换策略"
|
||
)
|
||
|
||
# ========== 并发配置 ==========
|
||
max_workers: int = Field(
|
||
default=1,
|
||
ge=1,
|
||
le=50,
|
||
description="最大并发任务数"
|
||
)
|
||
retry_limit: int = Field(
|
||
default=3,
|
||
ge=0,
|
||
le=10,
|
||
description="失败重试次数"
|
||
)
|
||
retry_delay: int = Field(
|
||
default=5,
|
||
ge=0,
|
||
description="重试延迟（秒）"
|
||
)
|
||
|
||
# ========== 日志配置 ==========
|
||
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = Field(
|
||
default="INFO",
|
||
description="日志级别"
|
||
)
|
||
log_to_file: bool = Field(
|
||
default=True,
|
||
description="是否记录日志到文件"
|
||
)
|
||
|
||
# ========== 邮箱配置 ==========
|
||
mail_enabled: bool = Field(default=False, description="是否启用邮箱功能")
|
||
mail_type: Literal["imap", "tempmail", "api", "manual", "cloudmail"] = Field(
|
||
default="manual",
|
||
description="邮箱类型"
|
||
)
|
||
mail_imap_host: Optional[str] = Field(default=None, description="IMAP 服务器")
|
||
mail_imap_port: int = Field(default=993, description="IMAP 端口")
|
||
mail_imap_username: Optional[str] = Field(default=None, description="邮箱用户名")
|
||
mail_imap_password: Optional[str] = Field(default=None, description="邮箱密码")
|
||
mail_api_key: Optional[str] = Field(default=None, description="临时邮箱 API Key")
|
||
mail_api_endpoint: Optional[str] = Field(default=None, description="API 端点")
|
||
|
||
# CloudMail 配置
|
||
mail_cloudmail_api_url: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail API URL"
|
||
)
|
||
mail_cloudmail_token: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail 身份令牌"
|
||
)
|
||
mail_cloudmail_domain: Optional[str] = Field(
|
||
default=None,
|
||
description="Cloud Mail 邮箱域名（例如 mygoband.com）"
|
||
)
|
||
|
||
# ========== Sentinel 配置 ==========
|
||
sentinel_enabled: bool = Field(default=False, description="是否启用 Sentinel")
|
||
sentinel_solver_type: Literal["external_script", "api", "module"] = Field(
|
||
default="external_script",
|
||
description="Sentinel 解决器类型"
|
||
)
|
||
sentinel_script_path: Optional[str] = Field(default=None, description="脚本路径")
|
||
sentinel_api_endpoint: Optional[str] = Field(default=None, description="API 端点")
|
||
sentinel_api_key: Optional[str] = Field(default=None, description="API Key")
|
||
sentinel_module_name: Optional[str] = Field(default=None, description="模块名称")
|
||
|
||
# Sentinel 内部配置
|
||
sentinel_debug: bool = Field(default=False, description="Sentinel 调试模式")
|
||
sentinel_sdk_path: str = Field(
|
||
default="sdk/sdk.js",
|
||
description="Sentinel SDK JS 文件路径（相对于项目根目录）"
|
||
)
|
||
|
||
# ========== TLS 指纹配置 ==========
|
||
tls_impersonate: Literal["chrome110", "chrome120", "chrome124"] = Field(
|
||
default="chrome124",
|
||
description="模拟的浏览器版本"
|
||
)
|
||
|
||
# ========== 其他配置 ==========
|
||
accounts_output_file: str = Field(
|
||
default="accounts.txt",
|
||
description="成功账号保存文件路径"
|
||
)
|
||
request_timeout: int = Field(
|
||
default=30,
|
||
ge=5,
|
||
le=300,
|
||
description="HTTP 请求超时时间（秒）"
|
||
)
|
||
|
||
@field_validator("proxy_pool")
|
||
@classmethod
|
||
def validate_proxy_pool(cls, v: str) -> str:
|
||
"""验证代理池格式"""
|
||
if not v:
|
||
return v
|
||
|
||
proxies = [p.strip() for p in v.split(",") if p.strip()]
|
||
for proxy in proxies:
|
||
# 基本格式检查
|
||
if not any(proxy.startswith(prefix) for prefix in ["http://", "https://", "socks5://"]):
|
||
raise ValueError(
|
||
f"Invalid proxy format: {proxy}. "
|
||
"Must start with http://, https://, or socks5://"
|
||
)
|
||
|
||
return v
|
||
|
||
@property
|
||
def proxy(self) -> ProxyConfig:
|
||
"""获取代理配置对象"""
|
||
pool = [p.strip() for p in self.proxy_pool.split(",") if p.strip()]
|
||
return ProxyConfig(
|
||
enabled=self.proxy_enabled,
|
||
pool=pool,
|
||
rotation=self.proxy_rotation
|
||
)
|
||
|
||
@property
|
||
def mail(self) -> MailConfig:
|
||
"""获取邮箱配置对象"""
|
||
return MailConfig(
|
||
enabled=self.mail_enabled,
|
||
type=self.mail_type,
|
||
imap_host=self.mail_imap_host,
|
||
imap_port=self.mail_imap_port,
|
||
imap_username=self.mail_imap_username,
|
||
imap_password=self.mail_imap_password,
|
||
api_key=self.mail_api_key,
|
||
api_endpoint=self.mail_api_endpoint,
|
||
cloudmail_api_url=self.mail_cloudmail_api_url,
|
||
cloudmail_token=self.mail_cloudmail_token,
|
||
cloudmail_domain=self.mail_cloudmail_domain,
|
||
)
|
||
|
||
@property
|
||
def sentinel(self) -> SentinelConfig:
|
||
"""获取 Sentinel 配置对象"""
|
||
return SentinelConfig(
|
||
enabled=self.sentinel_enabled,
|
||
solver_type=self.sentinel_solver_type,
|
||
script_path=self.sentinel_script_path,
|
||
api_endpoint=self.sentinel_api_endpoint,
|
||
api_key=self.sentinel_api_key,
|
||
module_name=self.sentinel_module_name,
|
||
)
|
||
|
||
@property
|
||
def fingerprint_config(self) -> Dict[str, Any]:
|
||
"""获取指纹配置"""
|
||
return {
|
||
"user_agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36",
|
||
"screen_width": 1920,
|
||
"screen_height": 1080,
|
||
"languages": ["en-US", "en"],
|
||
"hardware_concurrency": 8,
|
||
}
|
||
|
||
def validate_config(self) -> List[str]:
|
||
"""
|
||
验证配置完整性，返回警告列表
|
||
|
||
返回:
|
||
警告信息列表，空列表表示无警告
|
||
"""
|
||
warnings = []
|
||
|
||
# 检查代理配置
|
||
if self.proxy_enabled and not self.proxy_pool:
|
||
warnings.append("Proxy enabled but proxy_pool is empty")
|
||
|
||
# 检查邮箱配置
|
||
if self.mail_enabled:
|
||
if self.mail_type == "imap":
|
||
if not all([self.mail_imap_host, self.mail_imap_username, self.mail_imap_password]):
|
||
warnings.append("IMAP mail enabled but credentials incomplete")
|
||
elif self.mail_type in ["tempmail", "api"]:
|
||
if not self.mail_api_key:
|
||
warnings.append(f"{self.mail_type} enabled but api_key not configured")
|
||
elif self.mail_type == "cloudmail":
|
||
if not all([self.mail_cloudmail_api_url, self.mail_cloudmail_token, self.mail_cloudmail_domain]):
|
||
warnings.append(
|
||
"CloudMail enabled but config incomplete. "
|
||
"Required: MAIL_CLOUDMAIL_API_URL, MAIL_CLOUDMAIL_TOKEN, MAIL_CLOUDMAIL_DOMAIN"
|
||
)
|
||
|
||
# 检查 Sentinel 配置
|
||
if self.sentinel_enabled:
|
||
if self.sentinel_solver_type == "external_script" and not self.sentinel_script_path:
|
||
warnings.append("Sentinel external_script enabled but script_path not configured")
|
||
elif self.sentinel_solver_type == "api" and not self.sentinel_api_endpoint:
|
||
warnings.append("Sentinel API enabled but api_endpoint not configured")
|
||
elif self.sentinel_solver_type == "module" and not self.sentinel_module_name:
|
||
warnings.append("Sentinel module enabled but module_name not configured")
|
||
|
||
# 检查并发设置
|
||
if self.max_workers > 10 and not self.proxy_enabled:
|
||
warnings.append(
|
||
f"High concurrency ({self.max_workers} workers) without proxy may trigger rate limits"
|
||
)
|
||
|
||
return warnings
|
||
|
||
def print_summary(self):
|
||
"""打印配置摘要"""
|
||
from utils.logger import logger
|
||
|
||
logger.info("=" * 60)
|
||
logger.info("Configuration Summary")
|
||
logger.info("=" * 60)
|
||
logger.info(f"Proxy: {'Enabled' if self.proxy_enabled else 'Disabled'}")
|
||
if self.proxy_enabled:
|
||
logger.info(f" - Pool size: {len(self.proxy.pool)}")
|
||
logger.info(f" - Rotation: {self.proxy_rotation}")
|
||
|
||
logger.info(f"Mail: {'Enabled' if self.mail_enabled else 'Disabled'}")
|
||
if self.mail_enabled:
|
||
logger.info(f" - Type: {self.mail_type}")
|
||
|
||
logger.info(f"Sentinel: {'Enabled' if self.sentinel_enabled else 'Disabled'}")
|
||
if self.sentinel_enabled:
|
||
logger.info(f" - Solver: {self.sentinel_solver_type}")
|
||
|
||
logger.info(f"Concurrency: {self.max_workers} workers")
|
||
logger.info(f"Retry limit: {self.retry_limit}")
|
||
logger.info(f"Log level: {self.log_level}")
|
||
logger.info(f"TLS impersonate: {self.tls_impersonate}")
|
||
logger.info("=" * 60)
|
||
|
||
# 打印警告
|
||
warnings = self.validate_config()
|
||
if warnings:
|
||
logger.warning("Configuration warnings:")
|
||
for warning in warnings:
|
||
logger.warning(f" ⚠️ {warning}")
|
||
|
||
|
||
def load_config() -> AppConfig:
|
||
"""
|
||
加载配置
|
||
|
||
返回:
|
||
AppConfig 实例
|
||
|
||
示例:
|
||
config = load_config()
|
||
print(config.proxy.enabled)
|
||
print(config.mail.type)
|
||
"""
|
||
config = AppConfig()
|
||
return config
|
||
|
||
|
||
def create_default_env_file(path: str = ".env.example"):
|
||
"""
|
||
创建默认的 .env 示例文件
|
||
|
||
参数:
|
||
path: 文件保存路径
|
||
"""
|
||
content = """# OpenAI 账号自动注册系统 - 配置文件示例
|
||
# 复制此文件为 .env 并根据实际情况修改
|
||
|
||
# ========== 代理配置 ==========
|
||
# 是否启用代理（true/false）
|
||
PROXY_ENABLED=false
|
||
|
||
# 代理池（逗号分隔，支持 http/https/socks5）
|
||
# 格式: protocol://[user:pass@]ip:port
|
||
# 示例: http://user:pass@1.2.3.4:8080,socks5://5.6.7.8:1080
|
||
PROXY_POOL=
|
||
|
||
# 代理轮换策略（random: 随机选择, round_robin: 轮流使用）
|
||
PROXY_ROTATION=random
|
||
|
||
# ========== 并发配置 ==========
|
||
# 最大并发任务数（建议 1-5，过高可能触发风控）
|
||
MAX_WORKERS=1
|
||
|
||
# 失败重试次数
|
||
RETRY_LIMIT=3
|
||
|
||
# 重试延迟（秒）
|
||
RETRY_DELAY=5
|
||
|
||
# ========== 日志配置 ==========
|
||
# 日志级别（DEBUG, INFO, WARNING, ERROR）
|
||
LOG_LEVEL=INFO
|
||
|
||
# 是否记录日志到文件
|
||
LOG_TO_FILE=true
|
||
|
||
# ========== 邮箱配置 ==========
|
||
# 是否启用邮箱功能（true/false）
|
||
MAIL_ENABLED=false
|
||
|
||
# 邮箱类型（imap, tempmail, api, cloudmail, manual）
|
||
MAIL_TYPE=manual
|
||
|
||
# IMAP 配置（如果使用 IMAP）
|
||
MAIL_IMAP_HOST=imap.gmail.com
|
||
MAIL_IMAP_PORT=993
|
||
MAIL_IMAP_USERNAME=your@email.com
|
||
MAIL_IMAP_PASSWORD=your_app_password
|
||
|
||
# 临时邮箱 API 配置（如果使用临时邮箱）
|
||
MAIL_API_KEY=
|
||
MAIL_API_ENDPOINT=
|
||
|
||
# CloudMail 配置（如果使用 CloudMail）
|
||
# 1. 先通过 Cloud Mail 管理界面或 API 生成 Token
|
||
# 2. 将 Token 填入 MAIL_CLOUDMAIL_TOKEN
|
||
# 3. 填写你的邮箱域名（不带 @）
|
||
# 4. Token 失效时需要手动更新
|
||
MAIL_CLOUDMAIL_API_URL=https://your-cloudmail-domain.com
|
||
MAIL_CLOUDMAIL_TOKEN=9f4e298e-7431-4c76-bc15-4931c3a73984
|
||
MAIL_CLOUDMAIL_DOMAIN=mygoband.com
|
||
|
||
# ========== Sentinel 配置 ==========
|
||
# 是否启用 Sentinel 解决器（true/false）
|
||
SENTINEL_ENABLED=false
|
||
|
||
# Sentinel 解决器类型（external_script, api, module）
|
||
SENTINEL_SOLVER_TYPE=external_script
|
||
|
||
# 外部脚本路径（如果使用 external_script）
|
||
SENTINEL_SCRIPT_PATH=./sentinel_solver.js
|
||
|
||
# API 配置（如果使用 api）
|
||
SENTINEL_API_ENDPOINT=http://localhost:8000/solve
|
||
SENTINEL_API_KEY=
|
||
|
||
# Python 模块名称（如果使用 module）
|
||
SENTINEL_MODULE_NAME=
|
||
|
||
# ========== TLS 指纹配置 ==========
|
||
# 模拟的浏览器版本（chrome110, chrome120, chrome124）
|
||
TLS_IMPERSONATE=chrome124
|
||
|
||
# ========== 其他配置 ==========
|
||
# 成功账号保存文件路径
|
||
ACCOUNTS_OUTPUT_FILE=accounts.txt
|
||
|
||
# HTTP 请求超时时间（秒）
|
||
REQUEST_TIMEOUT=30
|
||
"""
|
||
|
||
Path(path).write_text(content, encoding="utf-8")
|
||
print(f"✅ Default .env.example created at: {path}")
|
||
|
||
|
||
# 导出主要接口
|
||
__all__ = [
|
||
"AppConfig",
|
||
"ProxyConfig",
|
||
"MailConfig",
|
||
"SentinelConfig",
|
||
"load_config",
|
||
"create_default_env_file",
|
||
]
|