CardScoreDataServer/app/core/config.py

from pathlib import Path
from typing import Dict
from typing import ClassVar
import socket
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    BASE_PATH: ClassVar[Path] = Path(__file__).parent.parent.parent.absolute()

    TEST_DATA: str = '环境变量读取测试, 目前使用本地字段'

    @property
    def TEST_DATA_END(self) -> str:
        return f'{self.TEST_DATA}/end'

    SERVER_PORT:int = 7755
    APP_NAME: str = "card-score-data-server"
    APP_ENV: str = "dev"
    LOG_LEVEL: str = "INFO"

    # --- MinIO 配置 ---
    MINIO_ENDPOINT: str = "192.168.77.249:9000"
    MINIO_ACCESS_KEY: str = "pZEwCGnpNN05KPnmC2Yh"
    MINIO_SECRET_KEY: str = "KfJRuWiv9pVxhIMcFqbkv8hZT9SnNTZ6LPx592D4"
    MINIO_SECURE: bool = False  # 是否使用 https
    MINIO_BUCKET: str = "grading"
    MINIO_BASE_PREFIX: str = "score_server_data"

    # DATA_HOST_URL 现在直接指向 MinIO 的前缀路径
    # (注意: 需要在 MinIO 后台将 grading 存储桶的访问权限配置为 Public 或开启特定读策略)
    @property
    def DATA_HOST_URL(self) -> str:
        return f"http://{self.MINIO_ENDPOINT}/{self.MINIO_BUCKET}/{self.MINIO_BASE_PREFIX}"

    # CONFIG_PATH: str = BASE_PATH / 'Config.json'
    API_PREFIX: str = "/api"  # 通用前缀
    SCORE_CONFIG_PATH: Path = BASE_PATH / "app/core/scoring_config.json"

    # 分数计算接口url
    SCORE_UPDATE_SERVER_URL: str = "http://127.0.0.1:7754"

    # --- Fluentd 日志配置 ---
    FLUENTD_ENABLED: bool = False
    FLUENTD_HOST: str = "fluentd"
    FLUENTD_PORT: int = 24224
    FLUENTD_TAG: str = "card_score.server"
    FLUENTD_TAG_PREFIX: str = "python"
    FLUENTD_HOSTNAME: str = ""
    FLUENTD_TIMEOUT_SEC: float = 3.0

    @property
    def FLUENTD_EFFECTIVE_TAG(self) -> str:
        host = self.FLUENTD_HOSTNAME or socket.gethostname()
        return f"{self.FLUENTD_TAG_PREFIX}.{self.FLUENTD_TAG}.{host}"

    @property
    def SCORE_RECALCULATE_ENDPOINT(self) -> str:
        return f"{self.SCORE_UPDATE_SERVER_URL}/api/card_score/score_recalculate"

    # 分数计算配置接口
    @property
    def SCORE_SERVER_CONFIG_URL(self) -> str:
        return f"{self.SCORE_UPDATE_SERVER_URL}/api/config/scoring_config"


    # --- 数据库配置 ---
    DB_NAME: str = 'card_score_gray_database'
    DB_CARD_TABLE_NAME: str = 'cards'
    DB_IMAGE_TABLE_NAME: str = 'card_images'
    DB_GRAY_IMAGE_TABLE_NAME: str = 'card_gray_images'  # 灰度图表名
    DB_USER_CARD_TABLE_NAME: str = 'user_card_bindings'
    RATING_REPORT_HISTORY_TABLE_NAME: str = "rating_report_history"

    DB_USER: str = 'root'
    DB_PASSWORD: str = '123456'
    DB_HOST: str = '127.0.0.1'
    DB_PORT: int = 3306

    @property
    def DATABASE_CONFIG(self) -> Dict[str, str]:
        return {
            'user': self.DB_USER,
            'password': self.DB_PASSWORD,
            'host': self.DB_HOST,
            'port': self.DB_PORT,
        }

    # 连接到指定数据库的配置
    @property
    def DATABASE_CONFIG_WITH_DB(self) -> Dict[str, str]:
        return {
            **self.DATABASE_CONFIG,
            'database': self.DB_NAME
        }

    # 优先级：系统环境变量 > .env 文件
    model_config = SettingsConfigDict(env_file=".env",
                                      env_file_encoding='utf-8')

    def get_full_url(self, path: str) -> str:
        """将相对路径转换为可以直接打开的 MinIO 绝对 URL"""
        if not path:
            return path
        if str(path).startswith("http"):
            return path
        # 移除开头的斜杠防止双斜杠 (如: /Data/xxx -> Data/xxx)
        clean_path = str(path).lstrip("/\\")
        return f"{self.DATA_HOST_URL}/{clean_path}"


settings = Settings()