76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""
|
|
代理池系统配置管理
|
|
统一管理所有配置项，支持环境变量覆盖
|
|
"""
|
|
import os
|
|
from typing import Optional
|
|
|
|
class Config:
|
|
# 数据库配置
|
|
DB_PATH: str = os.getenv("DB_PATH", "db/proxies.db")
|
|
|
|
# API服务配置
|
|
HOST: str = os.getenv("HOST", "0.0.0.0")
|
|
PORT: int = int(os.getenv("PORT", "3000"))
|
|
|
|
# 验证器配置
|
|
VALIDATOR_TIMEOUT: int = int(os.getenv("VALIDATOR_TIMEOUT", "5"))
|
|
VALIDATOR_MAX_CONCURRENCY: int = int(os.getenv("VALIDATOR_MAX_CONCURRENCY", "200"))
|
|
VALIDATOR_CONNECT_TIMEOUT: int = int(os.getenv("VALIDATOR_CONNECT_TIMEOUT", "3"))
|
|
|
|
# 爬虫配置
|
|
CRAWLER_NUM_VALIDATORS: int = int(os.getenv("CRAWLER_NUM_VALIDATORS", "50"))
|
|
CRAWLER_MAX_QUEUE_SIZE: int = int(os.getenv("CRAWLER_MAX_QUEUE_SIZE", "500"))
|
|
|
|
# 定时任务配置
|
|
SCHEDULER_INTERVAL_MINUTES: int = int(os.getenv("SCHEDULER_INTERVAL_MINUTES", "60"))
|
|
SCHEDULER_ENABLED: bool = os.getenv("SCHEDULER_ENABLED", "true").lower() == "true"
|
|
|
|
# 日志配置
|
|
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
|
|
LOG_DIR: str = os.getenv("LOG_DIR", "logs")
|
|
|
|
# 导出配置
|
|
EXPORT_MAX_RECORDS: int = int(os.getenv("EXPORT_MAX_RECORDS", "10000"))
|
|
|
|
# 代理评分配置
|
|
SCORE_VALID: int = int(os.getenv("SCORE_VALID", "10"))
|
|
SCORE_INVALID: int = int(os.getenv("SCORE_INVALID", "-5"))
|
|
SCORE_MIN: int = int(os.getenv("SCORE_MIN", "0"))
|
|
SCORE_MAX: int = int(os.getenv("SCORE_MAX", "100"))
|
|
|
|
# WebSocket配置
|
|
WS_PING_INTERVAL: int = int(os.getenv("WS_PING_INTERVAL", "20"))
|
|
WS_PING_TIMEOUT: int = int(os.getenv("WS_PING_TIMEOUT", "20"))
|
|
|
|
# 插件配置
|
|
PLUGINS_DIR: str = os.getenv("PLUGINS_DIR", "plugins")
|
|
|
|
# CORS配置
|
|
CORS_ORIGINS: str = os.getenv("CORS_ORIGINS", "http://localhost:8080,http://localhost:5173")
|
|
|
|
# API Key配置
|
|
API_KEY: str = os.getenv("API_KEY", "your-api-key-here")
|
|
ADMIN_API_KEY: str = os.getenv("ADMIN_API_KEY", "your-admin-api-key-here")
|
|
REQUIRE_AUTH: bool = os.getenv("REQUIRE_AUTH", "false").lower() == "true"
|
|
|
|
@classmethod
|
|
def get(cls, key: str, default=None):
|
|
"""获取配置项"""
|
|
return getattr(cls, key, default)
|
|
|
|
@classmethod
|
|
def set(cls, key: str, value):
|
|
"""设置配置项（仅限运行时）"""
|
|
setattr(cls, key, value)
|
|
|
|
@classmethod
|
|
def update(cls, updates: dict):
|
|
"""批量更新配置"""
|
|
for key, value in updates.items():
|
|
if hasattr(cls, key):
|
|
setattr(cls, key, value)
|
|
|
|
# 全局配置实例
|
|
config = Config()
|