178 lines
6.5 KiB
Python
178 lines
6.5 KiB
Python
"""
|
|
Core configuration management using Pydantic Settings.
|
|
All configuration loaded from environment variables with validation.
|
|
"""
|
|
|
|
from enum import StrEnum
|
|
from typing import Literal
|
|
from pydantic import Field, field_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class ProxyPoolType(StrEnum):
|
|
"""代理池类型枚举"""
|
|
|
|
DEFAULT = "default" # 默认代理池
|
|
EXPIRING = "expiring" # 带有效期的代理池
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
"""
|
|
Application settings with environment variable support.
|
|
|
|
Configuration is loaded from environment variables and .env files.
|
|
All settings are validated at startup using Pydantic validators.
|
|
"""
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
case_sensitive=False,
|
|
extra="ignore",
|
|
)
|
|
|
|
# Application Settings
|
|
app_name: str = Field(default="kami_spider", description="Application name")
|
|
environment: Literal["development", "staging", "production"] = Field(
|
|
default="development", description="Runtime environment"
|
|
)
|
|
debug: bool = Field(default=False, description="Debug mode")
|
|
host: str = Field(default="0.0.0.0", description="Server host")
|
|
port: int = Field(default=8000, description="Server port")
|
|
workers: int = Field(default=1, description="Number of worker processes")
|
|
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
|
|
default="INFO", description="Logging level"
|
|
)
|
|
|
|
# Database Settings
|
|
db_host: str = Field(default="localhost", description="MySQL host")
|
|
db_port: int = Field(default=3306, description="MySQL port")
|
|
db_name: str = Field(default="kami_spider", description="Database name")
|
|
db_user: str = Field(default="root", description="Database user")
|
|
db_password: str = Field(default="", description="Database password")
|
|
db_pool_size: int = Field(default=10, description="Database connection pool size")
|
|
db_max_overflow: int = Field(
|
|
default=20, description="Database max overflow connections"
|
|
)
|
|
db_pool_recycle: int = Field(
|
|
default=3600, description="Database pool recycle time in seconds"
|
|
)
|
|
db_pool_pre_ping: bool = Field(
|
|
default=True, description="Test connections before using"
|
|
)
|
|
db_echo: bool = Field(default=False, description="Echo SQL statements")
|
|
|
|
# Redis Settings
|
|
redis_host: str = Field(default="localhost", description="Redis host")
|
|
redis_port: int = Field(default=6379, description="Redis port")
|
|
redis_db: int = Field(default=0, description="Redis database number")
|
|
redis_password: str = Field(default="", description="Redis password")
|
|
redis_max_connections: int = Field(
|
|
default=50, description="Redis connection pool max connections"
|
|
)
|
|
redis_decode_responses: bool = Field(
|
|
default=True, description="Decode Redis responses to strings"
|
|
)
|
|
|
|
# OpenTelemetry Settings
|
|
otel_enabled: bool = Field(default=True, description="Enable OpenTelemetry")
|
|
otel_service_name: str = Field(
|
|
default="kami_spider", description="Service name for traces"
|
|
)
|
|
otel_exporter_endpoint: str = Field(
|
|
default="38.38.251.113:31547",
|
|
description="OpenTelemetry collector gRPC endpoint",
|
|
)
|
|
otel_exporter_insecure: bool = Field(
|
|
default=True, description="Use insecure gRPC connection"
|
|
)
|
|
otel_sample_rate: float = Field(
|
|
default=1.0, description="Trace sampling rate (0.0 to 1.0)"
|
|
)
|
|
|
|
# CORS Settings
|
|
cors_enabled: bool = Field(default=True, description="Enable CORS")
|
|
cors_allow_origins: list[str] = Field(
|
|
default=["*"], description="Allowed CORS origins"
|
|
)
|
|
cors_allow_credentials: bool = Field(
|
|
default=True, description="Allow credentials in CORS"
|
|
)
|
|
cors_allow_methods: list[str] = Field(
|
|
default=["*"], description="Allowed HTTP methods"
|
|
)
|
|
cors_allow_headers: list[str] = Field(
|
|
default=["*"], description="Allowed HTTP headers"
|
|
)
|
|
|
|
# 代理设置
|
|
proxy_enable: bool = Field(default=True, description="是否启用代理")
|
|
proxy_url: str = Field(
|
|
default="https://share.proxy.qg.net/get?key=7ASQH2BI&num=1&area=&isp=0&format=txt&seq=\r\n&distinct=false",
|
|
description="代理服务器地址",
|
|
)
|
|
proxy_type: ProxyPoolType = Field(
|
|
default=ProxyPoolType.DEFAULT, description="代理服务器类型"
|
|
)
|
|
proxy_username: str = Field(default="7ASQH2BI", description="代理服务器用户名")
|
|
proxy_password: str = Field(default="34D6652FE7B6", description="代理服务器密码")
|
|
|
|
# Security Settings
|
|
secret_key: str = Field(
|
|
default="change-me-in-production", description="Secret key for signing tokens"
|
|
)
|
|
|
|
@field_validator("workers")
|
|
@classmethod
|
|
def validate_workers(cls, v: int) -> int:
|
|
"""Ensure workers is at least 1."""
|
|
if v < 1:
|
|
raise ValueError("workers must be at least 1")
|
|
return v
|
|
|
|
@field_validator("otel_sample_rate")
|
|
@classmethod
|
|
def validate_sample_rate(cls, v: float) -> float:
|
|
"""Ensure sample rate is between 0.0 and 1.0."""
|
|
if not 0.0 <= v <= 1.0:
|
|
raise ValueError("otel_sample_rate must be between 0.0 and 1.0")
|
|
return v
|
|
|
|
@property
|
|
def database_url(self) -> str:
|
|
"""Generate async database URL for SQLModel/SQLAlchemy."""
|
|
password_part = f":{self.db_password}" if self.db_password else ""
|
|
return (
|
|
f"mysql+aiomysql://{self.db_user}{password_part}"
|
|
f"@{self.db_host}:{self.db_port}/{self.db_name}"
|
|
)
|
|
|
|
@property
|
|
def sync_database_url(self) -> str:
|
|
"""Generate sync database URL for Alembic migrations."""
|
|
password_part = f":{self.db_password}" if self.db_password else ""
|
|
return (
|
|
f"mysql+pymysql://{self.db_user}{password_part}"
|
|
f"@{self.db_host}:{self.db_port}/{self.db_name}"
|
|
)
|
|
|
|
@property
|
|
def redis_url(self) -> str:
|
|
"""Generate Redis URL."""
|
|
password_part = f":{self.redis_password}@" if self.redis_password else ""
|
|
return f"redis://{password_part}{self.redis_host}:{self.redis_port}/{self.redis_db}"
|
|
|
|
@property
|
|
def is_production(self) -> bool:
|
|
"""Check if running in production environment."""
|
|
return self.environment == "production"
|
|
|
|
@property
|
|
def is_development(self) -> bool:
|
|
"""Check if running in development environment."""
|
|
return self.environment == "development"
|
|
|
|
|
|
# Global settings instance
|
|
settings = Settings()
|