Fix/sync with backend

backend/chatsky_ui/services/json_converter_new2/base_converter.py

@@ -0,0 +1,10 @@
+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,71 @@
+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 = 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,15 @@
+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.http is not None:
+            return {"chatsky.messengers.HTTPMessengerInterface": {}}
+        elif self.interface.telegram is not None:
+            return {
+                "chatsky.messengers.TelegramInterface": {"token": {"external:os.getenv": "TG_BOT_TOKEN"}}
+            }

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

@@ -0,0 +1,57 @@
+from abc import ABC, abstractmethod
+
+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 BadConditionException(Exception):
+    pass
+
+
+class ConditionConverter(BaseConverter, ABC):
+    @abstractmethod
+    def get_pre_transitions():
+        raise NotImplementedError
+
+
+class CustomConditionConverter(ConditionConverter):
+    def __init__(self, condition: dict):
+        self.condition = None
+        try:
+            self.condition = CustomCondition(
+                name=condition["name"],
+                code=condition["data"]["python"]["action"],
+            )
+        except KeyError as missing_key:
+            raise BadConditionException("Missing key in custom condition data") from missing_key
+
+    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 = None
+        try:
+            self.condition = SlotCondition(slot_id=condition["data"]["slot"], name=condition["name"])
+        except KeyError as missing_key:
+            raise BadConditionException("Missing key in slot condition data") from missing_key
+
+    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]  # type: ignore
+        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,42 @@
+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 BadResponseException(Exception):
+    pass
+
+
+class ResponseConverter(BaseConverter):
+    pass
+
+
+class TextResponseConverter(ResponseConverter):
+    def __init__(self, response: dict):
+        try:
+            self.response = TextResponse(
+                name=response["name"],
+                text=next(iter(response["data"]))["text"],
+            )
+        except KeyError as e:
+            raise BadResponseException("Missing key in custom condition data") from e
+
+    def _convert(self):
+        return {"chatsky.Message": {"text": self.response.text}}
+
+
+class CustomResponseConverter(ResponseConverter):
+    def __init__(self, response: dict):
+        try:
+            self.response = CustomResponse(
+                name=response["name"],
+                code=next(iter(response["data"]))["python"]["action"],
+            )
+        except KeyError as e:
+            raise BadResponseException("Missing key in custom response data") from e
+
+    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/json_converter_new2/logic_component_converter/service_replacer.py

@@ -0,0 +1,82 @@
+import ast
+from ast import NodeTransformer
+from typing import Dict, List
+from pathlib import Path
+import logging
+from chatsky_ui.core.logger_config import get_logger
+
+
+class ServiceReplacer(NodeTransformer):
+    def __init__(self, new_services: List[str]):
+        self.new_services_classes = self._get_classes_def(new_services)
+        self._logger = None
+
+    @property
+    def logger(self):
+        if self._logger is None:
+            raise ValueError("Logger has not been configured. Call set_logger() first.")
+        return self._logger
+
+    def set_logger(self):
+        self._logger = get_logger(__name__)
+
+    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]
+        for idx, parsed_code in enumerate(parsed_codes):
+            classes = self._extract_class_defs(parsed_code, services_code[idx])
+        return classes
+
+    def _extract_class_defs(self, parsed_code: ast.Module, service_code: str):
+        classes = {}
+        for node in parsed_code.body:
+            if isinstance(node, ast.ClassDef):
+                classes[node.name] = node
+            else:
+                self.logger.error("No class definition found in new_service: %s", service_code)
+        return classes
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> ast.ClassDef:
+        self.logger.debug("Visiting class '%s' and comparing with: %s", node.name, self.new_services_classes.keys())
+        if node.name in self.new_services_classes:
+            return self._get_class_def(node)
+        return node
+
+    def _get_class_def(self, node: ast.ClassDef) -> ast.ClassDef:
+        service = self.new_services_classes[node.name]
+        del self.new_services_classes[node.name]
+        self.logger.info("Updating class '%s'", node.name)
+        return service
+
+    def generic_visit(self, node: ast.AST):
+        super().generic_visit(node)
+        if isinstance(node, ast.Module) and self.new_services_classes:
+            self._append_new_services(node)
+        return node
+
+    def _append_new_services(self, node: ast.Module):
+        self.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.set_logger()
+    replacer.visit(conditions_tree)
+
+    with open(services_path, "w") as file:
+        file.write(ast.unparse(conditions_tree))
+
+
+def get_all_classes(services_path):
+    with open(services_path, "r", encoding="UTF-8") as file:
+        conditions_tree = ast.parse(file.read())
+
+    return [
+        {"name": node.name, "body": ast.unparse(node)}
+        for node in conditions_tree.body
+        if isinstance(node, ast.ClassDef)
+    ]

