Spaces:
Sleeping
Sleeping
| """ | |
| Business logic services for NLP operations | |
| """ | |
| import logging | |
| from typing import List, Dict, Any | |
| from lib.providers.model_providers import ( | |
| SentimentModelProvider, | |
| NERModelProvider, | |
| SummarizationModelProvider, | |
| TranslationModelProvider, | |
| ParaphraseModelProvider, | |
| ) | |
| from lib.models import Entity, NERResponse, SentimentResponse, BatchSentimentResult, ParaphraseResponse, SummarizationResponse | |
| logger = logging.getLogger(__name__) | |
| class SentimentService: | |
| """Service for sentiment analysis operations""" | |
| def __init__(self, model_provider: SentimentModelProvider): | |
| self.model_provider = model_provider | |
| def analyze_sentiment(self, text: str) -> SentimentResponse: | |
| """ | |
| Analyze sentiment of a single text | |
| Args: | |
| text: The text to analyze | |
| Returns: | |
| SentimentResponse with sentiment, confidence, and scores | |
| """ | |
| results = self.model_provider.predict(text) | |
| # Extract the highest scoring sentiment | |
| if isinstance(results, list) and len(results) > 0: | |
| if isinstance(results[0], list): | |
| best_result = max(results[0], key=lambda x: x['score']) | |
| all_scores = results[0] | |
| else: | |
| best_result = results[0] | |
| all_scores = results | |
| else: | |
| best_result = results | |
| all_scores = None | |
| # Map sentiment labels to more user-friendly format | |
| sentiment_label = best_result['label'].lower() | |
| if 'positive' in sentiment_label: | |
| sentiment = "Positive" | |
| elif 'negative' in sentiment_label: | |
| sentiment = "Negative" | |
| else: | |
| sentiment = "Neutral" | |
| return SentimentResponse( | |
| sentiment=sentiment, | |
| confidence=round(best_result['score'], 3), | |
| all_scores=all_scores | |
| ) | |
| def analyze_batch(self, texts: List[str]) -> List[BatchSentimentResult]: | |
| """ | |
| Analyze sentiment for multiple texts | |
| Args: | |
| texts: List of texts to analyze | |
| Returns: | |
| List of BatchSentimentResult objects | |
| """ | |
| results = [] | |
| for text in texts: | |
| if text.strip(): | |
| analysis_result = self.analyze_sentiment(text) | |
| results.append(BatchSentimentResult( | |
| text=text, | |
| sentiment=analysis_result.sentiment, | |
| confidence=analysis_result.confidence | |
| )) | |
| return results | |
| class NERService: | |
| """Service for Named Entity Recognition operations""" | |
| def __init__(self, model_provider: NERModelProvider): | |
| self.model_provider = model_provider | |
| def extract_entities(self, text: str) -> NERResponse: | |
| """ | |
| Extract named entities from text | |
| Args: | |
| text: The text to process | |
| Returns: | |
| NERResponse with extracted entities | |
| """ | |
| entities_result = self.model_provider.predict(text) | |
| # Convert to Entity objects | |
| entities = [] | |
| for ent in entities_result: | |
| # Handle both aggregation strategies | |
| entity_text = ent.get('word') or ent.get('entity') | |
| entity_label = ent.get('entity_group') or ent.get('entity') | |
| entities.append(Entity( | |
| text=entity_text, | |
| label=entity_label, | |
| score=round(ent['score'], 3) | |
| )) | |
| return NERResponse( | |
| entities=entities, | |
| text=text | |
| ) | |
| class TranslationService: | |
| """Service for translation operations""" | |
| def __init__(self, model_provider: TranslationModelProvider): | |
| self.model_provider = model_provider | |
| def translate( | |
| self, | |
| text: str, | |
| source_lang: str = "en", | |
| target_lang: str = "ar" | |
| ) -> str: | |
| """ | |
| Translate text from source language to target language | |
| Args: | |
| text: The text to translate | |
| source_lang: Source language code | |
| target_lang: Target language code | |
| Returns: | |
| Translated text | |
| """ | |
| translation_result = self.model_provider.predict(text, source_lang, target_lang) | |
| return translation_result[0]['translation_text'] | |
| class ParaphraseService: | |
| """Service for paraphrasing operations""" | |
| def __init__(self, model_provider: ParaphraseModelProvider): | |
| self.model_provider = model_provider | |
| def paraphrase(self, text: str) -> ParaphraseResponse: | |
| """ | |
| Paraphrase the given text | |
| Args: | |
| text: The text to paraphrase | |
| Returns: | |
| ParaphraseResponse object containing the paraphrased text | |
| """ | |
| paraphrase_result = self.model_provider.predict(text) | |
| return ParaphraseResponse(paraphrased_text=paraphrase_result[0]['generated_text']) | |
| class SummarizationService: | |
| """Service for text summarization operations""" | |
| def __init__(self, model_provider: SummarizationModelProvider): | |
| self.model_provider = model_provider | |
| def summarize(self, text: str) -> SummarizationResponse: | |
| """ | |
| Summarize the given text | |
| Args: | |
| text: The text to summarize | |
| Returns: | |
| SummarizationResponse with summarized text | |
| """ | |
| summary_result = self.model_provider.predict(text) | |
| # Hugging Face summarization pipeline returns 'summary_text' key | |
| return SummarizationResponse(summary_text=summary_result[0]['summary_text']) | |