AutoScraperX/config/base.py

# base.py 修改后
import os
from dotenv import load_dotenv
from core.utils.path_utils import project_root, log_dir
from pydantic import Field, AnyUrl, field_validator, ConfigDict
from pydantic_settings import BaseSettings  # 注意：BaseSettings 现在在 pydantic_settings 中

load_dotenv(os.path.join(project_root, ".env"))


class Settings(BaseSettings):
    # Pydantic 2.x 使用 ConfigDict 替代内部 Config 类
    model_config = ConfigDict(
        env_file=".env",
        env_file_encoding='utf-8',
        case_sensitive=False,
        env_prefix="",  # 明确指定环境变量前缀
    )

    # 环境标识
    ENV: str = "prod"
    ENABLE_ALIYUN_LOG: bool = True

    # 日志配置
    LOG_DIR: str = log_dir
    LOG_LEVEL: str = "INFO"

    # 阿里云数据库配置
    DB_HOST: str = Field(..., validation_alias="DB_HOST")
    DB_PORT: int = Field(3306, validation_alias="DB_PORT")
    DB_USER: str = Field(..., validation_alias="DB_USER")
    DB_PASSWORD: str = Field(..., validation_alias="DB_PASSWORD")
    DB_NAME: str = Field(..., validation_alias="DB_NAME")
    DB_CHARSET: str = Field(..., validation_alias="DB_CHARSET")
    DB_POOL_SIZE: int = 20
    DB_POOL_RECYCLE: int = 3600

    # 阿里云RocketMQ配置
    ROCKETMQ_ENDPOINT: str = Field(..., validation_alias="ROCKETMQ_ENDPOINT")
    ROCKETMQ_ACCESS_KEY_ID: str = Field(..., validation_alias="ROCKETMQ_ACCESS_KEY_ID")
    ROCKETMQ_ACCESS_KEY_SECRET: str = Field(..., validation_alias="ROCKETMQ_ACCESS_KEY_SECRET")
    ROCKETMQ_INSTANCE_ID: str = Field(..., validation_alias="ROCKETMQ_INSTANCE_ID")
    ROCKETMQ_WAIT_SECONDS: int = 30
    ROCKETMQ_BATCH: int = 1

    # 飞书配置
    FEISHU_APPID: str = Field(..., validation_alias="FEISHU_APPID")
    FEISHU_APPSECRET: str = Field(..., validation_alias="FEISHU_APPSECRET")

    # 连接池配置
    CONNECTION_TIMEOUT: int = 10
    REQUEST_TIMEOUT: int = 30

    # 阿里云日志
    ALIYUN_ACCESS_KEY_ID: str = Field(..., validation_alias="ALIYUN_ACCESS_KEY_ID")
    ALIYUN_ACCESS_KEY_SECRET: str = Field(..., validation_alias="ALIYUN_ACCESS_KEY_SECRET")

    # redis
    REDIS_HOST: str = Field(..., validation_alias="REDIS_HOST")
    REDIS_PORT: int = Field(..., validation_alias="REDIS_PORT")
    REDIS_PASSWORD: str = Field(..., validation_alias="REDIS_PASSWORD")
    REDIS_DB: int = Field(0, validation_alias="REDIS_DB")
    REDIS_MAX_CONNECTIONS: int = Field(20, validation_alias="REDIS_MAX_CONNECTIONS")

    @property
    def redis_url(self) -> str:
        return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"

    # Pydantic 2.x 验证器语法
    @field_validator('DB_PORT', 'REDIS_PORT')
    @classmethod
    def validate_port(cls, v: int) -> int:
        if not 1 <= v <= 65535:
            raise ValueError('Port must be between 1 and 65535')
        return v

    @field_validator('DB_POOL_SIZE', 'DB_POOL_RECYCLE', 'REDIS_MAX_CONNECTIONS')
    @classmethod
    def validate_positive_int(cls, v: int) -> int:
        if v <= 0:
            raise ValueError('Value must be positive')
        return v

    @field_validator('ROCKETMQ_WAIT_SECONDS')
    @classmethod
    def validate_rocketmq_wait_seconds(cls, v: int) -> int:
        if not 1 <= v <= 30:
            raise ValueError('ROCKETMQ_WAIT_SECONDS must be between 1 and 30')
        return v

    @field_validator('ROCKETMQ_BATCH')
    @classmethod
    def validate_rocketmq_batch(cls, v: int) -> int:
        if not 1 <= v <= 16:
            raise ValueError('ROCKETMQ_BATCH must be between 1 and 16')
        return v

    @field_validator('CONNECTION_TIMEOUT', 'REQUEST_TIMEOUT')
    @classmethod
    def validate_timeouts(cls, v: int) -> int:
        if v <= 0:
            raise ValueError('Timeout must be positive')
        return v


settings = Settings()