AutoScraperX/config/base.py

import os

from dotenv import load_dotenv
from core.utils.path_utils import project_root,log_dir
from pydantic import BaseSettings, Field, AnyUrl

# 在 Settings 类之前加载 .env 文件

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


class Settings(BaseSettings):
    # 环境标识
    ENV: str = "prod"  # prod/dev
    ENABLE_ALIYUN_LOG: bool = True

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

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

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

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

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

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

    class Config:
        env_file = ".env"
        env_file_encoding = 'utf-8'
        case_sensitive = False

    # @property
    # def database_url(self) -> str:
    #     """生成安全的数据库连接字符串"""
    #     return f"mysql+asyncmy://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}?charset={self.DB_CHARSET}"


settings = Settings()