backend/chatsky_ui/services/json_converter_new2/node_converter.py

@@ -0,0 +1,105 @@
+from .base_converter import BaseConverter
+from ...schemas.front_graph_components.node import InfoNode, LinkNode
+from .logic_component_converter.response_converter import TextResponseConverter, CustomResponseConverter
+from .logic_component_converter.condition_converter import CustomConditionConverter, SlotConditionConverter
+
+from chatsky import RESPONSE, TRANSITIONS, PRE_TRANSITION, PRE_RESPONSE
+
+
+class NodeConverter(BaseConverter):
+    RESPONSE_CONVERTER = {
+        "text": TextResponseConverter,
+        "python": CustomResponseConverter,
+    }
+    CONDITION_CONVERTER = {
+        "python": CustomConditionConverter,
+        "slot": SlotConditionConverter,
+    }
+
+    def __init__(self, config: dict):
+        pass
+
+
+class InfoNodeConverter(NodeConverter):
+    MAP_TR2CHATSKY = {
+        "start": "dst.Start",
+        "fallback": "dst.Fallback",
+        "previous": "dst.Previous",
+        "repeat": "dst.Current",
+    }
+
+    def __init__(self, node: dict):
+        self.node = InfoNode(
+            id=node["id"],
+            name=node["data"]["name"],
+            response=node["data"]["response"],
+            conditions=node["data"]["conditions"],
+        )
+
+    def __call__(self, *args, **kwargs):
+        self.slots_conf = kwargs["slots_conf"]
+        return super().__call__(*args, **kwargs)
+
+    def _convert(self):
+        condition_converters = [
+            self.CONDITION_CONVERTER[condition["type"]](condition) for condition in self.node.conditions
+        ]
+        return {
+            RESPONSE: self.RESPONSE_CONVERTER[self.node.response["type"]](self.node.response)(),
+            TRANSITIONS: [
+                {
+                    "dst": condition["dst"]
+                    if "dst" in condition and condition["data"]["transition_type"] == "manual"
+                    else self.MAP_TR2CHATSKY[condition["data"]["transition_type"]],
+                    "priority": condition["data"]["priority"],
+                    "cnd": converter(slots_conf=self.slots_conf),
+                }
+                for condition, converter in zip(self.node.conditions, condition_converters)
+            ],
+            PRE_TRANSITION: {
+                key: value
+                for converter in condition_converters
+                for key, value in converter.get_pre_transitions().items()
+            },
+            PRE_RESPONSE: {"fill": {"chatsky.processing.FillTemplate": None}},
+        }
+
+
+class LinkNodeConverter(NodeConverter):
+    def __init__(self, config: dict):
+        self.node = LinkNode(
+            id=config["id"],
+            target_flow_name=config["data"]["transition"]["target_flow"],
+            target_node_id=config["data"]["transition"]["target_node"],
+        )
+
+    def __call__(self, *args, **kwargs):
+        self.mapped_flows = kwargs["mapped_flows"]
+        return super().__call__(*args, **kwargs)
+
+    def _convert(self):
+        return [
+            self.node.target_flow_name,
+            self.mapped_flows[self.node.target_flow_name][self.node.target_node_id]["data"]["name"],
+        ]
+
+
+# class ConfNodeConverter(NodeConverter):
+#     def __init__(self, config: dict):
+#         super().__init__(config)
+
+
+#     def _convert(self):
+#         return {
+#             # node.name: node._convert() for node in self.nodes
+#         }
+
+
+# class SlotsNodeConverter(ConfNodeConverter):
+#     def __init__(self, config: List[dict]):
+#         self.slots = config
+
+#     def _convert(self):
+#         return {
+#             # node.name: node._convert() for node in self.nodes
+#         }