#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 数据查询API路由 支持景区、产品、风格、受众等基础数据的查询 """ import logging from typing import List, Dict, Any, Optional from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field from api.services.database_service import DatabaseService from api.dependencies import get_database_service logger = logging.getLogger(__name__) router = APIRouter( prefix="/data", tags=["data"], responses={404: {"description": "Not found"}}, ) class ScenicSpotResponse(BaseModel): """景区响应模型""" id: int = Field(..., description="景区ID") name: str = Field(..., description="景区名称") address: Optional[str] = Field(None, description="地址") description: Optional[str] = Field(None, description="描述") advantage: Optional[str] = Field(None, description="优势") highlight: Optional[str] = Field(None, description="亮点") isPublic: bool = Field(..., description="是否公开") userId: int = Field(..., description="用户ID") class ProductResponse(BaseModel): """产品响应模型""" id: int = Field(..., description="产品ID") name: str = Field(..., description="产品名称") originPrice: Optional[float] = Field(None, description="原价") realPrice: Optional[float] = Field(None, description="实际价格") packageInfo: Optional[str] = Field(None, description="套票信息") description: Optional[str] = Field(None, description="描述") advantage: Optional[str] = Field(None, description="优势") highlight: Optional[str] = Field(None, description="亮点") isPublic: bool = Field(..., description="是否公开") userId: int = Field(..., description="用户ID") class StyleResponse(BaseModel): """风格响应模型""" id: int = Field(..., description="风格ID") styleName: str = Field(..., description="风格名称") description: Optional[str] = Field(None, description="描述") class AudienceResponse(BaseModel): """受众响应模型""" id: int = Field(..., description="受众ID") audienceName: str = Field(..., description="受众名称") description: Optional[str] = Field(None, description="描述") @router.get("/scenic-spots", response_model=List[ScenicSpotResponse], summary="获取景区列表") async def get_scenic_spots( user_id: Optional[int] = Query(None, description="用户ID，如果提供则只返回该用户的景区"), is_public: Optional[bool] = Query(None, description="是否公开，如果提供则过滤公开/私有景区"), db_service: DatabaseService = Depends(get_database_service) ): """ 获取景区列表 - **user_id**: 用户ID，如果提供则只返回该用户的景区 - **is_public**: 是否公开，如果提供则过滤公开/私有景区 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") spots = db_service.list_all_scenic_spots(user_id=user_id, is_public=is_public) return [ScenicSpotResponse(**spot) for spot in spots] except Exception as e: logger.error(f"获取景区列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取景区列表失败: {str(e)}") @router.get("/scenic-spots/{spot_id}", response_model=ScenicSpotResponse, summary="根据ID获取景区信息") async def get_scenic_spot_by_id( spot_id: int, db_service: DatabaseService = Depends(get_database_service) ): """ 根据ID获取景区信息 - **spot_id**: 景区ID """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") spot = db_service.get_scenic_spot_by_id(spot_id) if not spot: raise HTTPException(status_code=404, detail=f"未找到ID为{spot_id}的景区") return ScenicSpotResponse(**spot) except HTTPException: raise except Exception as e: logger.error(f"获取景区信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取景区信息失败: {str(e)}") @router.get("/products", response_model=List[ProductResponse], summary="获取产品列表") async def get_products( user_id: Optional[int] = Query(None, description="用户ID，如果提供则只返回该用户的产品"), is_public: Optional[bool] = Query(None, description="是否公开，如果提供则过滤公开/私有产品"), db_service: DatabaseService = Depends(get_database_service) ): """ 获取产品列表 - **user_id**: 用户ID，如果提供则只返回该用户的产品 - **is_public**: 是否公开，如果提供则过滤公开/私有产品 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") products = db_service.list_all_products(user_id=user_id, is_public=is_public) return [ProductResponse(**product) for product in products] except Exception as e: logger.error(f"获取产品列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取产品列表失败: {str(e)}") @router.get("/products/{product_id}", response_model=ProductResponse, summary="根据ID获取产品信息") async def get_product_by_id( product_id: int, db_service: DatabaseService = Depends(get_database_service) ): """ 根据ID获取产品信息 - **product_id**: 产品ID """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") product = db_service.get_product_by_id(product_id) if not product: raise HTTPException(status_code=404, detail=f"未找到ID为{product_id}的产品") return ProductResponse(**product) except HTTPException: raise except Exception as e: logger.error(f"获取产品信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取产品信息失败: {str(e)}") @router.get("/styles", response_model=List[StyleResponse], summary="获取风格列表") async def get_styles( db_service: DatabaseService = Depends(get_database_service) ): """ 获取所有风格列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") styles = db_service.list_all_styles() return [StyleResponse(**style) for style in styles] except Exception as e: logger.error(f"获取风格列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取风格列表失败: {str(e)}") @router.get("/styles/{style_id}", response_model=StyleResponse, summary="根据ID获取风格信息") async def get_style_by_id( style_id: int, db_service: DatabaseService = Depends(get_database_service) ): """ 根据ID获取风格信息 - **style_id**: 风格ID """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") style = db_service.get_style_by_id(style_id) if not style: raise HTTPException(status_code=404, detail=f"未找到ID为{style_id}的风格") return StyleResponse(**style) except HTTPException: raise except Exception as e: logger.error(f"获取风格信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取风格信息失败: {str(e)}") @router.get("/audiences", response_model=List[AudienceResponse], summary="获取受众列表") async def get_audiences( db_service: DatabaseService = Depends(get_database_service) ): """ 获取所有受众列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") audiences = db_service.list_all_audiences() return [AudienceResponse(**audience) for audience in audiences] except Exception as e: logger.error(f"获取受众列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取受众列表失败: {str(e)}") @router.get("/audiences/{audience_id}", response_model=AudienceResponse, summary="根据ID获取受众信息") async def get_audience_by_id( audience_id: int, db_service: DatabaseService = Depends(get_database_service) ): """ 根据ID获取受众信息 - **audience_id**: 受众ID """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") audience = db_service.get_audience_by_id(audience_id) if not audience: raise HTTPException(status_code=404, detail=f"未找到ID为{audience_id}的受众") return AudienceResponse(**audience) except HTTPException: raise except Exception as e: logger.error(f"获取受众信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"获取受众信息失败: {str(e)}") @router.post("/scenic-spots/batch", response_model=List[ScenicSpotResponse], summary="批量获取景区信息") async def get_scenic_spots_batch( spot_ids: List[int], db_service: DatabaseService = Depends(get_database_service) ): """ 批量获取景区信息 - **spot_ids**: 景区ID列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") spots = db_service.get_scenic_spots_by_ids(spot_ids) return [ScenicSpotResponse(**spot) for spot in spots] except Exception as e: logger.error(f"批量获取景区信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"批量获取景区信息失败: {str(e)}") @router.post("/products/batch", response_model=List[ProductResponse], summary="批量获取产品信息") async def get_products_batch( productIds: List[int], db_service: DatabaseService = Depends(get_database_service) ): """ 批量获取产品信息 - **productIds**: 产品ID列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") products = db_service.get_products_by_ids(productIds) return [ProductResponse(**product) for product in products] except Exception as e: logger.error(f"批量获取产品信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"批量获取产品信息失败: {str(e)}") @router.post("/styles/batch", response_model=List[StyleResponse], summary="批量获取风格信息") async def get_styles_batch( styleIds: List[int], db_service: DatabaseService = Depends(get_database_service) ): """ 批量获取风格信息 - **styleIds**: 风格ID列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") styles = db_service.get_styles_by_ids(styleIds) return [StyleResponse(**style) for style in styles] except Exception as e: logger.error(f"批量获取风格信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"批量获取风格信息失败: {str(e)}") @router.post("/audiences/batch", response_model=List[AudienceResponse], summary="批量获取受众信息") async def get_audiences_batch( audienceIds: List[int], db_service: DatabaseService = Depends(get_database_service) ): """ 批量获取受众信息 - **audienceIds**: 受众ID列表 """ try: if not db_service.is_available(): raise HTTPException(status_code=503, detail="数据库服务不可用") audiences = db_service.get_audiences_by_ids(audienceIds) return [AudienceResponse(**audience) for audience in audiences] except Exception as e: logger.error(f"批量获取受众信息失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"批量获取受众信息失败: {str(e)}")