#!/usr/bin/env python
# -*- coding:utf-8 -*-

import datetime
import json
import pandas as pd
import copy
import time
from mongoengine import *
from common.base import *
from bson import ObjectId


# ================= 优化工具函数 =================
def get_collection_with_filter(dataSetFlag, dataSetId, version=None):
    """
    获取集合名称
    
    :param dataSetFlag: 数据集标识
    :param dataSetId: 数据集ID
    :param version: 版本号（可选）
    :return: 集合名称
    """
    # 构建集合名称
    if dataSetFlag == 'vw':
        collection_name = 'vw_data_case_' + dataSetId
    elif dataSetFlag == 'True':
        collection_name = 'data_case_' + dataSetId
    else:
        collection_name = 'data_case'

    return collection_name


def execute_optimized_aggregate(
    res,
    dataSetFlag,
    dataSetId,
    pipeline,
    method_name,
    batch_size=10000,
    version=None,
    max_time_ms=None,
    allow_disk_use=True
):
    """
    执行优化的聚合查询
    
    :param res: Pymongoengine 实例
    :param dataSetFlag: 数据集标识 ('vw', 'True', 'False')
    :param dataSetId: 数据集ID
    :param pipeline: 聚合管道
    :param method_name: 方法名称，用于性能监控
    :param batch_size: 批量大小
    :param version: 版本号
    :param max_time_ms: 最大执行时间（毫秒）
    :param allow_disk_use: 是否允许使用磁盘
    :return: 查询结果
    """
    query_start_time = time.time()

    # 获取连接池
    pool = res.conn if hasattr(res, 'conn') else None

    # 使用辅助函数获取集合名称
    collection_name = get_collection_with_filter(dataSetFlag, dataSetId, version)

    try:
        # 使用优化的批量聚合方法
        if pool and hasattr(pool, 'batch_aggregate'):
            result_list = list(pool.batch_aggregate(
                collection_name,
                pipeline,
                allow_disk_use=allow_disk_use,
                batch_size=batch_size,
                data_set_id=dataSetId,
                data_set_flag=dataSetFlag,
                max_time_ms=max_time_ms,
                version=version
            ))

            # 记录查询性能
            duration_ms = (time.time() - query_start_time) * 1000
            if hasattr(pool, 'monitor_query'):
                pool.monitor_query(
                    f'{method_name}_{dataSetFlag}',
                    duration_ms,
                    len(result_list) if result_list else 0
                )

            return result_list
        else:
            # 回退到直接 pymongo 查询
            from mongoengine import get_db
            db = get_db(alias=res._alias_pid)
            collection = db[collection_name]
            
            # 构建匹配条件，同时处理version和dataSetId
            modified_pipeline = pipeline.copy()
            match_filter = {}
            
            if dataSetFlag not in ['vw', 'True']:
                match_filter["dataSetId"] = dataSetId
            
            if version is not None and dataSetFlag != 'vw':
                match_filter["version"] = version
            
            if match_filter:
                modified_pipeline.insert(0, {"$match": match_filter})
            
            # 构建聚合参数
            aggregate_kwargs = {
                "allowDiskUse": allow_disk_use,
                "batchSize": batch_size
            }
            
            if max_time_ms is not None:
                aggregate_kwargs["maxTimeMS"] = max_time_ms
            
            cursor = collection.aggregate(
                modified_pipeline,
                **aggregate_kwargs
            )
            return list(cursor)
    except Exception as e:
        print(f"{method_name} 批量聚合失败，回退到原始方法: {str(e)}")
        # 回退需要在各个调用方实现
        raise


# ================= 业务代码 =================

# 根据dataSetId查询projectId
class select_meta_dataset:
    def __init__(self, mongodb_cfg):  #libname="readInfo"
        self.libname = mongodb_cfg  #libname
        self.res = Pymongoengine(self.libname)

    def SelectMetaDataset (self,dataSetId,VariableID=None):
        self.res.connect_db()
        pipeline = [
            {"$match": {"_id": ObjectId(dataSetId)}}
        ]

        class meta_dataset(DynamicDocument):
            projectId = StringField(required=True, max_length=100)
            createUserId = StringField(required=True, max_length=100)
            createUserName = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        retProj = meta_dataset(self).objects(_id=ObjectId(dataSetId)).aggregate(*pipeline, allowDiskUse=True)
        return retProj

    def close(self):
        self.res.disconnect()


# 根据问卷id查询变量
class select_meta_variable:
    def __init__(self, mongodb_cfg):  #libname="readInfo"
        self.libname = mongodb_cfg  #libname
        self.res = Pymongoengine(self.libname)

    def SelectMetaVariable (self,PlatID,ProjID,dataSetId,version,VariableID=None):
        self.res.connect_db()
        pipeline = [
            {"$match":{"dataSetId": dataSetId}}
        ]

        if VariableID:
            pipeline = [
                {"$match": {"dataSetId": dataSetId, "varName": VariableID}}
            ]

        if ProjID == "1":
            class meta_variable(DynamicDocument):
                platId = StringField(required=True, max_length=100)
                projId = StringField(required=True, max_length=100)
                dataSetId = StringField(required=True, max_length=100)
                meta = {"db_alias": self.res._alias_pid}
            if version and version is not None:
                ret = meta_variable.objects(dataSetId=dataSetId, version=version).aggregate(
                    *pipeline, allowDiskUse=True)
            else:
                ret = meta_variable.objects(dataSetId=dataSetId).aggregate(
                    *pipeline, allowDiskUse=True)
        else:
            class meta_questionnaire_varial(DynamicDocument):
                projId = StringField(required=True, max_length=100)
                dataSetId = StringField(required=True, max_length=100)
                meta = {"db_alias": self.res._alias_pid}
            if version and version is not None:
                ret = meta_questionnaire_varial.objects(dataSetId=dataSetId, version=version).aggregate(
                    *pipeline, allowDiskUse=True)
            else:
                ret = meta_questionnaire_varial.objects(dataSetId=dataSetId).aggregate(
                    *pipeline, allowDiskUse=True)
        return ret
    def close(self):
        self.res.disconnect()


class mongoTemplateModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)
    def SelectTemplate(self, PlatID, ProjID, TempID):
        try:
            self.res.connect_db()
            pipeline = [
                {"$match": {"tempId": TempID}}
            ]
            class report_template_config(DynamicDocument):
                platId = StringField(required=True, max_length=100)
                tempId = StringField(required=True, max_length=100)
                meta = {"db_alias": self.res._alias_pid}
            retTemp = report_template_config.objects(tempId=TempID).aggregate(*pipeline, allowDiskUse=True)
            return retTemp
        except Exception as e:
            print(str(e))
            return None
    def SelectData(self, dataSetFlag, ProjID, dataSetId):
        self.res.connect_db()
        pipeline = [
            {"$match": {"dataSetId": dataSetId}}
        ]
        dataSetIdSave = dataSetId
        if dataSetFlag == 'vw':
            class vw_data_case_set(DynamicDocument):
                meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                projId = StringField(required=True, max_length=100)
                dataSetId = StringField(required=True, max_length=100)

            if version and version is not None:
                ret_data_case = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                               allowDiskUse=True)
            else:
                ret_data_case = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
        elif dataSetFlag == 'True':
            class data_case_set(DynamicDocument):
                meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                dataSetId = StringField(required=True, max_length=100)
                version = IntField(required=True)
            ret_data_case = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
        else:
            class data_case(DynamicDocument):
                PID = StringField(required=True, max_length=100)
                platId = StringField(required=True, max_length=100)
                projId = StringField(required=True, max_length=100)
                dataSetId = StringField(required=True, max_length=100)
                version = IntField(required=True)
                meta = {"db_alias": self.res._alias_pid}
            ret_data_case = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
        return ret_data_case
    def close(self):
        self.res.disconnect()


class mongoTemplateBackModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)
    def SelectTemplateBack(self, PlatID, ProjID, TempID):
        self.res.connect_db()
        class report_template_bak(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        pipeline = [
            {"$match": {"tempId": TempID}}
        ]

        # ret = client._get_collection("report_template_config").aggregate(*pipeline)
        retTemp = report_template_bak.objects(tempId=TempID).aggregate(
            *pipeline, allowDiskUse=True)
        return retTemp
    def delete(self, PlatID, ProjID, TempID):
        self.res.connect_db()

        class report_template_bak(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        report_template_bak.objects.filter(tempId=TempID).delete()
    def insert(self, data):
        self.res.connect_db()

        class report_template_bak(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        temp_insert = report_template_bak(**data)
        temp_insert.save()
        return True
    def close(self):
        self.res.disconnect()


class mongoTemplateFileModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)
    def SelectTemplateFile(self, PlatID, ProjID, TempID):
        self.res.connect_db()
        pipeline = [
            {"$match": {"_id": ObjectId(TempID)}}
        ]
        class report_template_file(DynamicDocument):
            userId = StringField(required=True, max_length=100)
            name = StringField(required=True, max_length=1000)
            templateFiles = ListField(required=True)
            meta = {"db_alias": self.res._alias_pid}
        retTemp = report_template_file.objects(_id=ObjectId(TempID)).aggregate(
            *pipeline, allowDiskUse=True)
        return retTemp
    def update(self, TempID, templateFiles):
        self.res.connect_db()
        class report_template_file(DynamicDocument):
            userId = StringField(required=True, max_length=100)
            name = StringField(required=True, max_length=1000)
            templateFiles = ListField(required=True)
            meta = {"db_alias": self.res._alias_pid}
        report_template_file.objects.filter(_id=ObjectId(TempID)).update(templateFiles=templateFiles)
    def close(self):
        self.res.disconnect()


class mongoAnalysisModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)
    def SelectAnalysis(self, PlatID, ProjID, TempID):
        self.res.connect_db()
        pipeline = [
            {"$match": {"tempId": TempID}}
        ]

        class data_analysis(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            reportId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        retAnalysis = data_analysis.objects(tempId=TempID).aggregate(
            *pipeline, allowDiskUse=True)
        return retAnalysis
    def SelectAnalysisReport(self, PlatID, ProjID, TempID, RepID):
        self.res.connect_db()

        class data_analysis(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            reportId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        pipeline = [
            {"$match": {"tempId": TempID}}
        ]
        ret_data_analysis = data_analysis.objects(tempId=TempID, reportId=RepID).aggregate(
            *pipeline, allowDiskUse=True)
        return ret_data_analysis
    def delete(self, PlatID, ProjID, TempID):
        self.res.connect_db()

        class data_analysis(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            reportId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        data_analysis.objects.filter(tempId=TempID).delete()
    def deleteReport(self, PlatID, ProjID, TempID, RepID):
        self.res.connect_db()

        class data_analysis(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            reportId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        data_analysis.objects.filter(tempId=TempID, reportId=RepID).delete()
    def insert(self, data):
        self.res.connect_db()

        class data_analysis(DynamicDocument):
            platId = StringField(required=True, max_length=100)
            projId = StringField(required=True, max_length=100)
            tempId = StringField(required=True, max_length=100)
            reportId = StringField(required=True, max_length=100)
            meta = {"db_alias": self.res._alias_pid}
        analysis_insert = data_analysis(**data)
        analysis_insert.save()
        return True
    def close(self):
        self.res.disconnect()


class dataversion:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def version(self,PlatID,ProjID,dataSetId):
        dataSetIdSave = dataSetId
        class vw_data_case_set(DynamicDocument):
            meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
            projId = StringField(required=True, max_length=100)
            dataSetId = StringField(required=True, max_length=100)
        class data_case_set(DynamicDocument):
            meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
            projId = StringField(required=True, max_length=100)
            dataSetId = StringField(required=True, max_length=100)
            version = IntField(required=True)
        self.res.connect_db()
        projectKeys, projectValues = ["version"], [1]
        projects = json.dumps(dict(zip(projectKeys, projectValues)))
        pipeline=[
            {"$project": json.loads(projects)},
            {"$group": {"_id":0,"version": {"$max":"$version"}}}
        ]
        num_data_case = data_case_set.objects(dataSetId=dataSetId).count()
        num_vw_data_case = vw_data_case_set.objects(dataSetId=dataSetId).count()
        if num_vw_data_case > 0:
            dataSetFlag = 'vw'
            ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
        else:
            if num_data_case > 0:
                dataSetFlag = 'True'
                ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                dataSetFlag = 'False'

                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
        return ret, dataSetFlag

    def close(self):
        self.res.disconnect()


# -----<频次分析>----- #
# noinspection PyProtectedMember
class mongoFrequencyModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def CountMongo ( self,dataSetFlag,ProjID,dataSetId,version,VariableID,weight,CasesCondition=None ):
        """
        优化的频数统计查询 - 使用批量聚合提升性能
        
        优化点：
        1. 使用统一的优化工具函数
        2. 优化聚合管道参数
        3. 添加查询性能监控
        4. 确保allowDiskUse=True支持大数据量
        """
        self.res.connect_db()
        
        if VariableID == 'M2Q3':
            testFlag = True
        project, projectToNum, VariableSelect, matchKeys, matchValues, VariableSort, count = \
            [],[], "", [], [],{}, 1
        if isinstance(VariableID, list):
            projectValuesItem = []
            for rowIDItem in VariableID:
                projectValuesItem.append("$" + rowIDItem)
            project = [VariableID[0]]
            projectToNum = [{"$concat": projectValuesItem}]
            VariableSelect = "$" + VariableID[0]
            VariableSort[VariableID[0]] = 1
        else:
            if VariableID.find(",") > -1:
                VariableGroup,VariableSelect = VariableID.split(','), {}
                for VariableItem in VariableGroup:
                    VariableSelect[VariableItem], VariableSort[VariableItem] = "$" + VariableItem, 1
                    project.append(VariableItem)
                    projectToNum.append(1)
            else:
                VariableSelect,VariableSort[VariableID], project, projectToNum = "$" + VariableID, 1, [VariableID], [1]
        if weight != "":
            project.append(weight)
            projectToNum.append({"$toDouble" : "$" + weight})
            matchKeys.append(weight)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            count = "$" + weight
            matchs = dict(zip(matchKeys, matchValues))
            pipeline = [
                {"$match": matchs}, {"$project": dict(zip(project, projectToNum))},
                {"$group": {"_id": VariableSelect, "count": {"$sum": count}}}, {"$sort": VariableSort}
            ]
        else:
            pipeline = [
                {"$project": dict(zip(project, projectToNum))},
                {"$group":{"_id": VariableSelect, "count": {"$sum": 1}}},
                {"$sort" : VariableSort}
            ]
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, project2, projectToNum2, matchKeys2, matchValues2, groupKeys, groupValues = \
                conditionReset(conditionGroup)
            project.extend(project2)
            projectToNum.extend(projectToNum2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            CasesConditionStr = json.dumps(dict(zip(CasesCondition[0], CasesCondition[1])), ensure_ascii=False)
            for projectOldIndex, projectOldItem in enumerate(projectOld):
                CasesConditionStr = CasesConditionStr.replace('"' + projectOldItem + '"',
                                                              '"' + project2[projectOldIndex] + '"', 1)
            conditioncase = json.loads(CasesConditionStr)
            matchs = dict(zip(matchKeys, matchValues))
            pipeline = [
                {"$match": matchs}, {"$project": dict(zip(project, projectToNum))},
                {"$match": conditioncase}, {"$group": {"_id": VariableSelect, "count": {"$sum": count}}},
                {"$sort": VariableSort}
            ]
        dataSetIdSave = dataSetId
        
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='CountMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"CountMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            
            return ret

    def CountMongoMultiple(self, dataSetFlag, ProjID, dataSetId, version, VariableIDList, weight, CasesCondition=None):
        """
        批量查询多个变量的计数结果（单次数据库调用）
        :param dataSetFlag: 数据集标识
        :param ProjID: 项目ID
        :param dataSetId: 数据集ID
        :param version: 版本号
        :param VariableIDList: 变量ID列表
        :param weight: 权重字段
        :param CasesCondition: 查询条件
        :return: dict {字段名: 查询结果列表}
        """
        self.res.connect_db()
        # 添加版本过滤条件
        query_filter = {}
        if version and version is not None and dataSetFlag != 'vw':
            query_filter['version'] = version
        
        # 构建基础匹配条件
        match_keys = []
        match_values = []
        if weight != "":
            match_keys.append(weight)
            match_values.append({"$nin": ["", "None", "null", "#NULL!", None]})
        
        # 处理 CasesCondition
        project = []
        projectToNum = []
        conditioncase = {}
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, project2, projectToNum2, matchKeys2, matchValues2, groupKeys, groupValues = \
                conditionReset(conditionGroup)
            match_keys.extend(matchKeys2)
            match_values.extend(matchValues2)
            project.extend(project2)
            projectToNum.extend(projectToNum2)
            
            CasesConditionStr = json.dumps(dict(zip(CasesCondition[0], CasesCondition[1])), ensure_ascii=False)
            for projectOldIndex, projectOldItem in enumerate(projectOld):
                CasesConditionStr = CasesConditionStr.replace('"' + projectOldItem + '"',
                                                              '"' + project2[projectOldIndex] + '"', 1)
            conditioncase = json.loads(CasesConditionStr)
        
        base_match = dict(zip(match_keys, match_values)) if match_keys else {}
        
        # 合并所有匹配条件
        if query_filter:
            base_match.update(query_filter)
        
        # 构建 $facet 管道
        facet_stages = {}
        
        for VariableID in VariableIDList:
            VariableSort = {VariableID: 1}
            
            # 构建子管道
            sub_pipeline = []
            
            if base_match:
                sub_pipeline.append({"$match": base_match})
            
            # 处理 CasesCondition 和 weight 的 $project
            if isinstance(CasesCondition, list) or weight != "":
                project_var = [VariableID]
                projectToNum_var = [1]
                
                if weight != "":
                    project_var.append(weight)
                    projectToNum_var.append({"$toDouble": "$" + weight})
                
                if isinstance(CasesCondition, list):
                    project_var.extend(project)
                    projectToNum_var.extend(projectToNum)
                    sub_pipeline.append({"$project": dict(zip(project_var, projectToNum_var))})
                    sub_pipeline.append({"$match": conditioncase})
                else:
                    sub_pipeline.append({"$project": dict(zip(project_var, projectToNum_var))})
            
            # 添加过滤空值的 match
            if weight != "":
                sub_pipeline.append({
                    "$match": {
                        VariableID: {"$nin": ["", "None", "null", "#NULL!", None], "$exists": True},
                        weight: {"$nin": ["", "None", "null", "#NULL!", None]}
                    }
                })
                count_expr = "$" + weight
            else:
                sub_pipeline.append({
                    "$match": {
                        VariableID: {"$nin": ["", "None", "null", "#NULL!", None], "$exists": True}
                    }
                })
                count_expr = 1
            
            sub_pipeline.append({
                "$group": {
                    "_id": "$" + VariableID,
                    "count": {"$sum": count_expr}
                }
            })
            sub_pipeline.append({"$sort": VariableSort})
            
            facet_stages[VariableID] = sub_pipeline
        
        # 使用 $facet 在单次查询中执行所有分支
        pipeline = [{"$facet": facet_stages}]
        
        dataSetIdSave = dataSetId
        result_dict = {}

        try:
            # 使用优化的聚合查询
            ret = execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='CountMongoMultiple',
                batch_size=10000
            )
            # 解析 $facet 返回的结果
            if ret and len(ret) > 0:
                facet_result = ret[0]
                for VariableID in VariableIDList:
                    result_dict[VariableID] = facet_result.get(VariableID, [])
            else:
                for VariableID in VariableIDList:
                    result_dict[VariableID] = []
        except Exception as e:
            print(f"CountMongoMultiple批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                
                if version and version is not None:
                    ret = list(vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True))
                else:
                    ret = list(vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True))
                
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                
                if version and version is not None:
                    ret = list(data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True))
                else:
                    ret = list(data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True))
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = list(data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True))
                else:
                    ret = list(data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True))
            
            # 解析 $facet 返回的结果
            if ret and len(ret) > 0:
                facet_result = ret[0]
                for VariableID in VariableIDList:
                    result_dict[VariableID] = facet_result.get(VariableID, [])
            else:
                for VariableID in VariableIDList:
                    result_dict[VariableID] = []
        
        return result_dict

    def close(self):
        self.res.disconnect()


