Spaces:
Running
Running
File size: 5,769 Bytes
e4eb82b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 |
"""
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'])
|