Update advanced_tools.py

advanced_tools.py

@@ -1,37 +1,11 @@
 # advanced_tools.py
 import json
 import time
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Optional
+from transformers import pipeline  # type: ignore[import]
+import torch
 import numpy as np
-
-# Safe imports with fallbacks
-try:
-    from transformers import pipeline as transformers_pipeline
-    HAS_TRANSFORMERS = True
-except ImportError:
-    HAS_TRANSFORMERS = False
-    transformers_pipeline = None
-
-try:
-    import torch
-    HAS_TORCH = True
-except ImportError:
-    HAS_TORCH = False
-    torch = None
-
-try:
-    from sentence_transformers import SentenceTransformer
-    HAS_SENTENCE_TRANSFORMERS = True
-except ImportError:
-    HAS_SENTENCE_TRANSFORMERS = False
-    SentenceTransformer = None
-
-try:
-    from sklearn.metrics.pairwise import cosine_similarity
-    HAS_SKLEARN = True
-except ImportError:
-    HAS_SKLEARN = False
-    cosine_similarity = None
+from sentence_transformers import SentenceTransformer
 
 class AdvancedTools:
     def __init__(self):
@@ -43,49 +17,35 @@ class AdvancedTools:
     def _load_sentiment(self):
         """Sentiment model'i yükle"""
         if "sentiment" not in self.models:
-            if not HAS_TRANSFORMERS:
-                raise ImportError("transformers library not available")
-            print("[LOAD] Loading sentiment model...")
-            try:
-                self.models["sentiment"] = transformers_pipeline(
-                    "sentiment-analysis",
-                    model="distilbert-base-uncased-finetuned-sst-2-english",
-                    device=-1  # CPU
-                )
-            except Exception as e:
-                print(f"[ERROR] Failed to load sentiment model: {e}")
-                raise
+            print("🔄 Loading sentiment model...")
+            self.models["sentiment"] = pipeline(  # type: ignore[call-overload]
+                "sentiment-analysis",
+                model="distilbert-base-uncased-finetuned-sst-2-english",
+                device=-1  # CPU
+            )
         return self.models["sentiment"]
     
     def _load_ner(self):
         """NER model'i yükle"""
         if "ner" not in self.models:
-            if not HAS_TRANSFORMERS:
-                raise ImportError("transformers library not available")
-            print("[LOAD] Loading NER model...")
-            try:
-                self.models["ner"] = transformers_pipeline(
-                    "ner",
-                    model="dslim/bert-base-NER",
-                    aggregation_strategy="simple",
-                    device=-1
-                )
-            except Exception as e:
-                print(f"[ERROR] Failed to load NER model: {e}")
-                raise
+            print("🔄 Loading NER model...")
+            self.models["ner"] = pipeline(  # type: ignore[call-overload]
+                "ner",
+                model="dslim/bert-base-NER",
+                aggregation_strategy="simple",
+                device=-1
+            )
         return self.models["ner"]
     
     def _load_embedder(self):
         """Embedding model'i yükle"""
         if self.embedder is None:
-            if not HAS_SENTENCE_TRANSFORMERS:
-                raise ImportError("sentence-transformers library not available")
-            print("[LOAD] Loading embedding model...")
-            try:
-                self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
-            except Exception as e:
-                print(f"[ERROR] Failed to load embedding model: {e}")
-                raise
+            print("🔄 Loading embedding model...")
+            # clean_up_tokenization_spaces parametresini açıkça belirt (future warning için)
+            self.embedder = SentenceTransformer(
+                'all-MiniLM-L6-v2',
+                tokenizer_kwargs={'clean_up_tokenization_spaces': True}
+            )
         return self.embedder
     
     def sentiment_analysis(self, input_data: Dict) -> Dict:
@@ -94,48 +54,42 @@ class AdvancedTools:
         if not text:
             return {"error": "No text provided"}
 
-        if not HAS_TRANSFORMERS:
-            return {"error": "transformers library not available"}
-
-        try:
-            model = self._load_sentiment()
+        model = self._load_sentiment()
 
-            # Metni cümlelere böl
-            sentences = text.split('. ')
-            results = []
+        # Metni cümlelere böl
+        sentences = text.split('. ')
+        results = []
 
