WTOAAM/dataAnalysisBehavior/behavior/outputProcessor.py

import os
from datetime import datetime, timezone, timedelta
from logging import Logger
import pandas as pd
from sqlalchemy.orm import Session
from sqlalchemy.sql import text
import shutil
from urllib.parse import quote, unquote
from common.commonBusiness import CommonBusiness
from common.appConfig import GetBusinessFoundationDbUtil
from algorithmContract.confBusiness import *
from algorithmContract.contract import Contract
from algorithmContract.const import *
from utils.minioUtil.threadSafeMinioClient import ThreadSafeMinioClient
from utils.rdbmsUtil.databaseUtil import DatabaseUtil

Const_FileURL = "url"


class OutputProcessor:
    def __init__(self, conf: Contract, logger: Logger, dbUtil: dict[str, DatabaseUtil], minioUtil: ThreadSafeMinioClient) -> None:
        self.conf = conf
        self.autoOrManual = 1 if self.conf.dataContract.autoOrManual == 'automatic' else 0
        self.powerFarmID = self.conf.dataContract.dataFilter.powerFarmID
        self.dataBatchNum = self.conf.dataContract.dataFilter.dataBatchNum

        self.logger = logger
        self.dbUtil = dbUtil
        self.minioUtil = minioUtil

        self.common = CommonBusiness()

    def uploadOfMioIO(self, bucketName: str, uploadFiles):
        """
        上传文件到minio
        """
        bucketName = bucketName.lower()
        if not self.minioUtil.bucket_exists(bucketName):
            self.minioUtil.create_bucket(bucketName)
            self.minioUtil.set_bucket_policy(bucketName)

        # Upload files
        upload_results = self.minioUtil.upload_files(bucketName, uploadFiles)

    def removeLocalFiles(self, powerFarmID: str, dataBatchNum: str):
        directory = f"output/{powerFarmID}/{dataBatchNum}"
        shutil.rmtree(directory)

    def analysisState(self, session: Session, batchNO: str, analysisState: int, errorState: int = ErrorState.NotErr.value, errorCode: str = None, errorInfo: str = None, timestamp: datetime = datetime.now(timezone.utc)+timedelta(hours=8), analysisProgress: float = 0):
        """
        写处理状态 至主表 analysis_result
        写总图（多机组一张图表）上传文件 至子表 analysis_general_file
        写单机组图（单机组一张图表）上传文件 至子表 analysis_diagram_relation
        """
        sql = text(f"INSERT INTO analysis_result(batch_code, analysis_state, err_state, err_code, err_info,  create_time,analysis_progress) \
                  VALUES(:batch_code, :analysis_state, :err_state,:err_code, :err_info, :create_time,:analysis_progress) \
                  ON DUPLICATE KEY \
                  UPDATE \
                  analysis_state=VALUES(analysis_state),err_state=VALUES(err_state),err_code=VALUES(err_code), \
                    err_info=VALUES(err_info),update_time=VALUES(update_time),analysis_progress=VALUES(analysis_progress);")

        params = {
            "batch_code": None if self.common.isNone(batchNO) else batchNO,
            "analysis_state": None if self.common.isNone(analysisState) else analysisState,
            "err_state": None if self.common.isNone(analysisState) else errorState,
            "err_code": None if self.common.isNone(errorCode) else errorCode,
            "err_info": None if self.common.isNone(errorInfo) else errorInfo,
            "create_time": timestamp,
            "update_time": timestamp,
            "analysis_progress": analysisProgress
        }

        session.execute(sql, params)

    def analysisResultForTurbine(self, session: Session, returnDataFrame: pd.DataFrame, timestamp: datetime):
        dataFrame = returnDataFrame[(returnDataFrame[Field_CodeOfTurbine] != 'total') & (
            returnDataFrame[Field_Return_IsSaveDatabase])]

        for index, row in dataFrame.iterrows():
            sql = text(f"""
                INSERT INTO analysis_diagram_relation 
                (batch_code, field_engine_code, analysis_type_code, file_addr, auto_analysis, create_time) 
                VALUES (:batch_code, :field_engine_code, :analysis_type_code, :file_addr, :auto_analysis, :create_time)
                ON DUPLICATE KEY UPDATE 
                field_engine_code=VALUES(field_engine_code),
                analysis_type_code=VALUES(analysis_type_code),
                file_addr=VALUES(file_addr),
                auto_analysis=VALUES(auto_analysis);
            """)

            params = {
                "batch_code": row[Field_Return_BatchCode],
                "field_engine_code": row[Field_CodeOfTurbine],
                "analysis_type_code": row[Field_Return_TypeAnalyst],
                "file_addr": row[Const_FileURL],
                "auto_analysis": self.autoOrManual,
                "create_time": timestamp
            }

            session.execute(sql, params)

    def analysisResultForTotal(self, session: Session, returnDataFrame: pd.DataFrame, timestamp: datetime):
        dataFrame = returnDataFrame[(returnDataFrame[Field_CodeOfTurbine] == 'total') & (
            returnDataFrame[Field_Return_IsSaveDatabase])]

        for index, row in dataFrame.iterrows():
            sql = text(f"""
                INSERT INTO analysis_general_file 
                (batch_code, analysis_type_code, file_addr, auto_analysis, create_time) 
                VALUES (:batch_code, :analysis_type_code, :file_addr, :auto_analysis, :create_time)
                ON DUPLICATE KEY UPDATE 
                analysis_type_code=VALUES(analysis_type_code),
                file_addr=VALUES(file_addr),
                auto_analysis=VALUES(auto_analysis);
            """)

            params = {
                "batch_code": row[Field_Return_BatchCode],
                "analysis_type_code": row[Field_Return_TypeAnalyst],
                "file_addr": row[Const_FileURL],
                "auto_analysis": self.autoOrManual,
                "create_time": timestamp
            }

            session.execute(sql, params)

    def process(self, powerFarmID: str, dataBatchNum: str, returnDataFrame: pd.DataFrame, timestamp: datetime = datetime.now(timezone.utc)+timedelta(hours=8)):
        try:
            uploadFiles = []
            if not returnDataFrame.empty:
                returnDataFrame[Const_FileURL] = None

            if Field_Return_IsSaveDatabase in returnDataFrame.columns:
                returnDataFrame[Field_Return_IsSaveDatabase].fillna(
                    True, inplace=True)
            else:
                returnDataFrame[Field_Return_IsSaveDatabase] = True

            for index, row in returnDataFrame.iterrows():
                directory, fileName = os.path.split(row[Field_Return_FilePath])
                basePath = f"output/{powerFarmID}"
                subPath = os.path.relpath(directory, basePath)
                remoteFilePath = os.path.join(
                    subPath, fileName).replace("\\", "/")
                # arr=["http://",self.minioUtil.client_pool.get_ip_address(),"/",powerFarmID.lower(),"/",remoteFilePath]
                arr = [powerFarmID.lower(), "/", remoteFilePath]
                fileURL = "".join(arr)
                returnDataFrame.at[index, Const_FileURL] = quote(fileURL)
                uploadFiles.append(
                    (remoteFilePath, row[Field_Return_FilePath]))

            self.uploadOfMioIO(self.powerFarmID, uploadFiles)

            foundationDB = GetBusinessFoundationDbUtil()

            with foundationDB.session_scope() as session:
                self.analysisState(session, self.dataBatchNum, AnalysisState.Analyzed.value,
                                   ErrorState.NotErr.value, None, None, timestamp, 100)
                self.analysisResultForTotal(
                    session, returnDataFrame, timestamp)
                self.analysisResultForTurbine(
                    session, returnDataFrame, timestamp)

            self.removeLocalFiles(powerFarmID, dataBatchNum)
        except Exception as e:
            self.logger.error(e)
            raise e