init

prompt/prompt.py

@@ -51,7 +51,7 @@ def handle_general_rules(data:dict):
 4、规则里有 结果备注 的请在函数中的 输出结果备注 处进行备注，没有的则无输出结果备注
 """
 
-# 流程图
+# 未使用
 prompt_parse_flow_image = """
 # 角色
 你是一个专业的Python代码生成器，能够根据给定的流程图和函数内容，用Python语言生成完整的流程执行代码。
@@ -111,7 +111,7 @@ if __name__ == "__main__":
 1、针对流程分支情况请参考给定的函数内容及函数的输出结果备注。
 2、开始节点与结束节点，无实际意义，结束节点直接返回其上一个节点输出的结果。
 3、只返回函数代码，不要多余的输出。
-4、代码逻辑应严格按照图中的逻辑，不要自行添加多余的逻辑。
+4、代码逻辑应严格按照流程描述中的逻辑，不要自行添加多余的逻辑。
 5、代码的语法要符合python的语法规范，返回的代码应该是可执行的。
 """
 
@@ -122,7 +122,7 @@ def f_get_prompt_parse_node(rules_name: str, rules: str, default_output: str):
         .replace("{rules}", rules) \
         .replace("{default_output}", default_output)
 
-
+# 未使用
 def f_get_prompt_parse_flow_image(node_list: list):
     func = ""
     node_func_map = ""

strategy_parse.py

@@ -53,6 +53,7 @@ class StrategyParse:
                 py_files.append(os.path.join(self._base_dir, file_name))
         return py_files
 
+    # 未使用
     def _f_parse_flow_image(self, ws, node_list: list):
         image = ws._images[0]
         img = Image.open(image.ref).convert("RGB")
@@ -84,40 +85,23 @@ class StrategyParse:
         with open(save_path, mode="w", encoding="utf8") as f:
             f.write("")
 
-    def _f_parse_flow(self, node_list: list, df: pd.DataFrame):
-
-        node_func_dict = {BaseConfig.flow_sheet_name: "flow.py"}
-        func = ""
-        node_func_map = ""
-        func_import = ""
-        for node_name, func_name, code in node_list:
-            node_func_dict[node_name] = f"{func_name}.py"
-            func = f"{func}{code}\n\n"
-            node_func_map = f"{node_func_map}{node_name}: {func_name}\n"
-            func_import = f"{func_import}from {func_name} import {func_name}\n"
-
-        save_path = self._f_get_save_path(BaseConfig.node_map_name)
-        with open(save_path, mode="w", encoding="utf8") as f:
-            f.write(json.dumps(node_func_dict, ensure_ascii=False))
-
-        flow = ""
-        for _, row in df.iterrows():
-            strategy = row["策略流描述"]
-            flow = f"{flow}{strategy}\n"
-        flow = flow.strip()
-
-        prompt = f_get_prompt_parse_flow(func, node_func_map, func_import, flow)
-        print(prompt)
-        llm_answer = call_llm(prompt)
-        print(llm_answer)
-        code = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
-        save_path = self._f_get_save_path("flow.py")
-        with open(save_path, mode="w", encoding="utf8") as f:
-            f.write(code)
-
-        save_path = self._f_get_save_path("__init__.py")
-        with open(save_path, mode="w", encoding="utf8") as f:
-            f.write("")
+    # 未使用
+    def _f_parse_strategy_image(self, file_path):
+        wb = load_workbook(file_path)
+        excel = pd.ExcelFile(file_path)
+        sheet_names = excel.sheet_names
+        if BaseConfig.flow_sheet_name not in sheet_names:
+            raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"sheet【{BaseConfig.flow_sheet_name}】不存在")
+        node_list = []
+        for node_name in tqdm(sheet_names):
+            if node_name == BaseConfig.flow_sheet_name:
+                continue
+            df = excel.parse(sheet_name=node_name)
+            func_name, code = self._f_parse_node(df, node_name)
+            node_list.append((node_name, func_name, code))
+        self._f_parse_flow_image(wb[BaseConfig.flow_sheet_name], node_list)
+        wb.close()
+        excel.close()
 
     def _f_parse_node(self, df: pd.DataFrame, node_name):
         rules = ""
@@ -153,40 +137,68 @@ class StrategyParse:
             default_output = str(default_output).replace("\n", " ")
             default_output = f"{default_output}"
 
+        # 构造提示词
         prompt = f_get_prompt_parse_node(node_name, rules, default_output)
         print(prompt)
+        # 调用大模型
         llm_answer = call_llm(prompt)
+        # 解析代码部分
         code = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
+        # 解析函数名
         func_name = re.findall(r"def (.*)\(data", code)[0]
+        # 保存节点代码
         save_path = self._f_get_save_path(f"{func_name}.py")
         print(code)
         with open(save_path, mode="w", encoding="utf8") as f:
             f.write(code)
         return func_name, code
 
-    def _f_parse_strategy_image(self, file_path):
-        wb = load_workbook(file_path)
-        excel = pd.ExcelFile(file_path)
-        sheet_names = excel.sheet_names
-        if BaseConfig.flow_sheet_name not in sheet_names:
-            raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"sheet【{BaseConfig.flow_sheet_name}】不存在")
-        node_list = []
-        for node_name in tqdm(sheet_names):
-            if node_name == BaseConfig.flow_sheet_name:
-                continue
-            df = excel.parse(sheet_name=node_name)
-            func_name, code = self._f_parse_node(df, node_name)
-            node_list.append((node_name, func_name, code))
-        self._f_parse_flow_image(wb[BaseConfig.flow_sheet_name], node_list)
-        wb.close()
-        excel.close()
+    def _f_parse_flow(self, node_list: list, df: pd.DataFrame):
+
+        node_func_dict = {BaseConfig.flow_sheet_name: "flow.py"}
+        func = ""
+        node_func_map = ""
+        func_import = ""
+        for node_name, func_name, code in node_list:
+            node_func_dict[node_name] = f"{func_name}.py"
+            func = f"{func}{code}\n\n"
+            node_func_map = f"{node_func_map}{node_name}: {func_name}\n"
+            func_import = f"{func_import}from {func_name} import {func_name}\n"
+
+        save_path = self._f_get_save_path(BaseConfig.node_map_name)
+        with open(save_path, mode="w", encoding="utf8") as f:
+            f.write(json.dumps(node_func_dict, ensure_ascii=False))
+
+        flow = ""
+        for _, row in df.iterrows():
+            strategy = row["策略流描述"]
+            flow = f"{flow}{strategy}\n"
+        flow = flow.strip()
+        # 构造提示词
+        prompt = f_get_prompt_parse_flow(func, node_func_map, func_import, flow)
+        print(prompt)
+        # 调用大模型
+        llm_answer = call_llm(prompt)
+        print(llm_answer)
+        # 解析代码部分
+        code = re.findall(r"```python\n(.*)\n```", llm_answer, flags=re.DOTALL)[0]
+        # 保存代码
+        save_path = self._f_get_save_path("flow.py")
+        with open(save_path, mode="w", encoding="utf8") as f:
+            f.write(code)
+
+        save_path = self._f_get_save_path("__init__.py")
+        with open(save_path, mode="w", encoding="utf8") as f:
+            f.write("")
 
     def f_parse_strategy(self, excel: pd.ExcelFile, progress=None):
         sheet_names = excel.sheet_names
         if BaseConfig.flow_sheet_name not in sheet_names:
             raise GeneralException(ResultCodesEnum.NOT_FOUND, message=f"sheet【{BaseConfig.flow_sheet_name}】不存在")
+        # 解析各节点
         node_list = []
         for node_name in tqdm(sheet_names):
+            # 忽略“流程”的sheet
             if node_name == BaseConfig.flow_sheet_name:
                 continue
             df = excel.parse(sheet_name=node_name)
@@ -194,8 +206,9 @@ class StrategyParse:
             node_list.append((node_name, func_name, code))
         if progress is not None:
             progress(0.9)
+        # 解析流程
         self._f_parse_flow(node_list, excel.parse(sheet_name=BaseConfig.flow_sheet_name))
-
+        # 打包文件
         save_path = self._f_get_save_path(BaseConfig.code_zip_name)
         py_files = self._f_get_py_files()
         f_create_zip(save_path, py_files)