import json import os from functools import lru_cache from typing import Annotated, Any, Literal from pydantic import Field, ValidationInfo, field_validator, model_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict from app.core.constants import ( ActorValue, ConfigErrorDetail, DEFAULT_MODEL_PROVIDER, DEFAULT_OPENCLAW_ACTION_JSON, ) from app.core.database.safety import database_password, database_target _DOTENV_DISABLED = os.getenv("COMPANY_AI_DISABLE_DOTENV", "").strip().lower() in { "1", "true", "yes", "on", } _DOTENV_FILE = None if _DOTENV_DISABLED else ".env" _PRODUCTION_SECRET_MIN_LENGTH = 24 _DATABASE_PASSWORD_MIN_LENGTH = 10 _PLACEHOLDER_SECRET_PARTS = ( "change-me", "changeme", "example", "placeholder", "replace-with", "your-", ) def _is_unsafe_production_secret(value: str) -> bool: normalized = value.strip().casefold() return ( len(normalized) < _PRODUCTION_SECRET_MIN_LENGTH or any(part in normalized for part in _PLACEHOLDER_SECRET_PARTS) ) def _is_unsafe_database_password(value: str) -> bool: normalized = value.strip().casefold() return ( len(normalized) < _DATABASE_PASSWORD_MIN_LENGTH or any(part in normalized for part in _PLACEHOLDER_SECRET_PARTS) ) class Settings(BaseSettings): """Runtime settings loaded from the environment and optional `.env` file.""" model_config = SettingsConfigDict( env_file=_DOTENV_FILE, env_file_encoding="utf-8", extra="ignore", hide_input_in_errors=True, ) app_name: str = "Company AI Management Platform" app_env: str = "local" debug: bool = False read_only_mode: bool = True api_prefix: str = "/api/v1" api_key: str | None = None api_actor: str = ActorValue.API api_keys: list[dict[str, Any]] = Field(default_factory=list) audit_api_key: str | None = None audit_api_actor: str = ActorValue.AUDITOR audit_api_keys: list[dict[str, Any]] = Field(default_factory=list) cors_origins: list[str] = Field(default_factory=lambda: ["*"]) mask_sensitive_responses: bool = True masked_response_fields: list[str] = Field(default_factory=list) database_url: str = "sqlite:///./company_ai.db" legacy_database_url: str | None = None legacy_project_query: str | None = None legacy_task_query: str | None = None legacy_allowed_queries: dict[str, str] = Field(default_factory=dict) legacy_project_code_prefix: str = "LEGACY" legacy_task_code_prefix: str = "LEGACY-TASK" redis_url: str = "redis://127.0.0.1:6379/0" feishu_base_url: str = "https://open.feishu.cn/open-apis" feishu_app_id: str | None = None feishu_app_secret: str | None = None feishu_app_type: Literal["self", "store"] = "self" feishu_app_ticket: str | None = None feishu_verification_token: str | None = None feishu_encrypt_key: str | None = None feishu_default_chat_id: str | None = None feishu_default_tenant_key: str | None = None feishu_event_transport: Literal[ "disabled", "webhook", "long_connection", ] = "disabled" feishu_user_features_enabled: bool = False feishu_admin_identities: Annotated[list[str], NoDecode] = Field(default_factory=list) model_provider: str = DEFAULT_MODEL_PROVIDER openclaw_base_url: str = "http://127.0.0.1:2070" openclaw_http_url: str | None = None openclaw_ws_url: str | None = None openclaw_api_key: str | None = None openclaw_gateway_token: str | None = None openclaw_allowed_tools: Annotated[list[str], NoDecode] = Field(default_factory=list) openclaw_allowed_actions: Annotated[list[str], NoDecode] = Field( default_factory=lambda: [DEFAULT_OPENCLAW_ACTION_JSON] ) hermes_base_url: str = "http://127.0.0.1:2073/v1" hermes_api_key: str | None = None hermes_model: str = "hermes-agent" hermes_session_id: str | None = None direct_llm_base_url: str = "https://api.openai.com/v1" direct_llm_api_key: str | None = None direct_llm_model: str = "gpt-4.1-mini" scheduler_enabled: bool = False task_queue_enabled: bool = False task_queue_always_eager: bool = False lifecycle_pipeline_enabled: bool = False finance_needs_enabled: bool = False market_analysis_enabled: bool = False market_data_provider: str = "tushare" market_data_base_url: str = "https://api.tushare.pro" market_data_token: str | None = None market_premarket_cron_hour: int = 8 market_premarket_cron_minute: int = 30 market_close_cron_hour: int = 15 market_close_cron_minute: int = 30 market_weekly_day_of_week: str = "sun" market_weekly_cron_hour: int = 20 market_weekly_cron_minute: int = 0 legacy_sync_enabled: bool = False celery_result_backend_url: str | None = None daily_brief_cron_hour: int = 9 daily_brief_cron_minute: int = 0 attendance_summary_cron_hour: int = 18 attendance_summary_cron_minute: int = 0 risk_progress_cron_hour: int = 17 risk_progress_cron_minute: int = 30 work_daily_cron_hour: int = 18 work_daily_cron_minute: int = 30 work_weekly_report_day_of_week: str = "fri" work_weekly_cron_hour: int = 18 work_weekly_cron_minute: int = 45 weekly_project_report_day_of_week: str = "mon" weekly_project_report_cron_hour: int = 9 weekly_project_report_cron_minute: int = 30 legacy_project_sync_cron_hour: int = 2 legacy_project_sync_cron_minute: int = 0 legacy_task_sync_cron_hour: int = 2 legacy_task_sync_cron_minute: int = 30 event_dispatch_enabled: bool = True event_dispatch_batch_size: int = 100 event_dispatch_max_attempts: int = 3 event_dispatch_retry_delay_seconds: int = 300 event_dispatch_lock_seconds: int = 300 event_dispatch_cron_minute: str = "*/5" heartbeat_interval_seconds: int = 60 heartbeat_retention_seconds: int = Field(default=86400, ge=1) ai_memory_enabled: bool = True ai_memory_auto_write_enabled: bool = True ai_memory_recall_limit: int = 5 ai_memory_auto_write_ttl_days: int = Field(default=90, ge=1, le=3650) ai_memory_blocked_content_terms: list[str] = Field( default_factory=lambda: [ "bank account", "budget", "cash flow", "financial statement", "id card", "payment", "salary", "成本", "付款", "工资", "收款", "现金流", "财务", "预算", ] ) ai_analysis_max_attempts: int = 3 ai_memory_forbidden_keys: list[str] = Field( default_factory=lambda: [ "authorization", "app_access_token", "app_ticket", "api_key", "apikey", "access_token", "tenant_access_token", "token", "secret", "password", "openclaw_gateway_token", "hermes_api_key", "direct_llm_api_key", "market_data_token", "feishu_app_secret", "feishu_app_ticket", "feishu_verification_token", ] ) @field_validator("cors_origins", mode="before") @classmethod def parse_cors_origins(cls, value: Any) -> list[str]: if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] if value is None: return [] text = str(value).strip() if text.startswith("["): data = json.loads(text) if not isinstance(data, list): raise ValueError(ConfigErrorDetail.CORS_ORIGINS_FORMAT) return [str(item).strip() for item in data if str(item).strip()] return [item.strip() for item in text.split(",") if item.strip()] @field_validator( "feishu_app_type", "feishu_event_transport", mode="before", ) @classmethod def normalize_feishu_choice(cls, value: Any, info: ValidationInfo) -> str: default = "self" if info.field_name == "feishu_app_type" else "disabled" return str(value or default).strip().lower() @field_validator( "openclaw_allowed_tools", "openclaw_allowed_actions", "ai_memory_forbidden_keys", "ai_memory_blocked_content_terms", "feishu_admin_identities", mode="before", ) @classmethod def parse_csv_list(cls, value: str | list[str] | None) -> list[str]: if value is None: return [] if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] text = value.strip() if not text: return [] if text.startswith("["): data = json.loads(text) if not isinstance(data, list): raise ValueError(ConfigErrorDetail.CORS_ORIGINS_FORMAT) return [str(item).strip() for item in data if str(item).strip()] return [item.strip() for item in text.split(",") if item.strip()] @field_validator("masked_response_fields", mode="before") @classmethod def parse_masked_response_fields(cls, value: Any) -> list[str]: if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] if value is None: return [] text = str(value).strip() if not text: return [] if text.startswith("["): data = json.loads(text) if not isinstance(data, list): raise ValueError(ConfigErrorDetail.CORS_ORIGINS_FORMAT) return [str(item).strip() for item in data if str(item).strip()] return [item.strip() for item in text.split(",") if item.strip()] @field_validator("api_keys", "audit_api_keys", mode="before") @classmethod def parse_service_keys(cls, value: Any) -> list[dict[str, Any]]: if value is None or value == "": return [] if isinstance(value, list): return [dict(item) for item in value if isinstance(item, dict)] if isinstance(value, str): data = json.loads(value) if not isinstance(data, list): raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT) if not all(isinstance(item, dict) for item in data): raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT) return [dict(item) for item in data] raise ValueError(ConfigErrorDetail.SERVICE_KEYS_FORMAT) @field_validator("legacy_allowed_queries", mode="before") @classmethod def parse_legacy_allowed_queries(cls, value: Any) -> dict[str, str]: if value is None or value == "": return {} if isinstance(value, dict): return {str(key): str(item) for key, item in value.items()} if isinstance(value, str): data = json.loads(value) if not isinstance(data, dict): raise ValueError(ConfigErrorDetail.LEGACY_ALLOWED_QUERIES_FORMAT) return {str(key): str(item) for key, item in data.items()} raise ValueError(ConfigErrorDetail.LEGACY_ALLOWED_QUERIES_FORMAT) @model_validator(mode="after") def validate_production_safety(self) -> "Settings": if self.app_env.lower() not in {"prod", "production"}: return self def _enabled_keys( legacy_key: str | None, configured_keys: list[dict[str, Any]], ) -> set[str]: values: set[str] = set() if legacy_key and legacy_key.strip(): values.add(legacy_key) for item in configured_keys: enabled = item.get("enabled", True) if not isinstance(enabled, bool): enabled = str(enabled).strip().lower() not in { "0", "false", "no", "off", "disabled" } key = item.get("key") if enabled and key is not None and str(key).strip(): values.add(str(key)) return values errors: list[str] = [] api_key_values = _enabled_keys(self.api_key, self.api_keys) audit_key_values = _enabled_keys(self.audit_api_key, self.audit_api_keys) platform_target = database_target(self.database_url) legacy_target = database_target(self.legacy_database_url) if platform_target is None or platform_target[0] != "postgresql": errors.append("DATABASE_URL must use PostgreSQL in production") platform_password = database_password(self.database_url) if platform_password and _is_unsafe_database_password(platform_password): errors.append( "DATABASE_URL password must use a non-placeholder value of at least " f"{_DATABASE_PASSWORD_MIN_LENGTH} characters in production" ) if platform_target is not None and platform_target == legacy_target: errors.append( "DATABASE_URL and LEGACY_DATABASE_URL must target different databases" ) if not api_key_values: errors.append("API_KEY or API_KEYS is required in production") elif any(_is_unsafe_production_secret(value) for value in api_key_values): errors.append( "API_KEY/API_KEYS must use non-placeholder values of at least " f"{_PRODUCTION_SECRET_MIN_LENGTH} characters in production" ) if not audit_key_values: errors.append("AUDIT_API_KEY or AUDIT_API_KEYS is required in production") elif any(_is_unsafe_production_secret(value) for value in audit_key_values): errors.append( "AUDIT_API_KEY/AUDIT_API_KEYS must use non-placeholder values of at least " f"{_PRODUCTION_SECRET_MIN_LENGTH} characters in production" ) if api_key_values & audit_key_values: errors.append( "API_KEY/API_KEYS and AUDIT_API_KEY/AUDIT_API_KEYS cannot overlap in production" ) if "*" in self.cors_origins: errors.append("CORS_ORIGINS cannot contain '*' in production") if self.debug: errors.append("DEBUG must be false in production") if not self.mask_sensitive_responses: errors.append("MASK_SENSITIVE_RESPONSES must be true in production") if not self.read_only_mode: errors.append("READ_ONLY_MODE must be true in production") if self.feishu_event_transport != "disabled": if not self.feishu_app_id or not self.feishu_app_secret: errors.append( "FEISHU_APP_ID and FEISHU_APP_SECRET are required when " "Feishu event transport is enabled" ) elif _is_unsafe_production_secret(self.feishu_app_secret): errors.append( "FEISHU_APP_SECRET must use a non-placeholder value of at least " f"{_PRODUCTION_SECRET_MIN_LENGTH} characters in production" ) if ( self.feishu_event_transport == "webhook" and not self.feishu_verification_token ): errors.append( "FEISHU_VERIFICATION_TOKEN is required for Feishu webhook transport" ) elif ( self.feishu_event_transport == "webhook" and self.feishu_verification_token and _is_unsafe_production_secret(self.feishu_verification_token) ): errors.append( "FEISHU_VERIFICATION_TOKEN must use a non-placeholder value of at " f"least {_PRODUCTION_SECRET_MIN_LENGTH} characters in production" ) if self.feishu_user_features_enabled: if not self.feishu_admin_identities: errors.append( "FEISHU_ADMIN_IDENTITIES is required when Feishu user features are enabled" ) else: try: from app.modules.feishu_users.constants import ( parse_admin_identities, ) parse_admin_identities(self.feishu_admin_identities) except ValueError as exc: errors.append(str(exc)) if self.feishu_event_transport == "disabled": errors.append( "FEISHU_EVENT_TRANSPORT must be webhook or long_connection " "when Feishu user features are enabled" ) if errors: raise ValueError("; ".join(errors)) return self @lru_cache def get_settings() -> Settings: """Return cached application settings.""" return Settings()