""" 农业投入Cd通量计算API接口 @description: 提供各个地区的农业投入输入Cd通量预测计算功能 """ from fastapi import APIRouter, HTTPException, Query, Body from pydantic import BaseModel, Field from typing import Dict, Any, Optional import logging from ..services.agricultural_input_service import AgriculturalInputService router = APIRouter() # ============================================================================= # 数据模型定义 # ============================================================================= class CdFluxCalculationRequest(BaseModel): """ 农业投入Cd通量计算请求模型 @description: 用户提供的计算参数 """ # 镉含量参数 (mg/kg) f3_nitrogen_cd_content: float = Field(..., description="氮肥镉含量平均值 (mg/kg)", ge=0) f4_phosphorus_cd_content: float = Field(..., description="磷肥镉含量平均值 (mg/kg)", ge=0) f5_potassium_cd_content: float = Field(..., description="钾肥镉含量平均值 (mg/kg)", ge=0) f6_compound_cd_content: float = Field(..., description="复合肥镉含量平均值 (mg/kg)", ge=0) f7_organic_cd_content: float = Field(..., description="有机肥镉含量平均值 (mg/kg)", ge=0) f8_pesticide_cd_content: float = Field(..., description="农药镉含量 (mg/kg)", ge=0) f9_farmyard_cd_content: float = Field(..., description="农家肥镉含量 (mg/kg)", ge=0) f10_film_cd_content: float = Field(..., description="农膜镉含量 (mg/kg)", ge=0) # 使用量参数 (t/ha/a) nf_nitrogen_usage: float = Field(..., description="氮肥单位面积使用量 (t/ha/a)", ge=0) pf_phosphorus_usage: float = Field(..., description="磷肥单位面积使用量 (t/ha/a)", ge=0) kf_potassium_usage: float = Field(..., description="钾肥单位面积使用量 (t/ha/a)", ge=0) cf_compound_usage: float = Field(..., description="复合肥单位面积使用量 (t/ha/a)", ge=0) of_organic_usage: float = Field(..., description="有机肥单位面积使用量 (t/ha/a)", ge=0) p_pesticide_usage: float = Field(..., description="农药单位面积使用量 (t/ha/a)", ge=0) ff_farmyard_usage: float = Field(..., description="农家肥单位面积使用量 (t/ha/a)", ge=0) af_film_usage: float = Field(..., description="农膜(存留)单位面积使用量 (t/ha/a)", ge=0) # 可选的标识信息 description: Optional[str] = Field(None, description="计算描述信息") model_config = { "json_schema_extra": { "example": { "f3_nitrogen_cd_content": 0.12, "f4_phosphorus_cd_content": 0.85, "f5_potassium_cd_content": 0.05, "f6_compound_cd_content": 0.45, "f7_organic_cd_content": 0.22, "f8_pesticide_cd_content": 0.08, "f9_farmyard_cd_content": 0.15, "f10_film_cd_content": 0.03, "nf_nitrogen_usage": 0.25, "pf_phosphorus_usage": 0.15, "kf_potassium_usage": 0.12, "cf_compound_usage": 0.30, "of_organic_usage": 2.50, "p_pesticide_usage": 0.02, "ff_farmyard_usage": 1.80, "af_film_usage": 0.05, "description": "测试地区农业投入Cd通量计算" } } } # 设置日志 logger = logging.getLogger(__name__) # ============================================================================= # 农业投入Cd通量计算接口 # ============================================================================= @router.get("/calculate-by-area", summary="根据地区计算农业投入Cd通量", description="根据指定地区从Parameters表中获取数据并计算农业投入输入Cd通量") async def calculate_cd_flux_by_area( area: str = Query(..., description="地区名称，如：韶关") ) -> Dict[str, Any]: """ 根据地区计算农业投入输入Cd通量 @param area: 地区名称 @returns: 计算结果包括总通量和各项明细 计算公式：农业投入输入Cd(g/ha/a) = F3*NF + F4*PF + F5*KF + F6*CF + F7*OF + F8*P + F9*FF + F10*AF """ try: service = AgriculturalInputService() result = service.calculate_cd_flux_by_area(area) if not result["success"]: raise HTTPException( status_code=404, detail=result["message"] ) return result except HTTPException: raise except Exception as e: logger.error(f"计算地区 '{area}' 的Cd通量失败: {str(e)}") raise HTTPException( status_code=500, detail=f"计算失败: {str(e)}" ) @router.get("/calculate-all-areas", summary="计算所有地区的农业投入Cd通量", description="计算数据库中所有地区的农业投入输入Cd通量，并提供统计汇总") async def calculate_all_areas_cd_flux() -> Dict[str, Any]: """ 计算所有地区的农业投入输入Cd通量 @returns: 所有地区的计算结果，按通量从高到低排序，包含统计汇总 """ try: service = AgriculturalInputService() result = service.calculate_all_areas_cd_flux() if not result["success"]: raise HTTPException( status_code=500, detail=result["message"] ) return result except HTTPException: raise except Exception as e: logger.error(f"计算所有地区Cd通量失败: {str(e)}") raise HTTPException( status_code=500, detail=f"计算失败: {str(e)}" ) @router.get("/available-areas", summary="获取可用地区列表", description="获取数据库中所有可用于计算的地区列表") async def get_available_areas() -> Dict[str, Any]: """ 获取数据库中所有可用的地区列表 @returns: 可用地区列表 """ try: service = AgriculturalInputService() result = service.get_available_areas() if not result["success"]: raise HTTPException( status_code=500, detail=result["message"] ) return result except HTTPException: raise except Exception as e: logger.error(f"获取可用地区列表失败: {str(e)}") raise HTTPException( status_code=500, detail=f"获取失败: {str(e)}" ) @router.post("/calculate-with-custom-data", summary="使用自定义数据计算农业投入Cd通量", description="接收用户提供的参数数据，直接进行农业投入输入Cd通量计算") async def calculate_cd_flux_with_custom_data( request: CdFluxCalculationRequest = Body(..., description="计算所需的参数数据") ) -> Dict[str, Any]: """ 使用用户提供的自定义数据计算农业投入输入Cd通量 @param request: 包含所有计算参数的请求体 @returns: 计算结果包括总通量和各项明细 计算公式：农业投入输入Cd(g/ha/a) = F3*NF + F4*PF + F5*KF + F6*CF + F7*OF + F8*P + F9*FF + F10*AF """ try: service = AgriculturalInputService() result = service.calculate_cd_flux_with_custom_data(request) if not result["success"]: raise HTTPException( status_code=400, detail=result["message"] ) return result except HTTPException: raise except Exception as e: logger.error(f"使用自定义数据计算Cd通量失败: {str(e)}") raise HTTPException( status_code=500, detail=f"计算失败: {str(e)}" ) @router.get("/calculation-formula", summary="获取计算公式说明", description="获取农业投入Cd通量的计算公式和参数说明") async def get_calculation_formula() -> Dict[str, Any]: """ 获取农业投入Cd通量的计算公式和参数说明 @returns: 公式和参数详细说明 """ try: formula_info = { "success": True, "message": "农业投入Cd通量计算公式", "data": { "formula": "农业投入输入Cd(g/ha/a) = F3*NF + F4*PF + F5*KF + F6*CF + F7*OF + F8*P + F9*FF + F10*AF", "unit": "g/ha/a", "parameters": { "F3": "氮肥镉含量平均值(mg/kg)", "F4": "磷肥镉含量平均值(mg/kg)", "F5": "钾肥镉含量平均值(mg/kg)", "F6": "复合肥镉含量平均值(mg/kg)", "F7": "有机肥镉含量平均值(mg/kg)", "F8": "农药镉含量(mg/kg)", "F9": "农家肥镉含量(mg/kg)", "F10": "农膜镉含量(mg/kg)", "NF": "氮肥单位面积使用量(t/ha/a)", "PF": "磷肥单位面积使用量(t/ha/a)", "KF": "钾肥单位面积使用量(t/ha/a)", "CF": "复合肥单位面积使用量(t/ha/a)", "OF": "有机肥单位面积使用量(t/ha/a)", "P": "农药单位面积使用量(t/ha/a)", "FF": "农家肥单位面积使用量(t/ha/a)", "AF": "农膜(存留)单位面积使用量(t/ha/a)" }, "components": { "nitrogen_fertilizer": "氮肥贡献量 = F3 × NF", "phosphorus_fertilizer": "磷肥贡献量 = F4 × PF", "potassium_fertilizer": "钾肥贡献量 = F5 × KF", "compound_fertilizer": "复合肥贡献量 = F6 × CF", "organic_fertilizer": "有机肥贡献量 = F7 × OF", "pesticide": "农药贡献量 = F8 × P", "farmyard_manure": "农家肥贡献量 = F9 × FF", "agricultural_film": "农膜贡献量 = F10 × AF" } } } return formula_info except Exception as e: logger.error(f"获取计算公式失败: {str(e)}") raise HTTPException( status_code=500, detail=f"获取失败: {str(e)}" )