config.py 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. from pathlib import Path
  2. from typing import Dict, List
  3. from enum import Enum
  4. from typing import ClassVar
  5. import json
  6. from pydantic_settings import BaseSettings, SettingsConfigDict
  7. class Settings(BaseSettings):
  8. BASE_PATH: ClassVar[Path] = Path(__file__).parent.parent.parent.absolute()
  9. TEST_DATA: str = '环境变量读取测试, 目前使用本地字段'
  10. @property
  11. def TEST_DATA_END(self) -> str:
  12. return f'{self.TEST_DATA}/end'
  13. SERVER_PORT:int = 7755
  14. APP_NAME: str = "card-score-data-server"
  15. APP_ENV: str = "dev"
  16. # --- MinIO 配置 ---
  17. MINIO_ENDPOINT: str = "192.168.77.249:9000"
  18. MINIO_ACCESS_KEY: str = "pZEwCGnpNN05KPnmC2Yh"
  19. MINIO_SECRET_KEY: str = "KfJRuWiv9pVxhIMcFqbkv8hZT9SnNTZ6LPx592D4"
  20. MINIO_SECURE: bool = False # 是否使用 https
  21. MINIO_BUCKET: str = "grading"
  22. MINIO_BASE_PREFIX: str = "score_server_data"
  23. # DATA_HOST_URL 现在直接指向 MinIO 的前缀路径
  24. # (注意: 需要在 MinIO 后台将 grading 存储桶的访问权限配置为 Public 或开启特定读策略)
  25. @property
  26. def DATA_HOST_URL(self) -> str:
  27. return f"http://{self.MINIO_ENDPOINT}/{self.MINIO_BUCKET}/{self.MINIO_BASE_PREFIX}"
  28. # CONFIG_PATH: str = BASE_PATH / 'Config.json'
  29. API_PREFIX: str = "/api" # 通用前缀
  30. SCORE_CONFIG_PATH: Path = BASE_PATH / "app/core/scoring_config.json"
  31. # 分数计算接口url
  32. SCORE_UPDATE_SERVER_URL: str = "http://127.0.0.1:7754"
  33. # --- Fluentd 日志配置 ---
  34. FLUENTD_ENABLED: bool = False
  35. FLUENTD_HOST: str = "127.0.0.1"
  36. FLUENTD_PORT: int = 24224
  37. FLUENTD_TAG: str = "card_score.server"
  38. FLUENTD_TIMEOUT_SEC: float = 3.0
  39. @property
  40. def SCORE_RECALCULATE_ENDPOINT(self) -> str:
  41. return f"{self.SCORE_UPDATE_SERVER_URL}/api/card_score/score_recalculate"
  42. # 分数计算配置接口
  43. @property
  44. def SCORE_SERVER_CONFIG_URL(self) -> str:
  45. return f"{self.SCORE_UPDATE_SERVER_URL}/api/config/scoring_config"
  46. # --- 数据库配置 ---
  47. DB_NAME: str = 'card_score_gray_database'
  48. DB_CARD_TABLE_NAME: str = 'cards'
  49. DB_IMAGE_TABLE_NAME: str = 'card_images'
  50. DB_GRAY_IMAGE_TABLE_NAME: str = 'card_gray_images' # 灰度图表名
  51. DB_USER_CARD_TABLE_NAME: str = 'user_card_bindings'
  52. RATING_REPORT_HISTORY_TABLE_NAME: str = "rating_report_history"
  53. DB_USER: str = 'root'
  54. DB_PASSWORD: str = '123456'
  55. DB_HOST: str = '127.0.0.1'
  56. DB_PORT: int = 3306
  57. @property
  58. def DATABASE_CONFIG(self) -> Dict[str, str]:
  59. return {
  60. 'user': self.DB_USER,
  61. 'password': self.DB_PASSWORD,
  62. 'host': self.DB_HOST,
  63. 'port': self.DB_PORT,
  64. }
  65. # 连接到指定数据库的配置
  66. @property
  67. def DATABASE_CONFIG_WITH_DB(self) -> Dict[str, str]:
  68. return {
  69. **self.DATABASE_CONFIG,
  70. 'database': self.DB_NAME
  71. }
  72. # 优先级:系统环境变量 > .env 文件
  73. model_config = SettingsConfigDict(env_file=".env",
  74. env_file_encoding='utf-8')
  75. def get_full_url(self, path: str) -> str:
  76. """将相对路径转换为可以直接打开的 MinIO 绝对 URL"""
  77. if not path:
  78. return path
  79. if str(path).startswith("http"):
  80. return path
  81. # 移除开头的斜杠防止双斜杠 (如: /Data/xxx -> Data/xxx)
  82. clean_path = str(path).lstrip("/\\")
  83. return f"{self.DATA_HOST_URL}/{clean_path}"
  84. settings = Settings()
  85. print("测试读取env", settings.TEST_DATA)
  86. print("测试拼接env", settings.TEST_DATA_END)