-            for sentence in sentences:
-                if len(sentence.strip()) > 3:
-                    result = model(sentence[:512])[0]
-                    results.append({
-                        "sentence": sentence[:50] + "..." if len(sentence) > 50 else sentence,
-                        "sentiment": result["label"],
-                        "confidence": result["score"]
-                    })
-
-            # Genel duygu hesapla
-            if results:
-                positive_count = sum(1 for r in results if r["sentiment"] == "POSITIVE")
-                negative_count = sum(1 for r in results if r["sentiment"] == "NEGATIVE")
-
-                overall = "POSITIVE" if positive_count > negative_count else "NEGATIVE"
-                confidence = max(r["confidence"] for r in results)
-            else:
-                overall = "NEUTRAL"
-                confidence = 0.5
+        for sentence in sentences:
+            if len(sentence.strip()) > 3:
+                result = model(sentence[:512])[0]  # type: ignore[misc]
+                results.append({
+                    "sentence": sentence[:50] + "..." if len(sentence) > 50 else sentence,
+                    "sentiment": result["label"],
+                    "confidence": float(result["score"])  # numpy type -> Python float
+                })
 
-            return {
-                "overall_sentiment": overall,
-                "confidence": confidence,
-                "sentence_analysis": results,
-                "summary": {
-                    "positive_sentences": sum(1 for r in results if r["sentiment"] == "POSITIVE"),
-                    "negative_sentences": sum(1 for r in results if r["sentiment"] == "NEGATIVE"),
-                    "total_sentences": len(results)
-                }
+        # Genel duygu hesapla
+        if results:
+            positive_count = sum(1 for r in results if r["sentiment"] == "POSITIVE")
+            negative_count = sum(1 for r in results if r["sentiment"] == "NEGATIVE")
+
+            overall = "POSITIVE" if positive_count > negative_count else "NEGATIVE"
+            confidence = float(max(r["confidence"] for r in results))  # Ensure Python float
+        else:
+            overall = "NEUTRAL"
+            confidence = 0.5
+
+        return {
+            "overall_sentiment": overall,
+            "confidence": confidence,
+            "sentence_analysis": results,
+            "summary": {
+                "positive_sentences": sum(1 for r in results if r["sentiment"] == "POSITIVE"),
+                "negative_sentences": sum(1 for r in results if r["sentiment"] == "NEGATIVE"),
+                "total_sentences": len(results)
             }
-        except Exception as e:
-            return {"error": f"Sentiment analysis failed: {str(e)}"}
+        }
     
     def entity_extraction(self, input_data: Dict) -> Dict:
         """Named Entity Recognition"""
@@ -143,35 +97,41 @@ class AdvancedTools:
         if not text:
             return {"error": "No text provided"}
 
-        if not HAS_TRANSFORMERS:
-            return {"error": "transformers library not available"}
-
-        try:
-            model = self._load_ner()
-            entities = model(text[:512])
-
-            # Entity'leri grupla
-            grouped = {}
-            for entity in entities:
-                entity_type = entity["entity_group"]
-                if entity_type not in grouped:
-                    grouped[entity_type] = []
-                grouped[entity_type].append({
+        model = self._load_ner()
+        entities = model(text[:512])  # type: ignore[misc]
+
+        # Entity'leri grupla ve numpy tiplerini Python tiplerine çevir
+        grouped = {}
+        serializable_entities = []
+
+        for entity in entities:
+            entity_type = entity["entity_group"]
+            if entity_type not in grouped:
+                grouped[entity_type] = []
+            grouped[entity_type].append({
+                "word": entity["word"],
+                "score": float(entity["score"])  # numpy.float32 -> Python float
+            })
+
+            # İlk 10 entity için serileştirilebilir versiyon
+            if len(serializable_entities) < 10:
+                serializable_entities.append({
+                    "entity_group": entity["entity_group"],
                     "word": entity["word"],
-                    "score": entity["score"]
+                    "score": float(entity["score"]),  # numpy.float32 -> Python float
+                    "start": int(entity["start"]) if "start" in entity else None,
+                    "end": int(entity["end"]) if "end" in entity else None
                 })
 
-            return {
-                "entities": entities[:10],  # İlk 10 entity
-                "grouped": grouped,
-                "summary": {
-                    "total_entities": len(entities),
-                    "entity_types": list(grouped.keys()),
-                    "most_common_type": max(grouped.keys(), key=lambda k: len(grouped[k])) if grouped else None
-                }
+        return {
+            "entities": serializable_entities,  # İlk 10 entity (serileştirilebilir)
+            "grouped": grouped,
+            "summary": {
+                "total_entities": len(entities),
+                "entity_types": list(grouped.keys()),
+                "most_common_type": max(grouped.keys(), key=lambda k: len(grouped[k])) if grouped else None
             }
-        except Exception as e:
-            return {"error": f"Entity extraction failed: {str(e)}"}
+        }
     
     def semantic_similarity(self, input_data: Dict) -> Dict:
         """İki metin arasındaki benzerlik"""
@@ -181,28 +141,24 @@ class AdvancedTools:
         if not text1 or not text2:
             return {"error": "Both text1 and text2 are required"}
 
-        if not HAS_SKLEARN:
-            return {"error": "scikit-learn not available for similarity calculation"}
+        embedder = self._load_embedder()
 
-        try:
-            embedder = self._load_embedder()
-            # Embed metinleri
-            embeddings = embedder.encode([text1, text2])
+        # Embed metinleri
+        embeddings = embedder.encode([text1, text2])  # type: ignore[misc]
 
-            # Cosine similarity hesapla
-            similarity_score = cosine_similarity(
-                [embeddings[0]],
-                [embeddings[1]]
-            )[0][0]
+        # Cosine similarity hesapla
+        from sklearn.metrics.pairwise import cosine_similarity
+        similarity_score = cosine_similarity(
+            [embeddings[0]],
+            [embeddings[1]]
+        )[0][0]
 
-            return {
-                "text1": text1[:100] + "..." if len(text1) > 100 else text1,
-                "text2": text2[:100] + "..." if len(text2) > 100 else text2,
-                "similarity_score": float(similarity_score),
-                "similarity_percentage": round(float(similarity_score) * 100, 2)
-            }
-        except Exception as e:
-            return {"error": f"Similarity calculation failed: {str(e)}"}
+        return {
+            "text1": text1[:100] + "..." if len(text1) > 100 else text1,
+            "text2": text2[:100] + "..." if len(text2) > 100 else text2,
+            "similarity_score": float(similarity_score),
+            "similarity_percentage": round(float(similarity_score) * 100, 2)
+        }
 
     def text_embedding(self, input_data: Dict) -> Dict:
         """Metni vector'e çevir (embedding)"""
@@ -210,21 +166,15 @@ class AdvancedTools:
         if not text:
             return {"error": "No text provided"}
 
-        if not HAS_SENTENCE_TRANSFORMERS:
-            return {"error": "sentence-transformers not available for embeddings"}
-
-        try:
-            embedder = self._load_embedder()
-            embedding = embedder.encode(text)
+        embedder = self._load_embedder()
+        embedding = embedder.encode(text)  # type: ignore[misc]
 
-            return {
-                "text": text[:100] + "..." if len(text) > 100 else text,
-                "embedding": embedding.tolist()[:50],  # İlk 50 dimension
-                "embedding_dimension": len(embedding),
-                "embedding_size_kb": round(len(embedding) * 4 / 1024, 2)
-            }
-        except Exception as e:
-            return {"error": f"Embedding failed: {str(e)}"}
+        return {
+            "text": text[:100] + "..." if len(text) > 100 else text,
+            "embedding": embedding.tolist()[:50],  # İlk 50 dimension
+            "embedding_dimension": len(embedding),
+            "embedding_size_kb": round(len(embedding) * 4 / 1024, 2)
+        }
 
     def smart_cache(self, input_data: Dict) -> Dict:
         """Caching ve cache stats"""
@@ -275,4 +225,4 @@ class AdvancedTools:
 
 
 # Global instance oluştur
-advanced_tools = AdvancedTools() 
+advanced_tools = AdvancedTools()