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"
    # 当下供需gap-分词来源库（demand_content 表）
    supply_mysql_host: str = ""
    supply_mysql_port: int = 3306
    supply_mysql_user: str = ""
    supply_mysql_password: str = ""
    supply_mysql_database: str = ""
    supply_mysql_charset: str = "utf8mb4"
    supply_demand_content_table: str = "demand_content"
    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
    strategy_config_table: str = "strategy_config"
    strategy_staging_table: str = "strategy_staging"
    strategy_staging_hourly_generate_enabled: bool = True
    strategy_staging_hourly_generate_start_hour: int = 3
    strategy_staging_hourly_generate_end_hour: int = 23
    strategy_staging_hourly_generate_minute: int = 0
    substance_element_base_table: str = "substance_element_base"
    substance_element_effect_table: str = "substance_element_effect_di"
    substance_element_daily_sync_enabled: bool = True
    substance_element_daily_sync_hour: int = 9
    substance_element_daily_sync_minute: int = 30
    vertical_category_base_table: str = "vertical_category_base"
    vertical_category_effect_table: str = "vertical_category_effect_di"
    vertical_category_daily_sync_enabled: bool = True
    vertical_category_daily_sync_hour: int = 9
    vertical_category_daily_sync_minute: int = 30

    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 supply_mysql_configured(self) -> bool:
        return bool(self.supply_mysql_host.strip() and self.supply_mysql_database.strip())

    @property
    def supply_mysql_dsn(self) -> str:
        charset = quote_plus(self.supply_mysql_charset or "utf8mb4")
        return (
            "mysql+pymysql://"
            f"{quote_plus(self.supply_mysql_user)}:"
            f"{quote_plus(self.supply_mysql_password)}"
            f"@{self.supply_mysql_host}:{self.supply_mysql_port}/"
            f"{quote_plus(self.supply_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 strategy_staging_hourly_cron_hours(self) -> str:
        """每小时触发，不跨午夜。默认 start=3、end=23 → '3-23'（末次 23:00，0 点不跑）。"""
        start = self.strategy_staging_hourly_generate_start_hour
        end = self.strategy_staging_hourly_generate_end_hour
        if not (0 <= start <= 23 and 0 <= end <= 23):
            raise ValueError("strategy staging hourly hours must be 0-23")
        if start > end:
            raise ValueError("start_hour cannot be greater than end_hour")
        if start == end:
            return str(start)
        return f"{start}-{end}"

    @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()