AcidMap/app/services/cd_prediction_service_v3.py

"""
Cd预测服务类 v3.0
@description: 完全自包含的服务，不依赖Cd_Prediction_Integrated_System
@author: AcidMap Team
@version: 3.0.0
"""

import os
import logging
import asyncio
import tempfile
import json
from datetime import datetime
from typing import Dict, Any, Optional, List
import pandas as pd
import glob

from ..config.cd_prediction_config import cd_config
from ..models.cd_prediction import CdPredictionEngine
from ..database import SessionLocal
from .admin_boundary_service import get_boundary_geojson_by_name


class CdPredictionServiceV3:
    """
    Cd预测服务类 v3.0 - 完全自包含版本
    
    @description: 使用自包含的预测模型，完全不依赖外部集成系统
    """
    
    def __init__(self):
        """
        初始化Cd预测服务
        """
        # 设置日志
        self.logger = logging.getLogger(__name__)
        
        # 获取配置
        self.config = cd_config
        
        # 设置输出目录（使用基础输出目录）
        self.output_base_dir = self.config.get_output_dir("base")
        
        # 初始化预测引擎
        self.engine = CdPredictionEngine(self.output_base_dir)
        
        # 输出子目录
        self.output_figures_dir = os.path.join(self.output_base_dir, "figures")
        self.output_raster_dir = os.path.join(self.output_base_dir, "raster")
        self.output_data_dir = os.path.join(self.output_base_dir, "data")
        
        # 支持的县市配置
        self.supported_counties = self._load_supported_counties()
        
        self.logger.info("Cd预测服务v3.0初始化完成（完全自包含）")
    
    def _load_supported_counties(self) -> Dict[str, Dict]:
        """
        加载支持的县市配置
        
        @returns {Dict[str, Dict]} 支持的县市配置信息
        """
        return {
            "乐昌市": {
                "region_code": "440282",
                "display_name": "乐昌市",
                "province": "广东省"
            }
            # 可以继续添加更多县市
        }
    
    def is_county_supported(self, county_name: str) -> bool:
        """
        检查县市是否被支持
        
        @param {str} county_name - 县市名称
        @returns {bool} 是否支持该县市
        """
        return county_name in self.supported_counties
    
    def get_supported_counties(self) -> List[str]:
        """
        获取支持的县市名称列表
        
        @returns {List[str]} 支持的县市名称列表
        """
        return list(self.supported_counties.keys())
    
    def get_supported_counties_info(self) -> List[Dict[str, Any]]:
        """
        获取支持的县市详细信息
        
        @returns {List[Dict[str, Any]]} 支持的县市详细信息列表
        """
        counties_info = []
        for county_name, config in self.supported_counties.items():
            counties_info.append({
                "name": county_name,
                "display_name": config.get("display_name", county_name),
                "province": config.get("province", ""),
                "region_code": config.get("region_code", ""),
                "supported": True  # v3.0版本默认都支持
            })
        return counties_info
    
    def validate_input_data(self, df: pd.DataFrame, county_name: str) -> Dict[str, Any]:
        """
        验证输入数据格式
        
        @param {pd.DataFrame} df - 输入数据
        @param {str} county_name - 县市名称
        @returns {Dict[str, Any]} 验证结果
        """
        # 基本要求：前两列为坐标，至少需要3列数据
        if df.shape[1] < 3:
            return {
                "valid": False,
                "errors": ["数据至少需要3列：前两列为经纬度，后续列为环境因子"],
                "warnings": [],
                "data_shape": df.shape,
                "county_supported": self.is_county_supported(county_name)
            }
        
        # 坐标列验证
        coordinate_issues = []
        try:
            # 检查前两列是否为数值型
            if not pd.api.types.is_numeric_dtype(df.iloc[:, 0]):
                coordinate_issues.append("第一列（经度）不是数值型数据")
            elif not df.iloc[:, 0].between(70, 140).all():
                coordinate_issues.append("经度值超出合理范围（70-140度）")
                
            if not pd.api.types.is_numeric_dtype(df.iloc[:, 1]):
                coordinate_issues.append("第二列（纬度）不是数值型数据")
            elif not df.iloc[:, 1].between(15, 55).all():
                coordinate_issues.append("纬度值超出合理范围（15-55度）")
        except Exception as e:
            coordinate_issues.append(f"坐标数据验证失败: {str(e)}")
        
        # 检查数据完整性
        null_counts = df.isnull().sum()
        high_null_columns = [col for col, count in null_counts.items() 
                           if count > len(df) * 0.1]
        
        # 检查环境因子列是否为数值型
        non_numeric_columns = []
        for i in range(2, df.shape[1]):
            col_name = df.columns[i]
            if not pd.api.types.is_numeric_dtype(df.iloc[:, i]):
                non_numeric_columns.append(col_name)
        
        warnings = []
        if high_null_columns:
            warnings.append(f"以下列空值较多: {', '.join(high_null_columns)}")
        if coordinate_issues:
            warnings.extend(coordinate_issues)
        if non_numeric_columns:
            warnings.append(f"以下列不是数值型: {', '.join(non_numeric_columns)}")
        
        # 如果有严重的坐标问题，标记为无效
        critical_errors = []
        if any("不是数值型数据" in issue for issue in coordinate_issues):
            critical_errors.extend([issue for issue in coordinate_issues if "不是数值型数据" in issue])
        
        return {
            "valid": len(critical_errors) == 0,
            "errors": critical_errors,
            "warnings": warnings,
            "data_shape": df.shape,
            "county_supported": self.is_county_supported(county_name)
        }
    
    def save_temp_data(self, df: pd.DataFrame, county_name: str) -> str:
        """
        保存临时数据文件
        
        @param {pd.DataFrame} df - 数据
        @param {str} county_name - 县市名称
        @returns {str} 临时文件路径
        """
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = f"{county_name}_temp_data_{timestamp}.csv"
        file_path = os.path.join(self.output_data_dir, "temp", filename)
        
        # 确保目录存在
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        
        # 保存数据
        df.to_csv(file_path, index=False, encoding='utf-8-sig')
        
        file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
        self.logger.info(f"临时数据文件已保存: {file_path} ({file_size:,} bytes)")
        
        # 清理旧的临时文件
        self._cleanup_temp_files(county_name)
        
        return file_path
    
    def _get_boundary_geojson(self, county_name: str) -> Optional[str]:
        """
        获取县市边界的GeoJSON文件
        
        @param {str} county_name - 县市名称
        @returns {Optional[str]} GeoJSON文件路径，如果失败则返回None
        """
        try:
            db = SessionLocal()
            feature = get_boundary_geojson_by_name(db, county_name, level="auto")
            fc = {"type": "FeatureCollection", "features": [feature]}
            
            # 创建临时GeoJSON文件
            tmp_dir = tempfile.mkdtemp()
            tmp_geojson = os.path.join(tmp_dir, "boundary.geojson")
            with open(tmp_geojson, 'w', encoding='utf-8') as f:
                json.dump(fc, f, ensure_ascii=False)
            
            return tmp_geojson
            
        except Exception as e:
            self.logger.warning(f"从数据库获取边界失败: {str(e)}")
            return None
        finally:
            try:
                db.close()
            except Exception:
                pass
    
    async def generate_crop_cd_prediction_for_county(
        self, 
        county_name: str, 
        data_file: Optional[str] = None,
        raster_config_override: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        为指定县市生成作物Cd预测
        
        @param {str} county_name - 县市名称
        @param {Optional[str]} data_file - 可选的数据文件路径
        @param {Optional[Dict[str, Any]]} raster_config_override - 栅格配置覆盖参数
        @returns {Dict[str, Any]} 预测结果信息
        """
        if not self.is_county_supported(county_name):
            raise ValueError(f"不支持的县市: {county_name}")
        
        try:
            self.logger.info(f"开始为{county_name}生成作物Cd预测（v3.0自包含版本）")
            
            # 读取数据文件
            if not data_file or not os.path.exists(data_file):
                raise ValueError("需要提供有效的数据文件")
            
            df = pd.read_csv(data_file, encoding='utf-8')
            # 确保前两列为经纬度
            df.columns = ['longitude', 'latitude'] + list(df.columns[2:])
            
            # 获取边界文件
            boundary_geojson = self._get_boundary_geojson(county_name)
            
            # 在线程池中运行预测
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                None, 
                self._run_crop_cd_prediction,
                df, county_name, boundary_geojson, raster_config_override
            )
            
            # 使用统一绘图接口，文件已直接生成在正确位置，无需复制
            api_result = self._format_api_result(result, county_name, "crop_cd")
            
            # 清理临时文件
            self._cleanup_after_prediction(boundary_geojson)
            
            return api_result
            
        except Exception as e:
            self.logger.error(f"为{county_name}生成作物Cd预测失败: {str(e)}")
            raise
    
    async def generate_effective_cd_prediction_for_county(
        self, 
        county_name: str, 
        data_file: Optional[str] = None,
        raster_config_override: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        为指定县市生成有效态Cd预测
        
        @param {str} county_name - 县市名称
        @param {Optional[str]} data_file - 可选的数据文件路径
        @param {Optional[Dict[str, Any]]} raster_config_override - 栅格配置覆盖参数
        @returns {Dict[str, Any]} 预测结果信息
        """
        if not self.is_county_supported(county_name):
            raise ValueError(f"不支持的县市: {county_name}")
        
        try:
            self.logger.info(f"开始为{county_name}生成有效态Cd预测（v3.0自包含版本）")
            
            # 读取数据文件
            if not data_file or not os.path.exists(data_file):
                raise ValueError("需要提供有效的数据文件")
            
            df = pd.read_csv(data_file, encoding='utf-8')
            # 确保前两列为经纬度
            df.columns = ['longitude', 'latitude'] + list(df.columns[2:])
            
            # 获取边界文件
            boundary_geojson = self._get_boundary_geojson(county_name)
            
            # 在线程池中运行预测
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                None, 
                self._run_effective_cd_prediction,
                df, county_name, boundary_geojson, raster_config_override
            )
            
            # 使用统一绘图接口，文件已直接生成在正确位置，无需复制
            api_result = self._format_api_result(result, county_name, "effective_cd")
            
            # 清理临时文件
            self._cleanup_after_prediction(boundary_geojson)
            
            return api_result
            
        except Exception as e:
            self.logger.error(f"为{county_name}生成有效态Cd预测失败: {str(e)}")
            raise
    
    def _run_crop_cd_prediction(self, df: pd.DataFrame, county_name: str, 
                              boundary_geojson: Optional[str],
                              raster_config_override: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        """
        执行作物Cd预测（同步方法）
        
        @param {pd.DataFrame} df - 输入数据
        @param {str} county_name - 县市名称
        @param {Optional[str]} boundary_geojson - 边界文件路径
        @param {Optional[Dict[str, Any]]} raster_config_override - 栅格配置覆盖参数
        @returns {Dict[str, Any]} 预测结果
        """
        try:
            # 准备边界数据
            boundary_gdf = None
            if boundary_geojson:
                try:
                    import geopandas as gpd
                    boundary_gdf = gpd.read_file(boundary_geojson)
                    self.logger.info(f"已加载边界文件: {boundary_geojson}")
                except Exception as e:
                    self.logger.warning(f"加载边界文件失败，将不使用边界限制: {str(e)}")
                    boundary_gdf = None
            
            # 执行预测和可视化（使用统一绘图接口，不保存栅格文件）
            result = self.engine.predict_and_visualize(
                input_data=df,
                model_type="crop_cd",
                county_name=county_name,
                boundary_gdf=boundary_gdf,
                raster_config_override=raster_config_override,
                save_raster=False  # 不保存栅格文件，节省存储空间
            )
            
            return result
            
        except Exception as e:
            self.logger.error(f"作物Cd预测执行失败: {str(e)}")
            raise
    
    def _run_effective_cd_prediction(self, df: pd.DataFrame, county_name: str, 
                                   boundary_geojson: Optional[str],
                                   raster_config_override: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        """
        执行有效态Cd预测（同步方法）
        
        @param {pd.DataFrame} df - 输入数据
        @param {str} county_name - 县市名称
        @param {Optional[str]} boundary_geojson - 边界文件路径
        @param {Optional[Dict[str, Any]]} raster_config_override - 栅格配置覆盖参数
        @returns {Dict[str, Any]} 预测结果
        """
        try:
            # 准备边界数据
            boundary_gdf = None
            if boundary_geojson:
                try:
                    import geopandas as gpd
                    boundary_gdf = gpd.read_file(boundary_geojson)
                    self.logger.info(f"已加载边界文件: {boundary_geojson}")
                except Exception as e:
                    self.logger.warning(f"加载边界文件失败，将不使用边界限制: {str(e)}")
                    boundary_gdf = None
                    
            # 执行预测和可视化（使用统一绘图接口，不保存栅格文件）
            result = self.engine.predict_and_visualize(
                input_data=df,
                model_type="effective_cd",
                county_name=county_name,
                boundary_gdf=boundary_gdf,
                raster_config_override=raster_config_override,
                save_raster=False  # 不保存栅格文件，节省存储空间
            )
            
            return result
            
        except Exception as e:
            self.logger.error(f"有效态Cd预测执行失败: {str(e)}")
            raise
    
    def _format_api_result(self, result: Dict[str, Any], county_name: str, 
                         model_type: str) -> Dict[str, Any]:
        """
        格式化API返回结果（使用统一绘图接口，无需复制文件）
        
        @param {Dict[str, Any]} result - 预测结果
        @param {str} county_name - 县市名称
        @param {str} model_type - 模型类型
        @returns {Dict[str, Any]} 格式化的API结果
        """
        try:
            model_prefix = f"{model_type}_{county_name}"
            
            # 直接使用引擎生成的文件路径，无需复制
            api_result = {
                'map_path': result.get('map_path'),
                'histogram_path': result.get('histogram_path'),
                'raster_path': result.get('raster_path'),
                'model_type': model_type,
                'county_name': county_name,
                'timestamp': result.get('timestamp'),
                'validation': result.get('validation', {}),
                'stats': self._get_file_stats(result.get('map_path'))
            }
            
            # 清理旧文件（保留）
            self._cleanup_old_files(model_prefix)
            
            self.logger.info(f"API结果格式化完成（统一接口）: {model_type}_{county_name}")
            return api_result
            
        except Exception as e:
            self.logger.error(f"格式化API结果失败: {str(e)}")
            raise
    
    def _cleanup_after_prediction(self, boundary_geojson: Optional[str]):
        """
        预测后清理工作
        
        @param {Optional[str]} boundary_geojson - 临时边界文件路径
        """
        try:
            # 清理预测引擎的临时文件
            self.engine.cleanup_temp_files()
            
            # 清理临时边界文件
            if boundary_geojson and os.path.exists(boundary_geojson):
                import shutil
                shutil.rmtree(os.path.dirname(boundary_geojson), ignore_errors=True)
                self.logger.info("临时边界文件已清理")
                
        except Exception as e:
            self.logger.warning(f"预测后清理失败: {str(e)}")
    
    def _cleanup_temp_files(self, county_name: str, max_files: int = 5):
        """
        清理临时数据文件
        
        @param {str} county_name - 县市名称
        @param {int} max_files - 最大保留文件数
        """
        try:
            temp_dir = os.path.join(self.output_data_dir, "temp")
            temp_pattern = os.path.join(temp_dir, f"{county_name}_temp_data_*.csv")
            
            temp_files = glob.glob(temp_pattern)
            
            if len(temp_files) > max_files:
                temp_files.sort(key=os.path.getmtime)
                files_to_delete = temp_files[:-max_files]
                
                for file_to_delete in files_to_delete:
                    os.remove(file_to_delete)
                    self.logger.info(f"已删除旧临时文件: {os.path.basename(file_to_delete)}")
                
        except Exception as e:
            self.logger.warning(f"清理临时文件失败: {str(e)}")
    
    def _cleanup_old_files(self, model_prefix: str):
        """
        清理旧的预测文件
        
        @param {str} model_prefix - 模型前缀
        """
        try:
            max_files = 10
            
            # 清理地图文件
            map_pattern = os.path.join(self.output_figures_dir, f"{model_prefix}_prediction_map_*.jpg")
            self._cleanup_files_by_pattern(map_pattern, max_files)
            
            # 清理直方图文件
            histogram_pattern = os.path.join(self.output_figures_dir, f"{model_prefix}_prediction_histogram_*.jpg")
            self._cleanup_files_by_pattern(histogram_pattern, max_files)
            
        except Exception as e:
            self.logger.warning(f"清理旧文件失败: {str(e)}")
    
    def _cleanup_files_by_pattern(self, pattern: str, max_files: int):
        """
        按模式清理文件
        
        @param {str} pattern - 文件模式
        @param {int} max_files - 最大保留文件数
        """
        try:
            files = glob.glob(pattern)
            if len(files) > max_files:
                files.sort(key=os.path.getmtime)
                for file_to_delete in files[:-max_files]:
                    os.remove(file_to_delete)
                    self.logger.info(f"已删除旧文件: {os.path.basename(file_to_delete)}")
        except Exception as e:
            self.logger.warning(f"清理文件失败 {pattern}: {str(e)}")
    
    def _get_file_stats(self, file_path: Optional[str]) -> Dict[str, Any]:
        """
        获取文件统计信息
        
        @param {Optional[str]} file_path - 文件路径
        @returns {Dict[str, Any]} 文件统计信息
        """
        if not file_path or not os.path.exists(file_path):
            return {}
        
        try:
            stat = os.stat(file_path)
            return {
                'file_size': stat.st_size,
                'created_time': datetime.fromtimestamp(stat.st_ctime).isoformat(),
                'modified_time': datetime.fromtimestamp(stat.st_mtime).isoformat()
            }
        except Exception:
            return {}
    
    def get_engine_info(self) -> Dict[str, Any]:
        """
        获取引擎信息
        
        @returns {Dict[str, Any]} 引擎信息
        """
        try:
            return self.engine.get_model_info()
        except Exception as e:
            self.logger.error(f"获取引擎信息失败: {str(e)}")
            return {"error": str(e)}
    
    # =============================================================================
    # 统计信息方法（简化版本，从最终数据文件中读取）
    # =============================================================================
    
    def get_crop_cd_statistics(self, county_name: str) -> Optional[Dict[str, Any]]:
        """
        获取作物Cd预测结果的统计信息
        
        @param {str} county_name - 县市名称
        @returns {Optional[Dict[str, Any]]} 统计信息
        """
        try:
            # 查找最新的最终数据文件
            final_pattern = os.path.join(self.output_data_dir, "final", f"Final_predictions_crop_cd_*.csv")
            final_files = glob.glob(final_pattern)
            
            if not final_files:
                return None
            
            latest_file = max(final_files, key=os.path.getmtime)
            df = pd.read_csv(latest_file)
            
            if 'Prediction' not in df.columns:
                return None
            
            predictions = df['Prediction']
            
            basic_stats = {
                "数据点总数": len(predictions),
                "均值": float(predictions.mean()),
                "中位数": float(predictions.median()),
                "标准差": float(predictions.std()),
                "最小值": float(predictions.min()),
                "最大值": float(predictions.max()),
                "25%分位数": float(predictions.quantile(0.25)),
                "75%分位数": float(predictions.quantile(0.75))
            }
            
            return {
                "模型类型": "作物Cd模型v3.0",
                "县市名称": county_name,
                "数据更新时间": datetime.fromtimestamp(os.path.getmtime(latest_file)).isoformat(),
                "基础统计": basic_stats
            }
            
        except Exception as e:
            self.logger.error(f"获取作物Cd统计信息失败: {str(e)}")
            return None
    
    def get_effective_cd_statistics(self, county_name: str) -> Optional[Dict[str, Any]]:
        """
        获取有效态Cd预测结果的统计信息
        
        @param {str} county_name - 县市名称
        @returns {Optional[Dict[str, Any]]} 统计信息
        """
        try:
            # 查找最新的最终数据文件
            final_pattern = os.path.join(self.output_data_dir, "final", f"Final_predictions_effective_cd_*.csv")
            final_files = glob.glob(final_pattern)
            
            if not final_files:
                return None
            
            latest_file = max(final_files, key=os.path.getmtime)
            df = pd.read_csv(latest_file)
            
            if 'Prediction' not in df.columns:
                return None
            
            predictions = df['Prediction']
            
            basic_stats = {
                "数据点总数": len(predictions),
                "均值": float(predictions.mean()),
                "中位数": float(predictions.median()),
                "标准差": float(predictions.std()),
                "最小值": float(predictions.min()),
                "最大值": float(predictions.max()),
                "25%分位数": float(predictions.quantile(0.25)),
                "75%分位数": float(predictions.quantile(0.75))
            }
            
            return {
                "模型类型": "有效态Cd模型v3.0",
                "县市名称": county_name,
                "数据更新时间": datetime.fromtimestamp(os.path.getmtime(latest_file)).isoformat(),
                "基础统计": basic_stats
            }
            
        except Exception as e:
            self.logger.error(f"获取有效态Cd统计信息失败: {str(e)}")
            return None
    
    def get_combined_statistics(self, county_name: str) -> Optional[Dict[str, Any]]:
        """
        获取综合预测统计信息
        
        @param {str} county_name - 县市名称
        @returns {Optional[Dict[str, Any]]} 综合统计信息
        """
        try:
            crop_stats = self.get_crop_cd_statistics(county_name)
            effective_stats = self.get_effective_cd_statistics(county_name)
            
            if not crop_stats and not effective_stats:
                return None
            
            return {
                "县市名称": county_name,
                "作物Cd统计": crop_stats,
                "有效态Cd统计": effective_stats,
                "生成时间": datetime.now().isoformat(),
                "版本": "v3.0自包含版本"
            }
            
        except Exception as e:
            self.logger.error(f"获取综合统计信息失败: {str(e)}")
            return None
    
    def get_all_counties_statistics(self) -> Dict[str, Any]:
        """
        获取所有支持县市的统计概览
        
        @returns {Dict[str, Any]} 所有县市的统计概览
        """
        try:
            all_stats = {
                "支持县市总数": len(self.supported_counties),
                "统计生成时间": datetime.now().isoformat(),
                "版本": "v3.0自包含版本",
                "县市统计": {},
                "汇总信息": {
                    "有作物Cd数据的县市": 0,
                    "有有效态Cd数据的县市": 0,
                    "数据完整的县市": 0
                }
            }
            
            for county_name in self.supported_counties.keys():
                county_stats = {
                    "县市名称": county_name,
                    "有作物Cd数据": False,
                    "有有效态Cd数据": False,
                    "数据完整": False
                }
                
                # 检查作物Cd数据
                crop_stats = self.get_crop_cd_statistics(county_name)
                if crop_stats:
                    county_stats["有作物Cd数据"] = True
                    all_stats["汇总信息"]["有作物Cd数据的县市"] += 1
                
                # 检查有效态Cd数据
                effective_stats = self.get_effective_cd_statistics(county_name)
                if effective_stats:
                    county_stats["有有效态Cd数据"] = True
                    all_stats["汇总信息"]["有有效态Cd数据的县市"] += 1
                
                # 检查数据完整性
                if county_stats["有作物Cd数据"] and county_stats["有有效态Cd数据"]:
                    county_stats["数据完整"] = True
                    all_stats["汇总信息"]["数据完整的县市"] += 1
                
                all_stats["县市统计"][county_name] = county_stats
            
            return all_stats
            
        except Exception as e:
            self.logger.error(f"获取所有县市统计概览失败: {str(e)}")
            return {
                "error": f"获取统计概览失败: {str(e)}",
                "支持县市总数": len(self.supported_counties),
                "统计生成时间": datetime.now().isoformat(),
                "版本": "v3.0自包含版本"
            }