新增nacos 配置 测试

app/core/config.py

@@ -1,5 +1,21 @@
+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):
@@ -81,6 +97,132 @@ class Settings(BaseSettings):
 
 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 register_service_to_nacos():
+    """启动时把服务注册到 Nacos"""
+    if not nacos_initialized_successfully:
+        logger.warning("Nacos 未初始化成功，跳过服务注册")
+        return
+
+    nacos_config_client = await NacosConfigService.create_config_service(client_config)
+
+    if not nacos_config_client:  # 如果配置客户端都没连上，就不注册
+        logger.warning("Nacos 配置客户端未初始化，跳过服务注册")
+        return
+
+    try:
+        nacos_naming_client = await NacosNamingService.create_naming_service(client_config)
+        param = RegisterInstanceParam(
+            service_name="aida.python",
+            group_name=NACOS_GROUP,
+            ip="210.0.194.163",
+            port=1671,
+            cluster_name="DEFAULT",
+            weight=1.0,
+            metadata={
+                "version": "1.0.0",
+                "env": settings.SERVE_ENV,
+                "framework": "fastapi",
+                "debug": str(settings.DEBUG),
+            },
+            enabled=True,
+            healthy=True,
+            ephemeral=True,  # 临时实例，推荐生产使用
+        )
+
+        await nacos_naming_client.register_instance(request=param)
+        logger.info(f"✅ 服务已成功注册到 Nacos！ → {SERVICE_NAME} | {"10.1.1.249"}:{settings.PORT} | env={settings.SERVE_ENV}")
+
+    except Exception as e:
+        logger.error(f"❌ 服务注册到 Nacos 失败: {e}")
+
+
+async def deregister_service_from_nacos():
+    """服务关闭时优雅注销（防止 Nacos 长时间显示不健康实例）"""
+    try:
+        nacos_naming_client = await NacosNamingService.create_naming_service(client_config)
+        param = DeregisterInstanceParam(
+            service_name=SERVICE_NAME,
+            group_name=NACOS_GROUP,
+            ip="210.0.194.163",
+            port=1671,
+            cluster_name='c1',
+            ephemeral=True,
+        )
+        await nacos_naming_client.deregister_instance(request=param)
+        logger.info(f"✅ 服务已从 Nacos 注销 → {SERVICE_NAME}")
+    except Exception as e:
+        logger.warning(f"服务注销时出现异常（通常可忽略）: {e}")
+
+
+nacos_initialized_successfully = False
+
+
+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
+        await register_service_to_nacos()
+    except Exception as e:
+        logger.error(f"❌ Nacos 初始化失败: {e}")
+        nacos_initialized_successfully = False
+
+
+# 提供给 FastAPI 的依赖
+def get_settings() -> Settings:
+    return settings
+
+
 """Design 服务"""
 # 推荐服装类别映射
 TABLE_CATEGORIES = {

app/main.py

@@ -1,5 +1,5 @@
-# 1. 这里的顺序至关重要！必须在最顶端
 import sys
+from contextlib import asynccontextmanager
 
 try:
     import asyncore
@@ -10,16 +10,14 @@ except ImportError:
 import logging.config
 
 import uvicorn
-from fastapi import FastAPI
+from fastapi import FastAPI, Depends
 from fastapi import HTTPException
 from fastapi.responses import JSONResponse
 
 from app.api.api_route import router
-from app.core.config import settings
-# from app.core.record_api_count import count_api_calls
+from app.core.config import settings, load_nacos_config, get_settings, Settings, register_service_to_nacos, deregister_service_from_nacos, nacos_initialized_successfully
 from app.schemas.response_template import ResponseModel
 from logging_env import LOGGER_CONFIG_DICT
-from dotenv import load_dotenv
 from starlette.middleware.cors import CORSMiddleware
 
 logging.config.dictConfig(LOGGER_CONFIG_DICT)
@@ -27,33 +25,28 @@ logging.getLogger("pika").setLevel(logging.WARNING)
 
 logger = logging.getLogger(__name__)
 
-load_dotenv()
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # 启动阶段
+    try:
+        await load_nacos_config()  # 里面已包含配置加载 + 服务注册
+        logger.info(f"🚀 服务启动完成 | SERVE_ENV={settings.SERVE_ENV} | PORT={settings.PORT} | Nacos={'已连接' if nacos_initialized_successfully else '未连接（使用.env）'}")
+    except Exception as e:
+        logger.error(f"启动时发生严重错误: {e}")
+        raise
+
+    yield
+
+    # 关闭阶段（优雅下线）
+    await deregister_service_from_nacos()
+    # 如果有 nacos client，可以在这里 shutdown
+    logger.info("✅ 服务已优雅关闭")
 
 
-def get_application() -> FastAPI:
-    application = FastAPI(
-        docs_url="/docs",
-        redoc_url='/re-docs',
-        openapi_url=f"/openapi.json",
-        description='''
-        Base frame with FastAPI 
-            - Super Resolution API
-
-        '''
-    )
-    application.add_middleware(
-        CORSMiddleware,
-        allow_origins=["*"],
-        allow_credentials=True,
-        allow_methods=["*"],
-        allow_headers=["*"],
-    )
-    # application.middleware("http")(count_api_calls)
-    application.include_router(router=router)
-    return application
-
-
-app = get_application()
+app = FastAPI(lifespan=lifespan, docs_url="/docs", redoc_url='/re-docs', openapi_url=f"/openapi.json", description=''' Base frame with FastAPI - Super Resolution API ''')
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )
+app.include_router(router=router)
 
 
 @app.exception_handler(HTTPException)
@@ -64,5 +57,16 @@ async def http_exception_handler(exc: HTTPException):
     )
 
 
+@app.get("/health")
+async def health():
+    return {"status": "ok", "env": settings.SERVE_ENV}
+
+
+@app.get("/config")
+async def show_config(s: Settings = Depends(get_settings)):
+    """查看当前完整配置（生产环境建议加权限）"""
+    return s.model_dump()
+
+
 if __name__ == '__main__':
     uvicorn.run(app, host="0.0.0.0", port=settings.PORT)