refactor

src/reviewsense/api/routes.py

 from fastapi import APIRouter
 from typing import Dict, Optional
 from reviewsense.api.schemas.product_review_input import ProductReviewInput
-from reviewsense.service.ReviewService import ReviewService
+from reviewsense.services.ReviewService import ReviewService
 
 
 router = APIRouter()

src/reviewsense/service/ReviewService.py

-import os
-import numpy as np
-from typing import List, Dict, Optional
-from langchain_community.docstore.document import Document
-from langchain_astradb import AstraDBVectorStore
-from langchain_huggingface import HuggingFaceEmbeddings
-from transformers import pipeline
-
-from reviewsense.core.config import get_settings
-__all__ = ['ReviewService'] 
-class ReviewService:
-    """Service class for handling product reviews"""
-    def __init__(
-        self, 
-        collection_name: str = "android_phone_reviews", 
-        embedding_model: str = "sentence-transformers/all-mpnet-base-v2"
-    ):
-        """
-        Initialize the review service with vector database and embeddings
-        
-        Args:
-            collection_name (str): Name of the vector database collection
-            embedding_model (str): Hugging Face embedding model to use
-        """
-        settings = get_settings()
-        
-        # Initialize embeddings
-        self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
-        
-        # Initialize the vector store
-        self.vector_store = AstraDBVectorStore(
-            collection_name=collection_name,
-            embedding=self.embeddings,
-            api_endpoint=settings.ASTRA_DB_API_ENDPOINT,
-            token=settings.ASTRA_DB_APPLICATION_TOKEN,
-            namespace=settings.ASTRA_DB_NAMESPACE,
-        )
-        
-        # Initialize sentiment analyzer
-        self.sentiment_analyzer = pipeline("sentiment-analysis")
-
-    # [Rest of the methods remain the same as in the original implementation]
-    def fetch_reviews(self, product_id: str, features: List[str], threshold: float = 0.6) -> Dict[str, List[str]]:
-        """
-        Fetch reviews for specific product features
-        
-        Args:
-            product_id (str): Identifier for the product
-            features (List[str]): Features to extract reviews for
-            threshold (float): Similarity threshold for review filtering
-        
-        Returns:
-            Dict containing reviews for each feature
-        """
-        feature_reviews = {}
-
-        for feature in features:
-            # Perform similarity search for the current feature
-            filter_criteria = {"title": product_id}
-            documents = self.vector_store.similarity_search_with_score_id(
-                query=feature, 
-                k=100, 
-                filter=filter_criteria
-            )
-
-            # Filter and collect reviews
-            filtered_reviews = [
-                doc.page_content for doc, score, _ in documents if score > threshold
-            ]
-
-            # Add the list of reviews to the result dictionary
-            feature_reviews[feature] = filtered_reviews
-
-        return feature_reviews
-
-    def generate_feature_ratings(self, reviews_by_feature: Dict[str, List[str]]) -> Dict[str, Optional[float]]:
-        """
-        Generate ratings for each feature based on sentiment analysis
-        
-        Args:
-            reviews_by_feature (Dict): Dictionary of reviews grouped by feature
-        
-        Returns:
-            Dictionary of feature ratings
-        """
-        feature_ratings = {}
-        
-        for feature, reviews in reviews_by_feature.items():
-            if not reviews:
-                feature_ratings[feature] = None
-                continue
-            
-            # Analyze sentiment for each review
-            ratings = []
-            for review in reviews:
-                sentiment = self.sentiment_analyzer(review)[0]
-                
-                # Map sentiment to numerical score
-                if "positive" in sentiment["label"].lower():
-                    ratings.append(5)
-                elif "negative" in sentiment["label"].lower():
-                    ratings.append(1)
-                else:  # Neutral sentiment
-                    ratings.append(3)
-            
-            # Compute average rating
-            feature_ratings[feature] = round(np.mean(ratings), 1) if ratings else None
-        
-        return feature_ratings
-
-    def add_review(self, product_id: str, review: str) -> str:
-        """
-        Add a new review to the vector database
-        
-        Args:
-            product_id (str): Identifier for the product
-            review (str): Review text to add
-        
-        Returns:
-            ID of the added document
-        """
-        # Create a new document with the review
-        review_document = Document(
-            page_content=review, 
-            metadata={"title": product_id}
-        )
-        
-        # Add document to vector store
-        return self.vector_store.add_documents([review_document])[0]
-

src/reviewsense/services/ReviewService.py

