convert the workflow in natural language to json format
(self, workflow_txt)
| 75 | return "OpenAI API error." |
| 76 | |
| 77 | def _txt2json(self, workflow_txt): |
| 78 | ''' convert the workflow in natural language to json format ''' |
| 79 | workflow = [] |
| 80 | try: |
| 81 | steps = workflow_txt.split('\n') |
| 82 | for step in steps: |
| 83 | if step[0:4] != "STEP": |
| 84 | continue |
| 85 | left_indices = [_.start() for _ in re.finditer("\[", step)] |
| 86 | right_indices = [_.start() for _ in re.finditer("\]", step)] |
| 87 | step_id = step[: left_indices[0]-2] |
| 88 | step_name = step[left_indices[0]+1: right_indices[0]] |
| 89 | step_description = step[left_indices[1]+1: right_indices[1]] |
| 90 | jump_str = step[left_indices[2]+1: right_indices[-1]] |
| 91 | if re.findall(re.compile(r'[A-Za-z]',re.S), jump_str) == []: |
| 92 | workflow.append({"stepId": step_id, "stepName": step_name, "stepDescription": step_description, "jumpLogic": [], "extension": []}) |
| 93 | continue |
| 94 | jump_logic = [] |
| 95 | left_indices = [_.start() for _ in re.finditer('\[', jump_str)] |
| 96 | right_indices = [_.start() for _ in re.finditer('\]', jump_str)] |
| 97 | i = 1 |
| 98 | while i < len(left_indices): |
| 99 | jump = {"Condition": jump_str[left_indices[i]+1: right_indices[i-1]], "Target": re.search(r'STEP\s\d', jump_str[left_indices[i+1]+1: right_indices[i]]).group(0)} |
| 100 | jump_logic.append(jump) |
| 101 | i += 3 |
| 102 | workflow.append({"stepId": step_id, "stepName": step_name, "stepDescription": step_description, "jumpLogic": jump_logic, "extension": []}) |
| 103 | return json.dumps(workflow) |
| 104 | except: |
| 105 | print("Format error, please try again.") |
no outgoing calls
no test coverage detected