refactor: Reorganize json_converter into classes

backend/chatsky_ui/cli.py

@@ -9,6 +9,7 @@
 import typer
 from cookiecutter.main import cookiecutter
 from typing_extensions import Annotated
+import yaml
 
 # Patch nest_asyncio before importing Chatsky
 nest_asyncio.apply = lambda: None
@@ -93,9 +94,12 @@ def build_scenario(
         raise NotADirectoryError(f"Directory {project_dir} doesn't exist")
     settings.set_config(work_directory=project_dir)
 
-    from chatsky_ui.services.json_converter import converter  # pylint: disable=C0415
+    from chatsky_ui.services.json_converter_new2.pipeline_converter import PipelineConverter  # pylint: disable=C0415
 
-    asyncio.run(converter(build_id=build_id))
+    pipeline_converter = PipelineConverter(pipeline_id=build_id)
+    pipeline_converter(
+        input_file=settings.frontend_flows_path, output_dir=settings.scripts_dir
+    ) #TODO: rename to frontend_graph_path
 
 
 @cli.command("run_bot")

backend/chatsky_ui/schemas/front_graph_components/base_component.py

@@ -0,0 +1,4 @@
+from pydantic import BaseModel
+
+class BaseComponent(BaseModel):
+    pass

backend/chatsky_ui/schemas/front_graph_components/flow.py

@@ -0,0 +1,9 @@
+from typing import List
+
+from .base_component import BaseComponent
+
+
+class Flow(BaseComponent):
+    name: str
+    nodes: List[dict]
+    edges: List[dict]

backend/chatsky_ui/schemas/front_graph_components/info_holders/condition.py

@@ -0,0 +1,13 @@
+from ..base_component import BaseComponent
+
+
+class Condition(BaseComponent):
+    name: str
+
+
+class CustomCondition(Condition):
+    code: str
+
+
+class SlotCondition(Condition):
+    slot_id: str # not the condition id

backend/chatsky_ui/schemas/front_graph_components/info_holders/response.py

@@ -0,0 +1,13 @@
+from ..base_component import BaseComponent
+
+
+class Response(BaseComponent):
+    name: str
+
+
+class TextResponse(Response):
+    text: str
+
+
+class CustomResponse(Response):
+    code: str

backend/chatsky_ui/schemas/front_graph_components/interface.py

@@ -0,0 +1,21 @@
+from pydantic import model_validator, RootModel
+from typing import Any
+
+from .base_component import BaseComponent
+
+
+class Interface(BaseComponent, RootModel):
+    @model_validator(mode="before")
+    def validate_interface(cls, v):
+        if not isinstance(v, dict):
+            raise ValueError('interface must be a dictionary')
+        if "telegram" in v:
+            if not isinstance(v['telegram'], dict):
+                raise ValueError('telegram must be a dictionary')
+            if 'token' not in v['telegram'] or not isinstance(v['telegram']['token'], str):
+                raise ValueError('telegram dictionary must contain a string token')
+        elif "cli" in v:
+            pass
+        else:
+            raise ValueError('interface must contain either telegram or cli')
+        return v

backend/chatsky_ui/schemas/front_graph_components/node.py

@@ -0,0 +1,22 @@
+from typing import List
+
+from .base_component import BaseComponent
+
+
+class Node(BaseComponent):
+    id: str
+
+
+class InfoNode(Node):
+    name: str
+    response: dict
+    conditions: List[dict]
+
+
+class LinkNode(Node):
+    target_flow_name: str
+    target_node_id: str
+
+
+class SlotsNode(Node):
+    groups: List[dict]

backend/chatsky_ui/schemas/front_graph_components/pipeline.py

@@ -0,0 +1,8 @@
+from typing import List
+
+from .base_component import BaseComponent
+
+
+class Pipeline(BaseComponent):
+    flows: List[dict]
+    interface: dict

backend/chatsky_ui/schemas/front_graph_components/script.py

@@ -0,0 +1,7 @@
+from typing import List
+
+from .base_component import BaseComponent
+
+
+class Script(BaseComponent):
+    flows: List[dict]

backend/chatsky_ui/schemas/front_graph_components/slot.py

@@ -0,0 +1,16 @@
+from typing import Optional, List
+
+from .base_component import BaseComponent
+
+class Slot(BaseComponent):
+    name: str
+
+
+class RegexpSlot(Slot):
+    id: str
+    regexp: str
+    match_group_idx: Optional[int]
+
+
+class GroupSlot(Slot):
+    slots: List[dict]

backend/chatsky_ui/services/json_converter_new2/base_converter.py

@@ -0,0 +1,9 @@
+from abc import ABC, abstractmethod
+
+class BaseConverter(ABC):
+    def __call__(self, *args, **kwargs):
+        return self._convert()
+
+    @abstractmethod
+    def _convert(self):
+        raise NotImplementedError

backend/chatsky_ui/services/json_converter_new2/consts.py

@@ -0,0 +1,3 @@
+RESPONSES_FILE="responses"
+CONDITIONS_FILE="conditions"
+CUSTOM_FILE="custom"

backend/chatsky_ui/services/json_converter_new2/flow_converter.py

@@ -0,0 +1,68 @@
+from typing import Dict, List, Any, Tuple
+from ...schemas.front_graph_components.flow import Flow
+from .node_converter import InfoNodeConverter, LinkNodeConverter
+from .base_converter import BaseConverter
+
+
+class FlowConverter(BaseConverter):
+    NODE_CONVERTERS = {
+        "default_node": InfoNodeConverter,
+        "link_node": LinkNodeConverter,
+    }
+
+    def __init__(self, flow: Dict[str, Any]):
+        self._validate_flow(flow)
+        self.flow = Flow(
+            name=flow["name"],
+            nodes=flow["data"]["nodes"],
+            edges=flow["data"]["edges"],
+        )
+
+    def __call__(self, *args, **kwargs):
+        self.mapped_flows = kwargs["mapped_flows"]
+        self.slots_conf = kwargs["slots_conf"]
+        self._integrate_edges_into_nodes()
+        return super().__call__(*args, **kwargs)
+
+    def _validate_flow(self, flow: Dict[str, Any]):
+        if "data" not in flow or "nodes" not in flow["data"] or "edges" not in flow["data"]:
+            raise ValueError("Invalid flow structure")
+
+    def _integrate_edges_into_nodes(self):
+        def _insert_dst_into_condition(node: Dict[str, Any], condition_id: str, target_node: Tuple[str, str]) -> Dict[str, Any]:
+            for condition in node["data"]["conditions"]:
+                if condition["id"] == condition_id:
+                    condition["dst"] = target_node
+            return node
+
+        maped_edges = self._map_edges()
+        nodes = self.flow.nodes.copy()
+        for edge in maped_edges:
+            for idx, node in enumerate(nodes):
+                if node["id"] == edge["source"]:
+                    nodes[idx] = _insert_dst_into_condition(node, edge["sourceHandle"], edge["target"])
+        self.flow.nodes = nodes
+
+    def _map_edges(self) -> List[Dict[str, Any]]:
+        def _get_flow_and_node_names(target_node):
+            node_type = target_node["type"]
+            if node_type == "link_node": #TODO: WHY CONVERTING HERE?
+                return LinkNodeConverter(target_node)(mapped_flows=self.mapped_flows)
+            elif node_type == "default_node":
+                return [self.flow.name, target_node["data"]["name"]]
+
+        edges = self.flow.edges.copy()
+        for edge in edges:
+            target_id = edge["target"]
+            # target_node = _find_node_by_id(target_id, self.flow.nodes)
+            target_node = self.mapped_flows[self.flow.name].get(target_id)
+            if target_node:
+                edge["target"] = _get_flow_and_node_names(target_node)
+        return edges
+
+    def _convert(self) -> Dict[str, Any]:
+        converted_flow = {self.flow.name: {}}
+        for node in self.flow.nodes:
+            if node["type"] == "default_node":
+                converted_flow[self.flow.name].update({node["data"]["name"]: InfoNodeConverter(node)(slots_conf=self.slots_conf)})
+        return converted_flow

backend/chatsky_ui/services/json_converter_new2/interface_converter.py

@@ -0,0 +1,9 @@
+from .base_converter import BaseConverter
+from ...schemas.front_graph_components.interface import Interface
+
+class InterfaceConverter(BaseConverter):
+    def __init__(self, interface: dict):
+        self.interface = Interface(**interface)
+
+    def _convert(self):
+        return self.interface.model_dump()

backend/chatsky_ui/services/json_converter_new2/logic_component_converter/condition_converter.py

@@ -0,0 +1,59 @@
+from abc import ABC, abstractmethod
+import ast
+
+from ..consts import CUSTOM_FILE, CONDITIONS_FILE
+from ..base_converter import BaseConverter
+from ....schemas.front_graph_components.info_holders.condition import CustomCondition, SlotCondition
+
+
+class ConditionConverter(BaseConverter, ABC):
+    @abstractmethod
+    def get_pre_transitions():
+        raise NotImplementedError
+
+
+class CustomConditionConverter(ConditionConverter):
+    def __init__(self, condition: dict):
+        self.condition = CustomCondition(
+            name=condition["name"],
+            code=condition["data"]["python"]["action"],
+        )
+
+    def _parse_code(self):
+        condition_code = next(iter(ast.parse(self.condition.code).body))
+
+        if not isinstance(condition_code, ast.ClassDef):
+            raise ValueError("Condition python code is not a ClassDef")
+        return condition_code
+
+    def _convert(self):
+        custom_cnd = {
+            f"{CUSTOM_FILE}.{CONDITIONS_FILE}.{self.condition.name}": None
+        }
+        return custom_cnd
+
+    def get_pre_transitions(self):
+        return {}
+
+
+class SlotConditionConverter(ConditionConverter):
+    def __init__(self, condition: dict):
+        self.condition = SlotCondition(
+            slot_id=condition["data"]["slot"],
+            name=condition["name"]
+        )
+
+    def __call__(self, *args, **kwargs):
+        self.slots_conf = kwargs["slots_conf"]
+        return super().__call__(*args, **kwargs)
+
+    def _convert(self):
+        return {"chatsky.conditions.slots.SlotsExtracted": self.slots_conf[self.condition.slot_id]}
+
+    def get_pre_transitions(self):
+        slot_path = self.slots_conf[self.condition.slot_id]
+        return {
+            slot_path: {
+                "chatsky.processing.slots.Extract": slot_path
+            }
+        }

backend/chatsky_ui/services/json_converter_new2/logic_component_converter/response_converter.py

@@ -0,0 +1,46 @@
+import ast
+
+from ..base_converter import BaseConverter
+from ....schemas.front_graph_components.info_holders.response import TextResponse, CustomResponse
+from ..consts import CUSTOM_FILE, RESPONSES_FILE
+
+
+class ResponseConverter(BaseConverter):
+    pass
+
+
+class TextResponseConverter(ResponseConverter):
+    def __init__(self, response: dict):
+        self.response = TextResponse(
+            name=response["name"],
+            text=next(iter(response["data"]))["text"],
+        )
+
+    def _convert(self):
+        return {
+            "chatsky.Message": {
+                "text": self.response.text
+            }
+        }
+
+
+class CustomResponseConverter(ResponseConverter):
+    def __init__(self, response: dict):
+        # self.code = 
+        self.response = CustomResponse(
+            name=response["name"],
+            code=next(iter(response["data"]))["python"]["action"],
+        )
+
+    def _parse_code(self):
+        response_code = next(iter(ast.parse(self.response.code).body))
+
+        if not isinstance(response_code, ast.ClassDef):
+            raise ValueError("Response python code is not a ClassDef")
+        return response_code
+
+    def _convert(self):
+        return {
+            f"{CUSTOM_FILE}.{RESPONSES_FILE}.{self.response.name}": None
+        }
+