+# services/review_service.py
+
+from typing import List, Dict, Optional
+from reviewsense.services.review_fetcher import ReviewFetcher
+from reviewsense.services.review_rater import ReviewRater
+from reviewsense.services.review_adder import ReviewAdder
+
+class ReviewService:
+    """Service class for handling product reviews"""
+
+    def __init__(self, collection_name: str = "android_phone_reviews", embedding_model: str = "sentence-transformers/all-mpnet-base-v2"):
+        """
+        Initialize the review service with vector database and embeddings
+        """
+        self.fetcher = ReviewFetcher(collection_name, embedding_model)
+        self.rater = ReviewRater()
+        self.adder = ReviewAdder(collection_name, embedding_model)
+
+    def fetch_reviews(self, product_id: str, features: List[str], threshold: float = 0.6) -> Dict[str, List[str]]:
+        return self.fetcher.fetch_reviews(product_id, features, threshold)
+
+    def generate_feature_ratings(self, reviews_by_feature: Dict[str, List[str]]) -> Dict[str, Optional[float]]:
+        return self.rater.generate_feature_ratings(reviews_by_feature)
+
+    def add_review(self, product_id: str, review: str) -> str:
+        return self.adder.add_review(product_id, review)

src/reviewsense/services/review_adder.py

+# services/review_adder.py
+
+from langchain_community.docstore.document import Document
+from reviewsense.core.config import get_settings
+from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_astradb import AstraDBVectorStore
+
+class ReviewAdder:
+    """Class for adding reviews to the vector store"""
+
+    def __init__(self, collection_name: str, embedding_model: str):
+        settings = get_settings()
+        self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
+        self.vector_store = AstraDBVectorStore(
+            collection_name=collection_name,
+            embedding=self.embeddings,
+            api_endpoint=settings.ASTRA_DB_API_ENDPOINT,
+            token=settings.ASTRA_DB_APPLICATION_TOKEN,
+            namespace=settings.ASTRA_DB_NAMESPACE,
+        )
+
+    def add_review(self, product_id: str, review: str) -> str:
+        review_document = Document(page_content=review, metadata={"title": product_id})
+        return self.vector_store.add_documents([review_document])[0]

src/reviewsense/services/review_fetcher.py

+# services/review_fetcher.py
+
+from typing import List, Dict
+from reviewsense.core.config import get_settings
+from langchain_astradb import AstraDBVectorStore
+from langchain_huggingface import HuggingFaceEmbeddings
+
+class ReviewFetcher:
+    """Class for fetching reviews from the vector store"""
+
+    def __init__(self, collection_name: str, embedding_model: str):
+        settings = get_settings()
+        self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model)
+        self.vector_store = AstraDBVectorStore(
+            collection_name=collection_name,
+            embedding=self.embeddings,
+            api_endpoint=settings.ASTRA_DB_API_ENDPOINT,
+            token=settings.ASTRA_DB_APPLICATION_TOKEN,
+            namespace=settings.ASTRA_DB_NAMESPACE,
+        )
+
+    def fetch_reviews(self, product_id: str, features: List[str], threshold: float = 0.6) -> Dict[str, List[str]]:
+        feature_reviews = {}
+
+        for feature in features:
+            filter_criteria = {"title": product_id}
+            documents = self.vector_store.similarity_search_with_score_id(query=feature, k=100, filter=filter_criteria)
+
+            filtered_reviews = [doc.page_content for doc, score, _ in documents if score > threshold]
+            feature_reviews[feature] = filtered_reviews
+
+        return feature_reviews

src/reviewsense/services/review_rater.py

+# services/review_rater.py
+
+from typing import Dict, List, Optional
+import numpy as np
+from langchain_huggingface import HuggingFaceEmbeddings
+from transformers import pipeline
+
+class ReviewRater:
+    """Class for generating feature-specific ratings based on sentiment analysis"""
+
+    def __init__(self):
+        self.sentiment_analyzer = pipeline("sentiment-analysis")
+
+    def generate_feature_ratings(self, reviews_by_feature: Dict[str, List[str]]) -> Dict[str, Optional[float]]:
+        feature_ratings = {}
+
+        for feature, reviews in reviews_by_feature.items():
+            if not reviews:
+                feature_ratings[feature] = None
+                continue
+            
+            ratings = []
+            for review in reviews:
+                sentiment = self.sentiment_analyzer(review)[0]
+                
+                if "positive" in sentiment["label"].lower():
+                    ratings.append(5)
+                elif "negative" in sentiment["label"].lower():
+                    ratings.append(1)
+                else:
+                    ratings.append(3)
+
+            feature_ratings[feature] = round(np.mean(ratings), 1) if ratings else None
+
+        return feature_ratings