DataMiddleware/SourceCode/DataMiddleware/app/data_process/process.py

import utils,json

from stores.project_store import ProjectStore
from models.project_data import SubProjectModel, SubProjectItemModel
from ai.fast_gpt import FastGPTAi


class Process:
    def __init__(self):
        self._logger= utils.get_logger()
        self._project_store= ProjectStore()
        self._ai = FastGPTAi()
        self._data={}

    def run(self, sub_project: SubProjectModel) ->bool:
        try:
            self._logger.debug(f"开始处理数据:{sub_project.sub_project_name}[{sub_project.id}]")
            self._project_store.update_sub_project_status(sub_project.id, 22)
            project_items = self._project_store.query_sub_project_items_by_project(sub_project.id)
            for project_item in project_items:
                self.process_item(project_item)
            self._project_store.update_sub_project_status(sub_project.id, 32)
            self._logger.info(f"处理数据完成:{sub_project.sub_project_name}[{sub_project.id}]")
            return True
        except Exception as e:
            self._logger.error(f"处理数据失败:{e}")
            self._project_store.update_sub_project_status(sub_project.id, 12)
            return False

    def process_item(self, sub_project_item: SubProjectItemModel) ->bool:
        try:
            self._logger.debug(f"开始处理明细:{sub_project_item.device_name}{sub_project_item.device_model}[{sub_project_item.id}]")
            self._project_store.update_sub_project_item_process_status(sub_project_item.id, 1)
            data = self.call_ai(sub_project_item)
            self._project_store.update_sub_project_item_standard_no(sub_project_item.id, data.standard_no)
            return True
        except Exception as e:
            self._logger.error(f"处理数据明细:{e}")
            self._project_store.update_sub_project_item_process_status(sub_project_item.id, 3)
            return False


    def call_ai(self, item: SubProjectItemModel) -> SubProjectItemModel:
        try:
            self._logger.info(f"开始处理数据:{item.id}")
            msg = """
            请分析提供的json数据，要求：
            1. 根据工作内容、类型、项目名称、规格型号、数量、单位查找计算出定额编号
            3. 提供的数据结构化信息:```typescript
            export interface item { 
            i: // ID
            n: string; //物料名称
            m: string; //型号规格
            u：string; //单位
            c: float; //数量
            }
            ```
            4. 需返回的结构体:```typescript
            export interface item { 
            i: int; // ID 与提供的ID保持一致
            s: string; // 定额编号
            }
            ```
            5. 返回结构体item的json字符串，压缩成一行。
            6. 数据如下：
                """
            msg += json.dumps(item.to_ai_json())
            api_key = utils.get_config_value("fastgpt.api_key_process")
            self._logger.info(f"开始调用AI:\n {msg}")
            json_data = self._ai.call_ai(msg, api_key)
            self._logger.info(f"AI返回结果:{json_data}")
            if not json_data:
                raise Exception("AI返回结果为空")
            # 检查 json_data 是否为数组，并取第一项
            if isinstance(json_data, list):
                json_data = json_data[0]
            data = SubProjectItemModel(
                item_id=json_data["i"],
                standard_no=json_data["s"]
            )
            return data
        except Exception as e:
            self._logger.error(f"AI调用失败:{e}")
            raise Exception(f"AI调用失败:{e}")

    @staticmethod
    def prompt_template(project, items) ->str:
        text = f"""
        请分析提供的json数据，要求：
        1. 根据工作内容、类型、项目名称、规格型号、数量、单位查找计算出定额编号
        2. 工程的工作内容及类型：{project.work_content}"""
        text += """
        3. 提供的数据结构化信息:```typescript
        export interface item { 
        i: // ID
        n: string; //物料名称
        m: string; //型号规格
        u：string; //单位
        c: float; //数量
        }
        ```
        4. 需返回的结构体:```typescript
        export interface item { 
        i: int; // ID 与提供的ID保持一致
        s: string; // 定额编号
        }
        ```
        5. 返回结构体item的数组的json数组格式，压缩成一行。
        6. 数据如下：
        """
        item_ai_json = SubProjectItemModel.list_to_ai_json(items)
        text += json.dumps(item_ai_json,ensure_ascii=False, separators=(',', ':'))
        return text

    def call_ai_batch(self, message:str) ->list[SubProjectItemModel]:
        try:
            api_key = utils.get_config_value("fastgpt.api_key_process")
            self._logger.info(f"开始调用AI:\n {message}")
            json_data = self._ai.call_ai(message,api_key)
            self._logger.info(f"AI返回结果:{json_data}")
            if not json_data:
                raise  Exception("AI返回结果为空")
            data=[]
            for item in json_data:
                data.append(SubProjectItemModel(
                    item_id=item["i"],
                    standard_no=item["s"]
                ))
            return data
        except Exception as e:
            self._logger.error(f"AI调用失败:{e}")
            raise Exception(f"AI调用失败:{e}")