refactor json converter2 (#95)

* refactor: Reorganize json_converter into classes

* refactor: Add store_custom_services

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,22 @@
+from pydantic import Field, model_validator
+from typing import Any
+
+from .base_component import BaseComponent
+from typing import Optional, Dict
+
+class Interface(BaseComponent):
+    telegram: Optional[Dict[str, Any]] = Field(default=None)
+    cli: Optional[Dict[str, Any]] = Field(default=None)
+
+    @model_validator(mode='after')
+    def check_one_not_none(cls, values):
+        telegram, cli = values.telegram, values.cli
+        if (telegram is None) == (cli is None):
+            raise ValueError('Exactly one of "telegram" or "cli" must be provided.')
+        return values
+
+    @model_validator(mode='after')
+    def check_telegram_token(cls, values):
+        if values.telegram is not None and 'token' not in values.telegram:
+            raise ValueError('Telegram token must be provided.')
+        return values

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: 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,14 @@
+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):
+        if self.interface.cli is not None:
+            return {"chatsky.messengers.console.CLIMessengerInterface": {}}
+        elif self.interface.telegram is not None:
+            return {
+                "chatsky.messengers.telegram.LongpollingInterface": {"token": self.interface.telegram["token"]}
+            }

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

@@ -0,0 +1,55 @@
+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
+from ....core.config import settings
+from .service_replacer import store_custom_service
+
+
+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 _convert(self):
+        store_custom_service(settings.conditions_path, [self.condition.code])
+        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,40 @@
+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
+from ....core.config import settings
+from .service_replacer import store_custom_service
+
+
+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.response = CustomResponse(
+            name=response["name"],
+            code=next(iter(response["data"]))["python"]["action"],
+        )
+
+    def _convert(self):
+        store_custom_service(settings.responses_path, [self.response.code])
+        return {
+            f"{CUSTOM_FILE}.{RESPONSES_FILE}.{self.response.name}": None
+        }

backend/chatsky_ui/services/condition_finder.py → backend/chatsky_ui/services/json_converter_new2/logic_component_converter/service_replacer.py

@@ -1,6 +1,7 @@
 import ast
 from ast import NodeTransformer
 from typing import Dict, List
+from pathlib import Path
 
 from chatsky_ui.core.logger_config import get_logger
 
@@ -13,17 +14,18 @@ def __init__(self, new_services: List[str]):
 
     def _get_classes_def(self, services_code: List[str]) -> Dict[str, ast.ClassDef]:
         parsed_codes = [ast.parse(service_code) for service_code in services_code]
-        result_nodes = {}
         for idx, parsed_code in enumerate(parsed_codes):
-            self._extract_class_defs(parsed_code, result_nodes, services_code[idx])
-        return result_nodes
+            classes = self._extract_class_defs(parsed_code, services_code[idx])
+        return classes
 
-    def _extract_class_defs(self, parsed_code: ast.Module, result_nodes: Dict[str, ast.ClassDef], service_code: str):
+    def _extract_class_defs(self, parsed_code: ast.Module, service_code: str):
+        classes = {}
         for node in parsed_code.body:
             if isinstance(node, ast.ClassDef):
-                result_nodes[node.name] = node
+                classes[node.name] = node
             else:
                 logger.error("No class definition found in new_service: %s", service_code)
+        return classes
 
     def visit_ClassDef(self, node: ast.ClassDef) -> ast.ClassDef:
         logger.debug("Visiting class '%s' and comparing with: %s", node.name, self.new_services_classes.keys())
@@ -46,3 +48,14 @@ def _append_new_services(self, node: ast.Module):
         logger.info("Services not found, appending new services: %s", list(self.new_services_classes.keys()))
         for _, service in self.new_services_classes.items():
             node.body.append(service)
+
+
+def store_custom_service(services_path: Path, services: List[str]):
+    with open(services_path, "r", encoding="UTF-8") as file:
+        conditions_tree = ast.parse(file.read())
+
+    replacer = ServiceReplacer(services)
+    replacer.visit(conditions_tree)
+
+    with open(services_path, "w") as file:
+        file.write(ast.unparse(conditions_tree))