demand_server/app/core/config.py

from urllib.parse import quote_plus

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "denamd-server"
    app_env: str = "dev"
    app_host: str = "0.0.0.0"
    app_port: int = 8000
    cors_allow_origins: str = "*"
    scheduler_heartbeat_seconds: int = 3600
    mysql_host: str = ""
    mysql_port: int = 3306
    mysql_user: str = "root"
    mysql_password: str = ""
    mysql_database: str = ""
    # 新热事件来源库（hot_content_* 表），与需求池库分离
    hot_content_mysql_host: str = ""
    hot_content_mysql_port: int = 3306
    hot_content_mysql_user: str = ""
    hot_content_mysql_password: str = ""
    hot_content_mysql_database: str = ""
    hot_content_mysql_charset: str = "utf8mb4"
    odps_access_id: str = "LTAI9EBa0bd5PrDa"
    odps_access_key: str = "vAalxds7YxhfOA2yVv8GziCg3Y87v5"
    odps_project: str = "loghubods"
    odps_endpoint: str = "http://service.odps.aliyun.com/api"
    demand_pool_source_table: str = "dwd_multi_demand_pool_di"
    demand_pool_secondary_source_table: str = "dwd_demand_pool_di"
    # 次源 dwd_demand_pool_di 同步开关；false 时仅同步主源 dwd_multi_demand_pool_di
    demand_pool_secondary_sync_enabled: bool = False
    demand_pool_secondary_strategy: str = "近期需求"
    demand_pool_secondary_default_ext_info: str = "{}"
    demand_pool_target_table: str = "multi_demand_pool_di"
    demand_pool_initial_partitions: str = "20260507,20260508,20260509"
    demand_pool_hourly_sync_enabled: bool = True
    demand_pool_hourly_sync_minute: int = 0
    demand_pool_daily_strategy_alert_enabled: bool = True
    demand_pool_daily_strategy_alert_hour: int = 10
    demand_pool_daily_strategy_alert_minute: int = 0
    feishu_webhook_url: str = ""
    feishu_webhook_timeout_seconds: int = 30
    # 默认不校验 HTTPS 证书，避免公司代理自签链导致发不出消息；生产若需严格校验可设为 true
    feishu_webhook_verify_ssl: bool = False
    hot_demand_pool_strategy: str = "新热事件"
    hot_content_wxindex_threshold: float = 1_000_000.0

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    @property
    def mysql_dsn(self) -> str:
        return (
            "mysql+pymysql://"
            f"{quote_plus(self.mysql_user)}:{quote_plus(self.mysql_password)}"
            f"@{self.mysql_host}:{self.mysql_port}/{quote_plus(self.mysql_database)}"
        )

    @property
    def hot_content_mysql_configured(self) -> bool:
        return bool(self.hot_content_mysql_host.strip() and self.hot_content_mysql_database.strip())

    @property
    def hot_content_mysql_dsn(self) -> str:
        charset = quote_plus(self.hot_content_mysql_charset or "utf8mb4")
        return (
            "mysql+pymysql://"
            f"{quote_plus(self.hot_content_mysql_user)}:"
            f"{quote_plus(self.hot_content_mysql_password)}"
            f"@{self.hot_content_mysql_host}:{self.hot_content_mysql_port}/"
            f"{quote_plus(self.hot_content_mysql_database)}?charset={charset}"
        )

    @property
    def demand_pool_initial_partition_list(self) -> list[str]:
        return [
            partition.strip()
            for partition in self.demand_pool_initial_partitions.split(",")
            if partition.strip()
        ]

    @property
    def cors_allow_origin_list(self) -> list[str]:
        if self.cors_allow_origins.strip() == "*":
            return ["*"]
        return [
            origin.strip()
            for origin in self.cors_allow_origins.split(",")
            if origin.strip()
        ]


settings = Settings()