# -----<均值>----- #
class mongoMeansModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def SumMongo (self,dataSetFlag,ProjID,dataSetId,version,VariableID,VariableIDgroup,divide,weight,
                  CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, totalValues, matchKeys, matchValues, totalValuesItem, projectKeysVariable, \
        conditioncase = ['_id'], [0], [], [] ,[], [], "", None
        if str(VariableID).strip() != "":
            projectKeys.append(VariableID)
            projectValues.append(1)
            if weight != "":
                projectKeys.append(weight)
                projectValues.append(1)
                matchKeys.append(weight)
                matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        else:
            for variableItem in VariableIDgroup:
                variableMore, functionStr = [], ""
                totalValuesItem.clear()
                if str(variableItem).find("*") > -1:
                    variableMore = str(variableItem).split('*')
                    functionStr = "$multiply"
                elif str(variableItem).find("/") > -1:
                    variableMore = str(variableItem).split('/')
                    functionStr = "$divide"
                elif str(variableItem).find("+") > -1:
                    variableMore = str(variableItem).split('+')
                    functionStr = "$add"
                elif str(variableItem).find("-") > -1:
                    variableMore = str(variableItem).split('-')
                    functionStr = "$subtract"
                else:
                    matchKeys.append(variableItem)
                    matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                    totalValues.append({"$toDouble": "$" + variableItem})
                if len(variableMore) > 0:
                    for variableMoreItem in variableMore:
                        if is_number(variableMoreItem):
                            totalValuesItem.append(float(variableMoreItem))
                        else:
                            matchKeys.append(variableMoreItem)
                            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                            totalValuesItem.append({"$toDouble": "$" + variableMoreItem})
                    totalValues.append({functionStr: copy.deepcopy(totalValuesItem)})
                    if projectKeysVariable == "":projectKeysVariable = variableMore[0]
            if projectKeysVariable == "":projectKeysVariable = VariableIDgroup[0]
            projectKeys.append(projectKeysVariable + "data")
            if divide == "" or divide == "0":
                projectValues.append({"$add": totalValues})
            else:
                projectValues.append({"$divide": [{"$add": totalValues}, float(divide)]})
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys, groupValues = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
        projects = dict(zip(projectKeys, projectValues))
        matchs = dict(zip(matchKeys, matchValues))
        if conditioncase is None:
            pipeline = [{"$match": matchs}, {"$project": projects}]
        else:
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match": conditioncase}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='SumMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"SumMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                if version and version is not None:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    sortData = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                     allowDiskUse=True)
                else:
                    sortData = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    sortData = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                 allowDiskUse=True)
                else:
                    sortData = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return sortData
    def SumMongoAvg (self,dataSetFlag,ProjID,dataSetId,version,VariableIDgroup,optionIDgroup,divide,weight,
                     CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, groupKeys, groupValues, totalValues, nowString, matchKeys, matchValues,\
        totalValuesItem, projectKeysVariable, count =  \
            ['_id'], [0], ["_id"], ["null"], [], datetime.datetime.now().strftime('%H%M%S%f'), [], [], [], "", 1
        if isinstance(optionIDgroup, str) and optionIDgroup == "":
            for variableItem in VariableIDgroup:
                variableMore, functionStr = [], ""
                totalValuesItem.clear()
                if str(variableItem).find("*") > -1:
                    variableMore = str(variableItem).split('*')
                    functionStr = "$multiply"
                elif str(variableItem).find("/") > -1:
                    variableMore = str(variableItem).split('/')
                    functionStr = "$divide"
                elif str(variableItem).find("+") > -1:
                    variableMore = str(variableItem).split('+')
                    functionStr = "$add"
                elif str(variableItem).find("-") > -1:
                    variableMore = str(variableItem).split('-')
                    functionStr = "$subtract"
                else:
                    matchKeys.append(variableItem)
                    matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                    totalValues.append({"$toDouble": "$" + variableItem})
                if len(variableMore) > 0:
                    for variableMoreItem in variableMore:
                        if is_number(variableMoreItem):
                            totalValuesItem.append(float(variableMoreItem))
                        else:
                            matchKeys.append(variableMoreItem)
                            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                            totalValuesItem.append({"$toDouble": "$" + variableMoreItem})
                    totalValues.append({functionStr: copy.deepcopy(totalValuesItem)})
                    if projectKeysVariable == "": projectKeysVariable = variableMore[0]
            if projectKeysVariable == "":projectKeysVariable = VariableIDgroup[0]
        else:
            projectKeysVariable = VariableIDgroup[0]
            matchKeys.append(VariableIDgroup[0])
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            matchKeys.append(VariableIDgroup[1])
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            for optionItem in optionIDgroup:
                if 0 < int(optionItem) <= 150:
                    totalValues.append({"$toDouble":{"$substr": ["$" + VariableIDgroup[0], int(optionItem) - 1, 1]}})
                else:
                    totalValues.append({"$toDouble": {"$substr": ["$" + VariableIDgroup[1], int(optionItem) - 151, 1]}})
        if weight != "":
            projectKeys.append(weight)
            projectValues.append({"$toDouble": "$" + weight})
            matchKeys.append(weight)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            count = "$" + weight
        projectKeys.append(projectKeysVariable + nowString)
        if divide == "" or divide == "0":
            projectValues.append({"$add": totalValues})
        else:
            projectValues.append({"$divide": [{"$add": totalValues}, float(divide)]})
        groupKeys.append("count" )
        groupValues.append({"$sum": count})
        groupKeys.append(projectKeysVariable + "avg")
        groupValues.append({"$avg" : "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "sum")
        groupValues.append({"$sum": "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "min")
        groupValues.append({"$min": "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "max")
        groupValues.append({"$max": "$" + projectKeysVariable + nowString})
        projects = dict(zip(projectKeys, projectValues))
        groups = dict(zip(groupKeys, groupValues))
        matchs = dict(zip(matchKeys, matchValues))
        pipeline = [{"$match": matchs}, {"$project": projects}, {"$group": groups}]
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            groupKeys.extend(groupKeys2)
            groupValues.extend(groupValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
            projects = dict(zip(projectKeys, projectValues))
            groups = dict(zip(groupKeys, groupValues))
            matchs = dict(zip(matchKeys, matchValues))
            pipeline = [{"$match": matchs}, {"$project": projects},{"$match": conditioncase}, {"$group": groups}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='SumMongoAvg',
                batch_size=10000
            )
        except Exception as e:
            print(f"SumMongoAvg批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                        allowDiskUse=True)
                else:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    sortData = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                 allowDiskUse=True)
                else:
                    sortData = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    sortData = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                 allowDiskUse=True)
                else:
                    sortData = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return sortData
    def SumMongoCompareAvg (self,dataSetFlag,ProjID,dataSetId,version,rowID,VariableIDgroup,optionIDgroup,weight,
                            CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, groupKeys, totalValues,nowString, matchKeys, matchValues, totalValuesItem = \
            ['_id'], [0], ["_id"], [], datetime.datetime.now().strftime('%H%M%S%f'), [], [], []
        projectKeysVariable, conditioncase, count = "", None, 1
        if isinstance(rowID, list):
            groupValues = [{rowID[0]:"$" + rowID[0]}]
        else:
            groupValues = [{rowID:"$" + rowID}]
        if isinstance(optionIDgroup, str) and optionIDgroup == "":
            for variableItem in VariableIDgroup:
                totalValuesItem.clear()
                variableMore, functionStr = [], ""
                if str(variableItem).find("*") > -1:
                    variableMore = str(variableItem).split('*')
                    functionStr = "$multiply"
                elif str(variableItem).find("/") > -1:
                    variableMore = str(variableItem).split('/')
                    functionStr = "$divide"
                elif str(variableItem).find("+") > -1:
                    variableMore = str(variableItem).split('+')
                    functionStr = "$add"
                elif str(variableItem).find("-") > -1:
                    variableMore = str(variableItem).split('-')
                    functionStr = "$subtract"
                else:
                    matchKeys.append(variableItem)
                    matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                    totalValues.append({"$toDouble": "$" + variableItem})
                if len(variableMore) > 0:
                    for variableMoreItem in variableMore:
                        if is_number(variableMoreItem):
                            totalValuesItem.append(float(variableMoreItem))
                        else:
                            matchKeys.append(variableMoreItem)
                            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                            totalValuesItem.append({"$toDouble": "$" + variableMoreItem})
                    totalValues.append({functionStr: copy.deepcopy(totalValuesItem)})
                    if projectKeysVariable == "": projectKeysVariable = variableMore[0]
            if projectKeysVariable == "":projectKeysVariable = VariableIDgroup[0]
        else:
            projectKeysVariable = VariableIDgroup[0]
            matchKeys.append(VariableIDgroup[0])
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            matchKeys.append(VariableIDgroup[1])
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            for optionItem in optionIDgroup:
                if 0 < int(optionItem) <= 150:
                    totalValues.append({"$toDouble":{"$substr": ["$" + VariableIDgroup[0], int(optionItem) - 1, 1]}})
                else:
                    totalValues.append({"$toDouble":{"$substr": ["$" + VariableIDgroup[1], int(optionItem) - 151, 1]}})
        if isinstance(rowID, list):
            projectValuesItem = []
            for rowIDItem in rowID:
                projectValuesItem.append("$" + rowIDItem)
            projectKeys.append(rowID[0])
            projectValues.append({"$concat": projectValuesItem})
        else:
            projectKeys.append(rowID)
            projectValues.append(1)
        if weight != "":
            projectKeys.append(weight)
            projectValues.append({"$toDouble": "$" + weight})
            matchKeys.append(weight)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            count = "$" + weight
        projectKeys.append(projectKeysVariable + nowString)
        projectValues.append({"$add": totalValues})
        groupKeys.append("count")
        groupValues.append({"$sum": count})
        groupKeys.append(projectKeysVariable + "avg")
        groupValues.append({"$avg" : "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "sum")
        groupValues.append({"$sum": "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "min")
        groupValues.append({"$min": "$" + projectKeysVariable + nowString})
        groupKeys.append(projectKeysVariable + "max")
        groupValues.append({"$max": "$" + projectKeysVariable + nowString})
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            groupKeys.extend(groupKeys2)
            groupValues.extend(groupValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
        projects = dict(zip(projectKeys, projectValues))
        groups = dict(zip(groupKeys, groupValues))
        matchs = dict(zip(matchKeys, matchValues))
        if conditioncase is None:
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$group": groups}]
        else:
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match": conditioncase}, {"$group": groups}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='SumMongoCompareAvg',
                batch_size=10000
            )
        except Exception as e:
            print(f"SumMongoCompareAvg批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                        allowDiskUse=True)
                else:
                    sortData = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    sortData = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                 allowDiskUse=True)
                else:
                    sortData = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    sortData = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                 allowDiskUse=True)
                else:
                    sortData = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return sortData
    def close(self):
        self.res.disconnect()


# -----<列联表分析>----- #
class mongoCrosstabsModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def CrosstabsCountMongo (self,dataSetFlag,ProjID,dataSetId,version,rowID,columnID,weight,CasesCondition=None):
        self.res.connect_db()
        matchKeys, matchValues,projectKeys, projectValues, matchKeys, matchValues, groupKeys, groupValues, sortKeys, \
        count,conditioncase  = [], [], [], [], [], [], [rowID], ["$" + rowID], {rowID:1}, 1, None
        matchKeys.append(rowID),projectKeys.append(rowID)
        matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]}),projectValues.append(1)
        if str(columnID).find(",") > -1:
            columnMore = str(columnID).split(',')
            for columnItem in columnMore:
                matchKeys.append(columnItem), projectKeys.append(columnItem)
                matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]}), projectValues.append(1)
                groupKeys.append(columnItem)
                groupValues.append("$" + columnItem)
                sortKeys[columnItem] = 1
        else:
            matchKeys.append(columnID),projectKeys.append(columnID)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]}),projectValues.append(1)
            groupKeys.append(columnID)
            groupValues.append("$" + columnID)
            sortKeys[columnID] = 1
        if weight != "":
            projectKeys.append(weight)
            projectValues.append({"$toDouble" : "$" + weight})
            matchKeys.append(weight)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            count = "$" + weight
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            groupKeys.extend(groupKeys2)
            groupValues.extend(groupValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
        matchs = dict(zip(matchKeys, matchValues))
        projects = dict(zip(projectKeys, projectValues))
        groups = dict(zip(groupKeys, groupValues))
        if conditioncase is None:
            pipeline = [{"$match": matchs}, {"$project": projects},
                        {"$group": {"_id": groups, "count": {"$sum": count}}}, {"$sort": sortKeys}]
        else:
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match":conditioncase},
                            {"$group":{"_id": groups,"count": {"$sum": count}}}, {"$sort" : sortKeys}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='CrosstabsCountMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"CrosstabsCountMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                        allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId,version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId,version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            # print(ret)
            return ret
    def close(self):
        self.res.disconnect()


# -----<比较均值>----- #
class mongoCompareMeansModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def CompareSumMongo(self, dataSetFlag, ProjID, dataSetId, version,rowID, columnID, CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, matchKeys, matchValues, conditioncase = [], [], [], [], None
        matchKeys.append(rowID), projectKeys.append(rowID)
        matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]}), projectValues.append(1)
        matchKeys.append(columnID), projectKeys.append(columnID)
        matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]}), projectValues.append(1)
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
        matchs = dict(zip(matchKeys, matchValues))
        projects = dict(zip(projectKeys, projectValues))
        if conditioncase is not None:
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match": conditioncase},
                        {"$group": {"_id": {rowID: "$" + rowID, columnID: "$" + columnID}, "count": {"$sum": 1}}},
                        {"$sort": {rowID: 1, columnID: 1}}]
        else:
            pipeline = [{"$match": matchs}, {"$project": projects},
                        {"$group": {"_id": {rowID: "$" + rowID, columnID: "$" + columnID}, "count": {"$sum": 1}}},
                        {"$sort": {rowID: 1, columnID: 1}}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='CompareSumMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"CompareSumMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return ret
    def sortDataMongo (self,dataSetFlag,ProjID,dataSetId,version,rowID,columnID,optionID,VariableIDgroup,weight,
                       CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, matchKeys, matchValues = \
            ['_id', rowID], [0, 1], [], []
        matchKeys.append(rowID)
        matchValues.append(optionID)
        if str(columnID).strip() != "":
            projectKeys.append(columnID)
            projectValues.append(1)
            matchKeys.append(columnID)
            matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
            if weight != "":
                projectKeys.append(weight)
                projectValues.append(1)
                matchKeys.append(weight)
                matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        else:
            for variableItem in VariableIDgroup:
                projectKeys.append(variableItem)
                projectValues.append(1)
        projects = dict(zip(projectKeys, projectValues))
        matchs = dict(zip(matchKeys, matchValues))
        pipeline = [{"$match": matchs}, {"$project": projects}]
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
            projects = dict(zip(projectKeys, projectValues))
            matchs = dict(zip(matchKeys, matchValues))
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match": conditioncase}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='sortDataMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"sortDataMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return ret
    def close(self):
        self.res.disconnect()


# -----<排序题分析>----- #
class mongoSortModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def SortMongo (self,dataSetFlag,ProjID,dataSetId,version,VariableID1,VariableID2,VariableID3,CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, matchKeys, matchValues = \
            [VariableID1, VariableID2, VariableID3], [1, 1, 1], [], []
        projects = dict(zip(projectKeys, projectValues))
        pipeline = [{"$project": projects},
                    {"$group":{"_id":{VariableID1:"$" + VariableID1, VariableID2:"$" + VariableID2,
                                      VariableID3:"$" + VariableID3},"count": {"$sum": 1}}},
                    {"$sort" : {VariableID1:1}}]
        if isinstance(CasesCondition, list):
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys2, projectValues2, matchKeys2, matchValues2, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            projectKeys.extend(projectKeys2)
            projectValues.extend(projectValues2)
            matchKeys.extend(matchKeys2)
            matchValues.extend(matchValues2)
            conditioncase = dict(zip(projectKeys2, CasesCondition[1]))
            projects = dict(zip(projectKeys, projectValues))
            matchs = dict(zip(matchKeys, matchValues))
            pipeline = [{"$match": matchs}, {"$project": projects}, {"$match":conditioncase},
                        {"$group": {"_id": {VariableID1: "$" + VariableID1, VariableID2: "$" + VariableID2,
                                            VariableID3: "$" + VariableID3}, "count": {"$sum": 1}}},
                        {"$sort": {VariableID1: 1}}]
        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='SortMongo',
                batch_size=10000
            )
        except Exception as e:
            print(f"SortMongo批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId,version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId,version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            # print(ret)
            return ret
    def close(self):
        self.res.disconnect()


# -----<线性回归>----- #
class mongoLinearRegressionAnalyModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def showDatas(self, dataSetFlag, ProjID, dataSetId, version, xVariableID, yVariableID, CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, projectKeys2, projectValues2, mkeys, mvalues = [], [], [], [], [], []
        if CasesCondition:
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys, projectValues, mkeys, mvalues, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            CasesConditionStr = json.dumps(dict(zip(CasesCondition[0], CasesCondition[1])), ensure_ascii=False)
            for projectOldIndex, projectOldItem in enumerate(projectOld):
                CasesConditionStr = CasesConditionStr.replace('"' + projectOldItem + '"',
                                                              '"' + projectKeys[projectOldIndex] + '"', 1)
            CasesCondition = json.loads(CasesConditionStr)
        projectKeys.append('_id'), projectKeys.append(xVariableID)
        projectValues.append(0), projectValues.append(1)
        projectKeys2.append('_id'), projectKeys2.append(xVariableID)
        projectValues2.append(0), projectValues2.append(1)
        mkeys.append(xVariableID)
        mvalues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        if isinstance(yVariableID, list):
            for variableItem in yVariableID:
                projectKeys.append(variableItem)
                projectValues.append(1)
                projectKeys2.append(variableItem)
                projectValues2.append(1)
                mkeys.append(variableItem)
                mvalues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        else:
            projectKeys.append(yVariableID)
            projectValues.append(1)
            projectKeys2.append(yVariableID)
            projectValues2.append(1)
            mkeys.append(yVariableID)
            mvalues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        matchs = json.dumps(dict(zip(mkeys, mvalues)))
        pipeline = [{"$match": json.loads(matchs)}]
        projects = json.dumps(dict(zip(projectKeys, projectValues)))
        pipeline.append({"$project": json.loads(projects)})
        if CasesCondition:
            pipeline.append({"$match": CasesCondition})
            pipeline.append({"$project": dict(zip(projectKeys2, projectValues2))})

        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='showDatas',
                batch_size=10000
            )
        except Exception as e:
            print(f"showDatas批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return ret
    def close(self):
        self.res.disconnect()


# -----<相关分析>----- #
class mongoCorrelationAnalysisModel:
    def __init__(self, mongodb_cfg):  # libname="readInfo"
        self.libname = mongodb_cfg  # libname
        self.res = Pymongoengine(self.libname)

    def showDatas(self, dataSetFlag, ProjID, dataSetId, version, yVariableIDs, CasesCondition=None):
        self.res.connect_db()
        projectKeys, projectValues, projectKeys2, projectValues2, mkeys, mvalues = [], [], [], [], [], []

        if CasesCondition:
            conditionGroup = []
            for keyIndex, keyCondition in enumerate(CasesCondition[0]):
                valueCondition = CasesCondition[1][keyIndex]
                if keyCondition == '$and' or keyCondition == '$nor' or keyCondition == '$or':
                    for valueConditionItem in valueCondition:
                        if isinstance(valueConditionItem, dict):
                            if '$and' in valueConditionItem.keys() or '$nor' in valueConditionItem.keys() \
                                    or '$or' in valueConditionItem.keys():
                                for keyValueConditionItem, valueValueConditionItem in valueConditionItem.items():
                                    conditionGroup.append(valueValueConditionItem)
                            else:
                                conditionGroup.append(valueConditionItem)
                else:
                    conditionGroup.append({keyCondition: valueCondition})
            projectOld, projectKeys, projectValues, mkeys, mvalues, groupKeys2, groupValues2 = \
                conditionReset(conditionGroup)
            CasesConditionStr = json.dumps(dict(zip(CasesCondition[0], CasesCondition[1])), ensure_ascii=False)
            for projectOldIndex, projectOldItem in enumerate(projectOld):
                CasesConditionStr = CasesConditionStr.replace('"' + projectOldItem + '"',
                                                              '"' + projectKeys[projectOldIndex] + '"', 1)
            CasesCondition = json.loads(CasesConditionStr)
        projectKeys.append('_id'), projectValues.append(0)
        projectKeys2.append('_id'), projectValues2.append(0)
        if isinstance(yVariableIDs, list):
            for variableItem in yVariableIDs:
                projectKeys.append(variableItem), projectValues.append(1)
                projectKeys2.append(variableItem), projectValues2.append(1)
                mkeys.append(variableItem), mvalues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        else:
            projectKeys.append(yVariableIDs), projectValues.append(1)
            projectKeys2.append(yVariableIDs), projectValues2.append(1)
            mkeys.append(yVariableIDs), mvalues.append({"$nin": ["", "None", "null", "#NULL!", None]})
        matchs = json.dumps(dict(zip(mkeys, mvalues)))
        pipeline = [{"$match": json.loads(matchs)}]
        projects = json.dumps(dict(zip(projectKeys, projectValues)))
        pipeline.append({"$project": json.loads(projects)})
        if CasesCondition:
            pipeline.append({"$match": CasesCondition})
            pipeline.append({"$project": dict(zip(projectKeys2, projectValues2))})

        dataSetIdSave = dataSetId
        try:
            # 使用优化的聚合查询
            return execute_optimized_aggregate(
                res=self.res,
                dataSetFlag=dataSetFlag,
                dataSetId=dataSetId,
                pipeline=pipeline,
                method_name='showDatas',
                batch_size=10000
            )
        except Exception as e:
            print(f"showDatas批量聚合失败，回退到原始方法: {str(e)}")
            # 回退到原始mongoengine方法
            if dataSetFlag == 'vw':
                class vw_data_case_set(DynamicDocument):
                    meta = {'collection': 'vw_data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)

                if version and version is not None:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,
                                                                                                   allowDiskUse=True)
                else:
                    ret = vw_data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            elif dataSetFlag == 'True':
                class data_case_set(DynamicDocument):
                    meta = {'collection': 'data_case_' + dataSetIdSave, "db_alias": self.res._alias_pid}
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                if version and version is not None:
                    ret = data_case_set.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline,allowDiskUse=True)
                else:
                    ret = data_case_set.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            else:
                class data_case(DynamicDocument):
                    PID = StringField(required=True, max_length=100)
                    platId = StringField(required=True, max_length=100)
                    projId = StringField(required=True, max_length=100)
                    dataSetId = StringField(required=True, max_length=100)
                    version = IntField(required=True)
                    meta = {"db_alias": self.res._alias_pid}
                if version and version is not None:
                    ret = data_case.objects(dataSetId=dataSetId, version=version).aggregate(*pipeline, allowDiskUse=True)
                else:
                    ret = data_case.objects(dataSetId=dataSetId).aggregate(*pipeline, allowDiskUse=True)
            return ret
    def close(self):
        self.res.disconnect()


def conditionReset(conditionGroup):
    projectOld, project, projectToNum, matchKeys, matchValues, groupKeys, groupValues, indexCondition, flagCondition = \
        [], [], [], [], [], [], [], 0, True
    while flagCondition:
        flagCondition = False
        for conditionGroupItem in conditionGroup:
            if isinstance(conditionGroupItem, list):
                flagCondition = True
                break
        if flagCondition: conditionGroup = conditionGroupReset(conditionGroup)
    for conditionGroupItem in conditionGroup:
        for keyCondition, valueCondition in conditionGroupItem.items():
            nowString = datetime.datetime.now().strftime('%H%M%S%f') + str(indexCondition)
            projectOld.append(keyCondition)
            project.append(keyCondition + copy.deepcopy(nowString))
            groupKeys.append(keyCondition + copy.deepcopy(nowString))
            if check_json_format(json.dumps(valueCondition)) or isinstance(valueCondition, dict):
                for keyFilter, valueFilter in valueCondition.items():
                    if isinstance(valueFilter, bool):
                        projectToNum.append("$" + keyCondition)
                    elif isinstance(valueFilter, int) or isinstance(valueFilter, float):
                        matchKeys.append(keyCondition)
                        matchValues.append({"$nin": ["", "None", "null", "#NULL!", None]})
                        projectToNum.append({"$toDouble": "$" + keyCondition})
                    else:
                        projectToNum.append("$" + keyCondition)
                    break
            else:
                projectToNum.append("$" + keyCondition)
            groupValues.append({"$max": "$" + keyCondition})
            indexCondition += 1
    return projectOld, project, projectToNum, matchKeys, matchValues, groupKeys, groupValues


def conditionGroupReset(conditionGroup):
    flagCondition = True
    while flagCondition:
        flagCondition = False
        for conditionGroupItem in conditionGroup:
            if isinstance(conditionGroupItem, list):
                flagCondition = True
                break
        if flagCondition:
            for conditionGroupItem in conditionGroup:
                if isinstance(conditionGroupItem, list):
                    groupIndex = conditionGroup.index(conditionGroupItem)
                    conditionGroup.remove(conditionGroupItem)
                    for conditionGroupItemIndex, conditionGroupItemValue in enumerate(conditionGroupItem):
                            groupIndex += conditionGroupItemIndex
                            conditionGroup.insert(groupIndex, conditionGroupItemValue)
    flagCondition = True
    while flagCondition:
        flagCondition = False
        for conditionGroupItem in conditionGroup:
            if not isinstance(conditionGroupItem, list) and (json.dumps(conditionGroupItem).find('$and') > -1 or
                                                             json.dumps(conditionGroupItem).find('$nor') > -1 or
                                                             json.dumps(conditionGroupItem).find('$or') > -1):
                flagCondition = True
                groupIndex = conditionGroup.index(conditionGroupItem)
                conditionGroup.remove(conditionGroupItem)
                for keyValueConditionItem, valueValueConditionItem in conditionGroupItem.items():
                    conditionGroup.insert(groupIndex, valueValueConditionItem)
    return conditionGroup


def check_json_format(raw_msg):
    if isinstance(raw_msg, str):
        try:
            json.loads(raw_msg)
            for key, value in json.loads(str(raw_msg).strip()).items():
                return True
        except Exception as e:
            pass
        return False
    else:
        return False


def is_number(s):
    try:  # 如果能运行float(s)语句，返回True（字符串s是浮点数）
        float(s)
        return True
    except ValueError:  # ValueError为Python的一种标准异常，表示"传入无效的参数"
        pass  # 如果引发了ValueError这种异常，不做任何事情（pass：不做任何事情，一般用做占位语句）
    try:
        import unicodedata  # 处理ASCii码的包
        for n in s:
            unicodedata.numeric(n)  # 把一个表示数字的字符串转换为浮点数返回的函数
            #return True
        return True
    except (TypeError, ValueError):
        pass
    return False


if __name__ == '__main__':
    # print(ret)
    PlatID = "readInfo"
    ProjID = "1"
    dataSetId = "5ee0835b23c32810e9305b5c"
    rowID="P2Q1"
    rowID2 = "P2Q7"
    VariableID = "zkeyUserInfoSYXMC"

    version=1
    versiondata=dataversion(PlatID)
    data=versiondata.version(PlatID, ProjID, dataSetId)
    for row in data:
        print(row)
        version=int(row["version"])

    variable=select_meta_variable(PlatID)
    variable.SelectMetaVariable(PlatID, ProjID, dataSetId, version,VariableID)

    frequency = mongoFrequencyModel(PlatID)
    data = frequency.CountMongo(PlatID, ProjID, dataSetId, version, VariableID,{'zkeyUserInfoSYXDM': '1001100'})
    i = 0
    for u in data:
        print(u)

    #print(json.dumps(data))