Merge pull request #4306 from ozer550/Manager_Class_Appnexus

Implements Automation Manager Class

contentcuration/contentcuration/tests/utils/test_automation_manager.py

@@ -0,0 +1,40 @@
+import unittest
+from unittest.mock import MagicMock
+
+from contentcuration.utils.automation_manager import AutomationManager
+
+
+class AutomationManagerTestCase(unittest.TestCase):
+    def setUp(self):
+        self.automation_manager = AutomationManager()
+
+    def test_creation(self):
+        # Check if an instance of AutomationManager is created successfully
+        self.assertIsInstance(self.automation_manager, AutomationManager)
+
+    def test_generate_embedding(self):
+        text = "Some text that needs to be embedded"
+        # Mock the generate_embedding method of RecommendationsAdapter
+        # as the implementation is yet to be done
+        self.automation_manager.recommendations_backend_adapter.generate_embedding = MagicMock(return_value=[0.1, 0.2, 0.3])
+        embedding_vector = self.automation_manager.generate_embedding(text)
+        self.assertIsNotNone(embedding_vector)
+
+    def test_embedding_exists(self):
+        embedding_vector = [0.1, 0.2, 0.3]
+        # Currently no solid implementation exists for this
+        # So the embadding_exists function returns true anyways
+        exists = self.automation_manager.embedding_exists(embedding_vector)
+        self.assertTrue(exists)
+
+    def test_load_recommendations(self):
+        embedding_vector = [0.1, 0.2, 0.3]
+        self.automation_manager.recommendations_backend_adapter.get_recommendations = MagicMock(return_value=["item1", "item2"])
+        recommendations = self.automation_manager.load_recommendations(embedding_vector)
+        self.assertIsInstance(recommendations, list)
+
+    def test_cache_embeddings(self):
+        embeddings_list = [[0.1, 0.2, 0.3]]
+        # Currently the function returns true anyways
+        success = self.automation_manager.cache_embeddings(embeddings_list)
+        self.assertTrue(success)

contentcuration/contentcuration/utils/automation_manager.py

@@ -0,0 +1,52 @@
+from contentcuration.utils.recommendations import RecommendationsAdapter
+from contentcuration.utils.recommendations import RecommendationsBackendFactory
+
+
+class AutomationManager:
+    def __init__(self):
+        self.recommendations_backend_factory = RecommendationsBackendFactory()
+        self.recommendations_backend_instance = self.recommendations_backend_factory.create_backend()
+        self.recommendations_backend_adapter = RecommendationsAdapter(self.recommendations_backend_instance)
+
+    def generate_embedding(self, text):
+        """
+        Generate an embedding vector for the given text.
+        Args:
+            text (str): The text for which to generate an embedding.
+        Returns:
+            Vector: The generated embedding vector.
+        """
+        embedding_vector = self.recommendations_backend_adapter.generate_embedding(text=text)
+        return embedding_vector
+
+    def embedding_exists(self, embedding):
+        """
+        Check if the given embedding vector exists.
+        Args:
+            embedding (Vector): The embedding vector to check.
+        Returns:
+            bool: True if the embedding exists, False otherwise.
+        """
+        return self.recommendations_backend_adapter.embedding_exists(embedding=embedding)
+
+    def load_recommendations(self, embedding):
+        """
+        Load recommendations based on the given embedding vector.
+        Args:
+            embedding (Vector): The embedding vector to use for recommendations.
+        Returns:
+            list: A list of recommended items.
+        """
+        # Need to extract the recommendation list from the ResponseObject and change the return statement
+        self.recommendations_backend_adapter.get_recommendations(embedding=embedding)
+        return []
+
+    def cache_embeddings(self, embeddings):
+        """
+        Cache a list of embedding vectors.
+        Args:
+            embeddings (list): A list of embedding vectors to cache.
+        Returns:
+            bool: True if caching was successful, False otherwise.
+        """
+        return self.recommendations_backend_adapter.cache_embeddings(embeddings)

contentcuration/contentcuration/utils/recommendations.py

@@ -47,8 +47,27 @@ def generate_embedding(self, text) -> EmbeddingsResponse:
         request = EmbeddingsRequest()
         return self.backend.make_request(request)
 
-    def get_recommendations(self) -> RecommendationsResponse:
-        request = RecommedationsRequest()
+    def embedding_exists(self, embedding) -> bool:
+        # Need to implement the logic to check if the embeddigns exist
+        # Return True if the embedding exists, or False otherwise
+        return True
+
+    def cache_embeddings(self, embeddings_list) -> bool:
+        for embedding in embeddings_list:
+            try:
+                # Attempt to cache the embedding
+                # Write the caching logic
+                # A conrner case to look at here is if one of the embedding fails to get cached
+                # we need to handel it so that only the once that were not succesfull
+                # are attempted to cache again
+                pass
+            except Exception as e:
+                print(e)
+                return False
+        return True
+
+    def get_recommendations(self, embedding) -> RecommendationsResponse:
+        request = RecommedationsRequest(embedding)
         return self.backend.make_request(request)