eryao/backend/src/core/config/settings.py

from __future__ import annotations

from pathlib import Path
from typing import ClassVar, Literal
from urllib.parse import quote

from pydantic import (
    BaseModel,
    Field,
    SecretStr,
    computed_field,
    field_validator,
    model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict


def _resolve_project_root() -> Path:
    current = Path(__file__).resolve()
    for parent in current.parents:
        if (
            (parent / "pyproject.toml").is_file()
            and (parent / "backend").is_dir()
            and (parent / "infra").is_dir()
        ):
            return parent

    for parent in current.parents:
        if parent.name == "backend":
            return parent.parent

    return Path.cwd().resolve()


class RuntimeSettings(BaseModel):
    environment: Literal["dev", "test", "prod"] = "dev"
    service_name: str = "app"
    debug: bool = True
    log_level: str = "INFO"
    log_json: bool = True
    log_rotation: Literal["time", "size", "none"] = "time"
    log_rotation_when: str = "midnight"
    log_rotation_interval: int = 1
    log_rotation_backup_count: int = 14
    log_rotation_max_bytes: int = 10_000_000
    log_dir: str = "logs"
    log_error_dir: str = "logs/errors"
    log_file_name: str = ""
    log_error_file_name: str = ""
    log_sensitive_fields: list[str] = Field(
        default_factory=lambda: [
            "password",
            "secret",
            "token",
            "api_key",
            "authorization",
            "cookie",
            "client_ip",
            "user_id",
        ]
    )
    sql_log_queries: bool = False
    trusted_proxy_ips: list[str] = Field(default_factory=list)

    @field_validator("log_dir", mode="before")
    @classmethod
    def lock_log_dir(cls, _: object) -> str:
        return "logs"

    @field_validator("log_error_dir", mode="before")
    @classmethod
    def lock_log_error_dir(cls, _: object) -> str:
        return "logs/errors"

    @model_validator(mode="after")
    def ensure_service_scoped_log_file_names(self) -> "RuntimeSettings":
        service = "".join(
            char if char.isalnum() or char in {"-", "_"} else "-"
            for char in self.service_name
        ).strip("-_")
        service_name = service or "app"

        if not self.log_file_name.strip():
            self.log_file_name = f"{service_name}.log"
        if not self.log_error_file_name.strip():
            self.log_error_file_name = f"{service_name}.error.log"

        return self


class CorsSettings(BaseModel):
    allow_origins: list[str] = Field(
        default_factory=lambda: [
            "http://localhost",
            "http://localhost:3000",
        ]
    )
    allow_credentials: bool = True
    allow_methods: list[str] = Field(default_factory=lambda: ["*"])
    allow_headers: list[str] = Field(default_factory=lambda: ["*"])


class RedisSettings(BaseModel):
    host: str = "redis"
    port: int = 6379
    password: str | None = None
    db: int = 0
    socket_connect_timeout: float = 1.0
    socket_timeout: float = 1.0
    max_connections: int = 10

    @computed_field
    @property
    def url(self) -> str:
        if self.password:
            password = quote(self.password, safe="")
            return f"redis://:{password}@{self.host}:{self.port}/{self.db}"
        return f"redis://{self.host}:{self.port}/{self.db}"


class SupabaseSettings(BaseModel):
    public_url: str
    anon_key: str = "CHANGE_ME"
    service_role_key: str = "CHANGE_ME"
    jwt_secret: SecretStr | None = Field(default=None, exclude=True)
    jwt_algorithm: Literal["HS256"] = "HS256"
    jwt_issuer: str | None = None

    @model_validator(mode="after")
    def compute_defaults(self) -> "SupabaseSettings":
        base = str(self.public_url).rstrip("/")
        if self.jwt_issuer is None:
            self.jwt_issuer = f"{base}/auth/v1"

        return self

    @computed_field
    @property
    def url(self) -> str:
        return str(self.public_url)


class StorageSettings(BaseModel):
    provider: Literal["supabase"] = "supabase"
    signed_url_ttl_seconds: int = Field(default=600, ge=60, le=3600)
    retention_days: int = Field(default=30, ge=1, le=3650)

    class AttachmentSettings(BaseModel):
        bucket: str = Field(default="eryao-attachments", min_length=3, max_length=63)
        max_size_mb: int = Field(default=20, ge=1, le=200)

    class AvatarSettings(BaseModel):
        bucket: str = Field(default="avatars", min_length=3, max_length=63)
        max_size_mb: int = Field(default=2, ge=1, le=10)

    class FeedbackSettings(BaseModel):
        bucket: str = Field(default="feedback-images", min_length=3, max_length=63)
        max_size_mb: int = Field(default=5, ge=1, le=20)

    attachment: AttachmentSettings = Field(default_factory=AttachmentSettings)
    avatar: AvatarSettings = Field(default_factory=AvatarSettings)
    feedback: FeedbackSettings = Field(default_factory=FeedbackSettings)


class LlmSettings(BaseModel):
    provider_keys: dict[str, str] = Field(default_factory=dict)


class DatabaseSettings(BaseModel):
    host: str = "localhost"
    port: int = 5432
    name: str = "postgres"
    user: str = "postgres"
    password: str = "CHANGE_ME"

    @computed_field
    @property
    def url(self) -> str:
        password = quote(self.password, safe="")
        return (
            f"postgresql+asyncpg://{self.user}:{password}"
            f"@{self.host}:{self.port}/{self.name}"
        )


class SensitiveWordSettings(BaseModel):
    use_aliyun: bool = True
    fallback_to_local: bool = True


class TestSettings(BaseModel):
    phone: str = ""
    password: str = ""


class TaskiqSettings(BaseModel):
    broker_url: str | None = None
    result_backend_url: str | None = None


class AgentRuntimeSettings(BaseModel):
    redis_stream_prefix: str = "agent:events"
    redis_stream_read_count: int = 100
    redis_stream_block_ms: int = 30000
    user_context_cache_prefix: str = "agent:user-context"
    user_context_cache_ttl_seconds: int = 86400
    user_context_cache_max_turns: int = 100
    context_messages_cache_prefix: str = "agent:context-messages"
    context_messages_cache_ttl_seconds: int = 86400
    attachment_content_cache_prefix: str = "agent:attachment-content"
    attachment_content_cache_ttl_seconds: int = 86400
    attachment_content_cache_max_base64_bytes: int = 262144


class PointsPolicySettings(BaseModel):
    register_bonus_points: int = Field(default=60, ge=0, le=1_000_000)
    register_bonus_hmac_key: SecretStr = SecretStr("")

    @model_validator(mode="after")
    def validate_hmac_key(self) -> "PointsPolicySettings":
        key = self.register_bonus_hmac_key.get_secret_value().strip()
        if not key:
            raise ValueError("points_policy.register_bonus_hmac_key must not be empty")
        if key.upper() == "CHANGE_ME":
            raise ValueError(
                "points_policy.register_bonus_hmac_key must not be CHANGE_ME"
            )
        return self


class AppleIapSettings(BaseModel):
    bundle_id: str = Field(default="com.meeyao.qianwen", min_length=1)
    root_cert_url: str = "https://www.apple.com/certificateauthority/AppleIncRootCertificate.cer"
    jws_x5c_cert_url: str = "https://api.storekit.itunes.apple.com/v1/verificationKeys"
    server_api_issuer_id: str | None = None
    server_api_key_id: str | None = None
    server_api_private_key: SecretStr | None = None
    sandbox_tester_email: str | None = None
    sandbox_tester_password: SecretStr | None = None
    server_notifications_url: str | None = None


def _resolve_env_file() -> str:
    current = Path(__file__).resolve()
    for parent in [current, *current.parents]:
        candidate = parent / ".env"
        if candidate.is_file():
            return str(candidate)
    return ".env"


PROJECT_ROOT = _resolve_project_root()


class FeedbackReportSettings(BaseModel):
    email: str = Field(default="support@example.com", description="客服邮箱")
    cron: str = Field(default="0 10 * * *", description="报告生成cron表达式")
    enabled: bool = Field(default=False, description="是否启用报告推送")


class EmailSettings(BaseModel):
    host: str = Field(default="smtp.feishu.cn", description="SMTP 服务器地址")
    port: int = Field(default=465, ge=1, le=65535, description="SMTP 端口")
    username: str = Field(default="", description="SMTP 用户名")
    password: SecretStr = Field(default=SecretStr(""), description="SMTP 密码")
    use_ssl: bool = Field(default=True, description="是否使用 SSL")
    from_address: str = Field(default="noreply@example.com", description="发件人地址")
    from_name: str = Field(default="Eryao Feedback", description="发件人显示名称")


class Settings(BaseSettings):
    runtime: RuntimeSettings = RuntimeSettings()
    cors: CorsSettings = CorsSettings()
    redis: RedisSettings = RedisSettings()
    supabase: SupabaseSettings = Field(
        default_factory=lambda: SupabaseSettings(public_url="http://localhost:8001")
    )
    storage: StorageSettings = StorageSettings()
    llm: LlmSettings = LlmSettings()
    database: DatabaseSettings = DatabaseSettings()
    sensitive_word: SensitiveWordSettings = Field(default_factory=SensitiveWordSettings)
    test: TestSettings = Field(default_factory=TestSettings)
    taskiq: TaskiqSettings = Field(default_factory=TaskiqSettings)
    agent_runtime: AgentRuntimeSettings = Field(default_factory=AgentRuntimeSettings)
    points_policy: PointsPolicySettings = Field(default_factory=PointsPolicySettings)
    apple_iap: AppleIapSettings = Field(default_factory=AppleIapSettings)
    feedback_report: FeedbackReportSettings = Field(
        default_factory=FeedbackReportSettings
    )
    email: EmailSettings = Field(default_factory=EmailSettings)

    @computed_field
    @property
    def database_url(self) -> str:
        return self.database.url

    @computed_field
    @property
    def taskiq_broker_url(self) -> str:
        return self.taskiq.broker_url or self.redis.url

    @computed_field
    @property
    def taskiq_result_backend_url(self) -> str:
        return self.taskiq.result_backend_url or self.redis.url

    model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
        env_file=_resolve_env_file(),
        env_prefix="ERYAO_",
        env_nested_delimiter="__",
        case_sensitive=False,
        extra="ignore",
    )


config = Settings()  # type: ignore[reportCallIssue]