import logging from typing import Dict, Any import yaml from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict from v2.nacos import ClientConfigBuilder, GRPCConfig, NacosConfigService, ConfigParam, NacosNamingService, RegisterInstanceParam, DeregisterInstanceParam logger = logging.getLogger(__name__) # ====================== Nacos 配置 ====================== NACOS_SERVER_ADDRESSES = "18.167.251.121:28848" NACOS_NAMESPACE = "zcr" NACOS_USERNAME = "nacos" NACOS_PASSWORD = "Aidlab123123!" NACOS_GROUP = "LOCAL" NACOS_DATA_ID = "aida.python" SERVICE_NAME = "fastapi-service" # ←←← 必须修改！建议格式：项目名-环境，例如 ai-image-service-dev class Settings(BaseSettings): """ 应用配置类。Pydantic Settings 会自动从环境变量和 .env 文件中加载这些值。 """ model_config = SettingsConfigDict( env_file='.env', env_file_encoding='utf-8', # extra='ignore' # 忽略环境变量中多余的键 ) # --- 服务端口配置信息 --- PORT: int = Field(default=8001, description="") # --- 服务环境 配置信息 --- SERVE_ENV: str = Field(default='', description="") # --- 开发状态 配置信息 --- DEBUG: bool = Field(default=False, description="") # --- 千问api 配置信息 --- QWEN_API_KEY: str = Field(default="", description="") # --- ComfyUI 配置信息 --- COMFYUI_SERVER_ADDRESS: str = Field(default='', description="") # --- minio 配置信息 --- MINIO_URL: str = Field(default='', description="") MINIO_ACCESS: str = Field(default='', description="") MINIO_SECRET: str = Field(default='', description="") MINIO_SECURE: bool = Field(default=True, description="") # --- redis 配置信息 --- REDIS_HOST: str = Field(default='', description="") REDIS_PORT: str = Field(default='', description="") REDIS_DB: int = Field(default=0, description="") # --- mysql 配置信息 --- MYSQL_HOST: str = Field(default='', description="") MYSQL_PORT: int = Field(default=3306, description="") MYSQL_USER: str = Field(default='', description="") MYSQL_PASSWORD: str = Field(default='', description="") MYSQL_DB: str = Field(default='', description="") MYSQL_CHARSET: str = Field(default='utf8mb4', description="") # --- rabbit-mq 配置信息 --- MQ_HOST: str = Field(default='', description="") MQ_PORT: str = Field(default='', description="") MQ_USERNAME: str = Field(default='', description="") MQ_PASSWORD: str = Field(default='', description="") MQ_VIRTUAL_HOST: str = Field(default='/', description="") MQ_ENV: str = Field(default='', description="") # --- milvus 配置信息 --- MILVUS_URL: str = Field(default='', description="") MILVUS_TOKEN: str = Field(default='', description="") MILVUS_ALIAS: str = Field(default='', description="") # --- ollama 配置信息 --- CHROMADB_PATH: str = Field(default='', description="") # --- ollama 配置信息 --- OLLAMA_URL: str = Field(default='', description="") # --- Design Callback Java 接口 --- JAVA_STREAM_API_URL: str = Field(default='', description="") # --- flux2 klein model url --- FLUX2_GEN_IMG_MODEL_URL: str = Field(default='', description="") # --- 服务器IP --- A6000_SERVICE_HOST: str = Field(default='', description="") B_4_X_4090_SERVICE_HOST: str = Field(default='', description="") # --- 其他配置信息 以下均为Docker容器内配置--- LOGS_PATH: str = Field(default="/logs/", description="") CATEGORY_PATH: str = Field(default="/app/service/attribute/config/descriptor/category/category_dis.csv", description="") SEG_CACHE_PATH: str = Field(default="/seg_cache/", description="") RECOMMEND_PATH_PREFIX: str = Field(default="/app/service/recommend/", description="") SERVE_PORT: int = Field(default=2010, description="") settings = Settings() # ====================== Nacos 配置管理 ====================== client_config = (ClientConfigBuilder() .server_address(NACOS_SERVER_ADDRESSES) .username(NACOS_USERNAME) .password(NACOS_PASSWORD) .namespace_id(NACOS_NAMESPACE) .log_level('INFO') .grpc_config(GRPCConfig(grpc_timeout=50000)) .build()) # ====================== Nacos 配置管理 ====================== nacos_config_data: Dict[str, Any] = {} async def load_nacos_config() -> None: """初始化 Nacos 配置并监听变化""" global nacos_config_data, settings global nacos_initialized_successfully try: client = await NacosConfigService.create_config_service(client_config) # 1. 第一次获取配置 content = await client.get_config(ConfigParam(data_id=NACOS_DATA_ID, group=NACOS_GROUP)) if content: loaded = yaml.safe_load(content) or {} nacos_config_data = loaded # 用 Nacos 配置覆盖 settings for key, value in loaded.items(): if hasattr(settings, key): setattr(settings, key, value) logger.info(f"✅ Nacos 配置加载成功: {NACOS_DATA_ID} | 覆盖字段数量: {len(loaded)}") else: logger.warning("Nacos 返回配置为空，使用 .env + 默认值") # 2. 注册动态监听器（配置变更自动刷新） async def listener(tenant: str, data_id: str, group: str, content: str): global nacos_config_data, settings try: new_config = yaml.safe_load(content) if content else {} nacos_config_data = new_config # 实时覆盖 settings for key, value in new_config.items(): if hasattr(settings, key): old_val = getattr(settings, key) setattr(settings, key, value) if old_val != value: logger.info(f"🔄 配置更新 → {key}: {old_val} → {value}") logger.info(f"【Nacos 动态更新】{NACOS_DATA_ID}") except Exception as e: logger.error(f"Nacos 配置解析失败: {e}") await client.add_listener(NACOS_DATA_ID, NACOS_GROUP, listener) logger.info("✅ Nacos 配置监听器已注册（支持热更新）") nacos_initialized_successfully = True except Exception as e: logger.error(f"❌ Nacos 初始化失败: {e}") nacos_initialized_successfully = False # 提供给 FastAPI 的依赖 def get_settings() -> Settings: return settings """Design 服务""" # 推荐服装类别映射 TABLE_CATEGORIES = { "female_dress": "female/dress", "female_outwear": "female/outwear", "female_trousers": "female/trousers", "female_skirt": "female/skirt", "female_blouse": "female/blouse", "male_tops": "male/tops", "male_bottoms": "male/bottoms", "male_outwear": "male/outwear" } # Design前后排优先级 PRIORITY_DICT = { 'earring_front': 99, 'bag_front': 98, 'hairstyle_front': 97, 'outwear_front': 20, 'tops_front': 19, 'dress_front': 18, 'blouse_front': 17, 'skirt_front': 16, 'trousers_front': 15, 'bottoms_front': 14, 'shoes_right': 1, 'shoes_left': 1, 'body': 0, 'bottoms_back': -14, 'trousers_back': -15, 'skirt_back': -16, 'blouse_back': -17, 'dress_back': -18, 'tops_back': -19, 'outwear_back': -20, 'hairstyle_back': -97, 'bag_back': -98, 'earring_back': -99, } # Design 关键点字段 KEYPOINT_RESULT_TABLE_FIELD_SET = ('neckline_left', 'neckline_right', 'shoulder_left', 'shoulder_right', 'armpit_left', 'armpit_right', 'cuff_left_in', 'cuff_left_out', 'cuff_right_in', 'cuff_right_out', 'waistband_left', 'waistband_right') # milvus配置信息 MILVUS_TABLE_KEYPOINT = "keypoint_cache_2" # ollama 地址 OLLAMA_URL = f"http://{settings.A6000_SERVICE_HOST}:11434/api/embeddings" """Triton Server Config""" # Design DESIGN_MODEL_URL = f'{settings.A6000_SERVICE_HOST}:10000' DESIGN_MODEL_NAME = 'seg_knet' # Seg Product SEG_PRODUCT_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:30000' # Generate Image GI_MODEL_URL = f'{settings.A6000_SERVICE_HOST}:10061' GI_MODEL_NAME = 'flux' # Generate Single Logo GSL_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:10041' GSL_MODEL_NAME = 'stable_diffusion_xl_transparent' # Generate Product （整套和单品） GPI_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:10051' GPI_MODEL_NAME_OVERALL = 'diffusion_ensemble_all' GPI_MODEL_NAME_SINGLE = 'stable_diffusion_1_5_cnet' # 以下停用中...************* # 多视角生成 GMV_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:10081' GMV_MODEL_NAME = 'multi_view' # 超分 SR_MODEL_NAME = "super_resolution" SR_TRITON_URL = f"{settings.A6000_SERVICE_HOST}:10031" # 打光 GRI_MODEL_URL = f'{settings.A6000_SERVICE_HOST}:10051' GRI_MODEL_NAME_OVERALL = 'diffusion_relight_ensemble' GRI_MODEL_NAME_SINGLE = 'stable_diffusion_1_5_relight' # agent 图片生成 FAST_GI_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:10011' FAST_GI_MODEL_NAME = 'stable_diffusion_xl' # 图转视频 triton版 PT_MODEL_URL = f'{settings.B_4_X_4090_SERVICE_HOST}:10061' # ************* """MQ 队列信息""" # 生成图片 moodboard printboard sketchboard GI_RABBITMQ_QUEUES = f"GenerateImage-{settings.SERVE_ENV}" # 生成slogan SLOGAN_RABBITMQ_QUEUES = f"Slogan-{settings.SERVE_ENV}" # 转产品图 GPI_RABBITMQ_QUEUES = f"ToProductImage-{settings.SERVE_ENV}" # 产品图转视频 PS_RABBITMQ_QUEUES = f"PoseTransform-{settings.SERVE_ENV}" # 以下停用中...************* # 产品图打光 GRI_RABBITMQ_QUEUES = f"Relight-{settings.SERVE_ENV}" # 超分 SR_RABBITMQ_QUEUES = f"SuperResolution-{settings.SERVE_ENV}" # 生成多视图 GMV_RABBITMQ_QUEUES = f"GenerateMultiView-{settings.SERVE_ENV}" # 批量转产品图 BATCH_GPI_RABBITMQ_QUEUES = f"BatchToProductImage-{settings.SERVE_ENV}" # 批量打光 BATCH_GRI_RABBITMQ_QUEUES = f"BatchRelight-{settings.SERVE_ENV}" # 批量图片转视频 BATCH_PS_RABBITMQ_QUEUES = f"BatchPoseTransform-{settings.SERVE_ENV}" # 批量design BATCH_DESIGN_RABBITMQ_QUEUES = f"DesignBatch-{settings.SERVE_ENV}" # *************