Upload app_ultimate.py with huggingface_hub

app_ultimate.py

@@ -20,6 +20,20 @@ import pickle
 from urllib.parse import urljoin, urlparse
 import threading
 from pathlib import Path
+import numpy as np
+from sklearn.ensemble import IsolationForest
+from sklearn.metrics.pairwise import cosine_similarity
+import warnings
+warnings.filterwarnings('ignore')
+
+# AI/ML Imports for enhanced functionality
+try:
+    from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
+    from sentence_transformers import SentenceTransformer
+    ML_AVAILABLE = True
+except ImportError:
+    ML_AVAILABLE = False
+    st.warning("⚠️ ML libraries not available. Some AI features will be disabled.")
 
 # Enhanced Page Configuration
 st.set_page_config(
@@ -131,6 +145,439 @@ DB_PATH = "ultimate_data_harvester.db"
 SESSION_PATH = "harvester_session.pkl"
 ENDPOINTS_CACHE = "discovered_endpoints.json"
 
+# AI Enhancement Classes
+class AIDataQualityAssessor:
+    """AI-powered data quality assessment using transformers"""
+    
+    def __init__(self):
+        self.quality_model = None
+        self.embeddings_model = None
+        self._initialize_models()
+    
+    def _initialize_models(self):
+        """Initialize AI models for quality assessment"""
+        if ML_AVAILABLE:
+            try:
+                # Initialize quality classifier
+                self.quality_model = pipeline(
+                    "text-classification",
+                    model="distilbert-base-uncased-finetuned-sst-2-english",
+                    return_all_scores=True
+                )
+                
+                # Initialize embeddings model for similarity
+                self.embeddings_model = SentenceTransformer('all-MiniLM-L6-v2')
+                
+                st.success("✅ AI models loaded successfully!")
+            except Exception as e:
+                st.warning(f"⚠️ Failed to load AI models: {e}")
+                ML_AVAILABLE = False
+    
+    def assess_data_quality(self, data: Any, api_name: str) -> Dict:
+        """Comprehensive AI-powered data quality assessment"""
+        if not ML_AVAILABLE or not self.quality_model:
+            return self._basic_quality_assessment(data, api_name)
+        
+        try:
+            # Convert data to text for analysis
+            text_data = self._data_to_text(data)
+            
+            # AI quality scoring
+            ai_scores = self.quality_model(text_data[:512])  # Limit to 512 chars
+            quality_score = max([score['score'] for score in ai_scores[0]])
+            
+            # Basic quality metrics
+            completeness = self._check_completeness(data)
+            consistency = self._check_consistency(data, api_name)
+            structure_quality = self._assess_structure(data)
+            
+            # Anomaly detection
+            anomalies = self._detect_anomalies(data)
+            
+            return {
+                "ai_quality_score": round(quality_score, 3),
+                "completeness_score": completeness,
+                "consistency_score": consistency,
+                "structure_score": structure_quality,
+                "anomaly_count": len(anomalies),
+                "anomalies": anomalies[:5],  # Top 5 anomalies
+                "overall_grade": self._calculate_overall_grade(
+                    quality_score, completeness, consistency, structure_quality
+                ),
+                "recommendations": self._generate_quality_recommendations(
+                    quality_score, completeness, consistency, anomalies
+                )
+            }
+        
+        except Exception as e:
+            st.warning(f"AI quality assessment failed: {e}")
+            return self._basic_quality_assessment(data, api_name)
+    
+    def _data_to_text(self, data: Any) -> str:
+        """Convert any data format to text for AI analysis"""
+        if isinstance(data, str):
+            return data
+        elif isinstance(data, dict):
+            return json.dumps(data, ensure_ascii=False)[:1000]
+        elif isinstance(data, list):
+            return str(data)[:1000]
+        else:
+            return str(data)[:1000]
+    
+    def _check_completeness(self, data: Any) -> float:
+        """Check data completeness"""
+        if isinstance(data, dict):
+            total_fields = len(data)
+            complete_fields = sum(1 for v in data.values() if v is not None and v != "")
+            return complete_fields / total_fields if total_fields > 0 else 0.0
+        elif isinstance(data, list):
+            if not data:
+                return 0.0
+            if isinstance(data[0], dict):
+                return np.mean([self._check_completeness(item) for item in data])
+            return 1.0
+        return 1.0 if data is not None else 0.0
+    
+    def _check_consistency(self, data: Any, api_name: str) -> float:
+        """Check data consistency based on API expectations"""
+        consistency_score = 1.0
+        
+        if isinstance(data, list):
+            if len(data) > 1:
+                # Check if all items have similar structure
+                first_item = data[0] if data else {}
+                if isinstance(first_item, dict):
+                    first_keys = set(first_item.keys())
+                    consistency_scores = []
+                    for item in data[1:6]:  # Check first 5 items
+                        if isinstance(item, dict):
+                            item_keys = set(item.keys())
+                            similarity = len(first_keys & item_keys) / len(first_keys | item_keys)
+                            consistency_scores.append(similarity)
+                    
+                    if consistency_scores:
+                        consistency_score = np.mean(consistency_scores)
+        
+        return consistency_score
+    
+    def _assess_structure(self, data: Any) -> float:
+        """Assess data structure quality"""
+        if isinstance(data, dict):
+            # Check for nested structure, proper keys, etc.
+            score = 0.8  # Base score for dictionary
+            if len(data) > 0:
+                score += 0.1
+            if any(isinstance(v, (dict, list)) for v in data.values()):
+                score += 0.1  # Bonus for nested structure
+            return min(score, 1.0)
+        elif isinstance(data, list):
+            return 0.9 if data else 0.5
+        else:
+            return 0.6  # Basic data
+    
+    def _detect_anomalies(self, data: Any) -> List[str]:
+        """Detect data anomalies"""
+        anomalies = []
+        
+        if isinstance(data, dict):
+            # Check for suspicious values
+            for key, value in data.items():
+                if value is None:
+                    anomalies.append(f"Null value in field: {key}")
+                elif isinstance(value, str) and len(value) > 1000:
+                    anomalies.append(f"Unusually long string in field: {key}")
+                elif isinstance(value, (int, float)) and abs(value) > 1e10:
+                    anomalies.append(f"Extreme numeric value in field: {key}")
+        
+        elif isinstance(data, list):
+            if len(data) > 10000:
+                anomalies.append(f"Very large dataset: {len(data)} items")
+            
+            # Check for inconsistent types
+            if data:
+                first_type = type(data[0])
+                if not all(isinstance(item, first_type) for item in data[:10]):
+                    anomalies.append("Inconsistent data types in list")
+        
+        return anomalies
+    
+    def _calculate_overall_grade(self, ai_score: float, completeness: float, 
+                               consistency: float, structure: float) -> str:
+        """Calculate overall data quality grade"""
+        overall_score = (ai_score + completeness + consistency + structure) / 4
+        
+        if overall_score >= 0.9:
+            return "A+ (Excellent)"
+        elif overall_score >= 0.8:
+            return "A (Very Good)"
+        elif overall_score >= 0.7:
+            return "B (Good)"
+        elif overall_score >= 0.6:
+            return "C (Fair)"
+        else:
+            return "D (Poor)"
+    
+    def _generate_quality_recommendations(self, ai_score: float, completeness: float, 
+                                        consistency: float, anomalies: List[str]) -> List[str]:
+        """Generate AI-powered recommendations for data quality improvement"""
+        recommendations = []
+        
+        if ai_score < 0.7:
+            recommendations.append("📝 Consider data validation and cleaning")
+        
+        if completeness < 0.8:
+            recommendations.append("🔍 Investigate missing data fields")
+        
+        if consistency < 0.8:
+            recommendations.append("⚙️ Standardize data format across records")
+        
+        if len(anomalies) > 3:
+            recommendations.append("🚨 Multiple anomalies detected - requires investigation")
+        
+        if not recommendations:
+            recommendations.append("✅ Data quality is good - no immediate action needed")
+        
+        return recommendations
+    
+    def _basic_quality_assessment(self, data: Any, api_name: str) -> Dict:
+        """Basic quality assessment without AI"""
+        return {
+            "ai_quality_score": 0.0,
+            "completeness_score": self._check_completeness(data),
+            "consistency_score": 0.8,  # Default
+            "structure_score": self._assess_structure(data),
+            "anomaly_count": 0,
+            "anomalies": [],
+            "overall_grade": "C (Basic Assessment)",
+            "recommendations": ["Install ML libraries for advanced AI assessment"]
+        }
+
+class SemanticDataAnalyzer:
+    """Semantic analysis and similarity detection"""
+    
+    def __init__(self):
+        self.embeddings_model = None
+        self.stored_embeddings = {}
+        self._initialize_model()
+    
+    def _initialize_model(self):
+        """Initialize sentence transformer model"""
+        if ML_AVAILABLE:
+            try:
+                self.embeddings_model = SentenceTransformer('all-MiniLM-L6-v2')
+            except Exception as e:
+                st.warning(f"Failed to load embeddings model: {e}")
+    
+    def find_similar_datasets(self, new_data: Any, api_name: str, threshold: float = 0.85) -> List[Dict]:
+        """Find semantically similar datasets"""
+        if not self.embeddings_model:
+            return []
+        
+        try:
+            # Convert data to text and create embedding
+            text_data = self._data_to_text(new_data)
+            new_embedding = self.embeddings_model.encode([text_data])
+            
+            # Compare with stored embeddings
+            similar_datasets = []
+            for stored_key, stored_embedding in self.stored_embeddings.items():
+                similarity = cosine_similarity(new_embedding, [stored_embedding])[0][0]
+                if similarity > threshold:
+                    similar_datasets.append({
+                        "dataset": stored_key,
+                        "similarity": float(similarity),
+                        "api_name": stored_key.split("_")[0] if "_" in stored_key else "unknown"
+                    })
+            
+            # Store new embedding
+            embedding_key = f"{api_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+            self.stored_embeddings[embedding_key] = new_embedding[0]
+            
+            return sorted(similar_datasets, key=lambda x: x['similarity'], reverse=True)
+        
+        except Exception as e:
+            st.warning(f"Semantic analysis failed: {e}")
+            return []
+    
+    def _data_to_text(self, data: Any) -> str:
+        """Convert data to text for embedding"""
+        if isinstance(data, str):
+            return data[:500]
+        elif isinstance(data, dict):
+            # Extract key information
+            text_parts = []
+            for key, value in list(data.items())[:10]:  # First 10 keys
+                text_parts.append(f"{key}: {str(value)[:100]}")
+            return " | ".join(text_parts)
+        elif isinstance(data, list) and data:
+            return str(data[0])[:500]
+        else:
+            return str(data)[:500]
+
+class APIHealthMonitor:
+    """Intelligent API health monitoring with anomaly detection"""
+    
+    def __init__(self):
+        self.anomaly_detector = IsolationForest(contamination=0.1, random_state=42)
+        self.health_history = {}
+        self.is_trained = False
+    
+    def monitor_api_health(self, api_name: str, response_time: float, 
+                          success_rate: float, data_size: int) -> Dict:
+        """Comprehensive API health assessment"""
+        current_metrics = {
+            "response_time": response_time,
+            "success_rate": success_rate,
+            "data_size": data_size,
+            "timestamp": time.time()
+        }
+        
+        # Store health history
+        if api_name not in self.health_history:
+            self.health_history[api_name] = []
+        
+        self.health_history[api_name].append(current_metrics)
+        
+        # Keep only last 50 measurements
+        if len(self.health_history[api_name]) > 50:
+            self.health_history[api_name] = self.health_history[api_name][-50:]
+        
+        # Calculate health score
+        health_score = self._calculate_health_score(current_metrics)
+        
+        # Detect anomalies if we have enough data
+        anomaly_score = 0.0
+        if len(self.health_history[api_name]) >= 10:
+            anomaly_score = self._detect_performance_anomaly(api_name, current_metrics)
+        
+        # Generate recommendations
+        recommendations = self._generate_health_recommendations(
+            current_metrics, health_score, anomaly_score
+        )
+        
+        return {
+            "health_score": health_score,
+            "status": self._get_health_status(health_score),
+            "anomaly_score": anomaly_score,
+            "is_anomaly": anomaly_score < -0.5,
+            "recommendations": recommendations,
+            "trend": self._calculate_trend(api_name),
+            "metrics": current_metrics
+        }
+    
+    def _calculate_health_score(self, metrics: Dict) -> float:
+        """Calculate overall health score (0-1)"""
+        # Response time score (lower is better)
+        time_score = max(0, 1 - (metrics["response_time"] / 10000))  # 10s max
+        
+        # Success rate score
+        success_score = metrics["success_rate"]
+        
+        # Data size score (normalized)
+        size_score = min(1.0, metrics["data_size"] / 1000000)  # 1MB reference
+        
+        # Weighted average
+        health_score = (time_score * 0.4 + success_score * 0.5 + size_score * 0.1)
+        return max(0, min(1, health_score))
+    
+    def _detect_performance_anomaly(self, api_name: str, current_metrics: Dict) -> float:
+        """Detect performance anomalies using isolation forest"""
+        try:
+            history = self.health_history[api_name]
+            
+            # Prepare training data
+            training_data = []
+            for h in history[:-1]:  # Exclude current measurement
+                training_data.append([
+                    h["response_time"],
+                    h["success_rate"],
+                    h["data_size"]
+                ])
+            
+            if len(training_data) >= 5:
+                # Train anomaly detector
+                self.anomaly_detector.fit(training_data)
+                
+                # Check current metrics
+                current_data = [[
+                    current_metrics["response_time"],
+                    current_metrics["success_rate"],
+                    current_metrics["data_size"]
+                ]]
+                
+                anomaly_score = self.anomaly_detector.decision_function(current_data)[0]
+                return float(anomaly_score)
+        
+        except Exception as e:
+            st.warning(f"Anomaly detection failed: {e}")
+        
+        return 0.0
+    
+    def _get_health_status(self, health_score: float) -> str:
+        """Get health status based on score"""
+        if health_score >= 0.9:
+            return "🟢 Excellent"
+        elif health_score >= 0.7:
+            return "🟡 Good"
+        elif health_score >= 0.5:
+            return "🟠 Fair"
+        else:
+            return "🔴 Poor"
+    
+    def _generate_health_recommendations(self, metrics: Dict, health_score: float, 
+                                       anomaly_score: float) -> List[str]:
+        """Generate health improvement recommendations"""
+        recommendations = []
+        
+        if metrics["response_time"] > 5000:
+            recommendations.append("⏱️ High response time detected - consider caching")
+        
+        if metrics["success_rate"] < 0.9:
+            recommendations.append("❌ Low success rate - check API status")
+        
+        if anomaly_score < -0.5:
+            recommendations.append("🚨 Performance anomaly detected - investigate")
+        
+        if health_score < 0.6:
+            recommendations.append("⚠️ Overall poor health - consider alternatives")
+        
+        if not recommendations:
+            recommendations.append("✅ API performing well")
+        
+        return recommendations
+    
+    def _calculate_trend(self, api_name: str) -> str:
+        """Calculate performance trend"""
+        if api_name not in self.health_history or len(self.health_history[api_name]) < 5:
+            return "📊 Insufficient data"
+        
+        recent_scores = []
+        for metrics in self.health_history[api_name][-5:]:
+            score = self._calculate_health_score(metrics)
+            recent_scores.append(score)
+        
+        if len(recent_scores) >= 3:
+            trend = np.polyfit(range(len(recent_scores)), recent_scores, 1)[0]
+            
+            if trend > 0.02:
+                return "📈 Improving"
+            elif trend < -0.02:
+                return "📉 Declining"
+            else:
+                return "➡️ Stable"
+        
+        return "📊 Monitoring"
+
+# Initialize AI components
+if ML_AVAILABLE:
+    ai_quality_assessor = AIDataQualityAssessor()
+    semantic_analyzer = SemanticDataAnalyzer()
+    health_monitor = APIHealthMonitor()
+else:
+    ai_quality_assessor = None
+    semantic_analyzer = None
+    health_monitor = None
+
 # Comprehensive API Discovery Configuration
 DEEP_API_CONFIG = {
     "Skolverket": {
@@ -1050,7 +1497,7 @@ class UltimateDataHarvester:
     def _save_harvested_data(self, api_name: str, endpoint_path: str, data: Any, 
                            session_id: str, fetch_duration: int, record_count: int, 
                            data_size: int, status: str = "success", error_message: str = None):
-        """Save harvested data with intelligent categorization"""
+        """Save harvested data with AI-enhanced intelligent categorization"""
         conn = sqlite3.connect(DB_PATH)
         cursor = conn.cursor()
         
@@ -1058,22 +1505,114 @@ class UltimateDataHarvester:
         data_str = json.dumps(data, sort_keys=True, default=str)
         data_hash = hashlib.sha256(data_str.encode()).hexdigest()
         
+        # AI Quality Assessment
+        quality_assessment = {}
+        if ai_quality_assessor and status == "success":
+            quality_assessment = ai_quality_assessor.assess_data_quality(data, api_name)
+        
+        # Semantic Similarity Analysis
+        similar_datasets = []
+        if semantic_analyzer and status == "success":
+            similar_datasets = semantic_analyzer.find_similar_datasets(data, api_name)
+        
+        # API Health Monitoring
+        health_info = {}
+        if health_monitor:
+            success_rate = 1.0 if status == "success" else 0.0
+            health_info = health_monitor.monitor_api_health(
+                api_name, fetch_duration, success_rate, data_size
+            )
+        
         try:
             cursor.execute('''
                 INSERT OR REPLACE INTO harvested_data 
                 (api_name, endpoint_path, data_hash, raw_data, processed_data, 
                  record_count, data_size_bytes, fetch_duration_ms, status, 
-                 error_message, session_id)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                 error_message, session_id, quality_score, health_score, similar_datasets)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             ''', (
                 api_name, endpoint_path, data_hash, data_str, 
                 json.dumps(data, default=str), record_count, data_size, 
-                fetch_duration, status, error_message, session_id
+                fetch_duration, status, error_message, session_id,
+                quality_assessment.get('ai_quality_score', 0.0),
+                health_info.get('health_score', 0.0),
+                json.dumps(similar_datasets[:3], default=str)  # Top 3 similar datasets
             ))
             
             conn.commit()
+            
+            # Display AI insights if available
+            if quality_assessment and st.session_state.get('show_ai_insights', True):
+                self._display_ai_insights(api_name, quality_assessment, health_info, similar_datasets)
+                
         except sqlite3.IntegrityError:
             pass  # Data already exists
+        except sqlite3.OperationalError:
+            # Handle case where AI columns don't exist yet - add them
+            self._upgrade_database_schema()
+            # Retry with basic data
+            cursor.execute('''
+                INSERT OR REPLACE INTO harvested_data 
+                (api_name, endpoint_path, data_hash, raw_data, processed_data, 
+                 record_count, data_size_bytes, fetch_duration_ms, status, 
+                 error_message, session_id)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ''', (
+                api_name, endpoint_path, data_hash, data_str, 
+                json.dumps(data, default=str), record_count, data_size, 
+                fetch_duration, status, error_message, session_id
+            ))
+            conn.commit()
+        finally:
+            conn.close()
+    
+    def _display_ai_insights(self, api_name: str, quality_assessment: Dict, 
+                           health_info: Dict, similar_datasets: List[Dict]):
+        """Display AI-powered insights in real-time"""
+        if quality_assessment:
+            with st.expander(f"🤖 AI Insights for {api_name}", expanded=False):
+                col1, col2, col3 = st.columns(3)
+                
+                with col1:
+                    st.metric("Quality Grade", quality_assessment.get('overall_grade', 'N/A'))
+                    st.metric("Completeness", f"{quality_assessment.get('completeness_score', 0):.2f}")
+                
+                with col2:
+                    if health_info:
+                        st.metric("Health Status", health_info.get('status', 'Unknown'))
+                        st.metric("Performance Trend", health_info.get('trend', 'N/A'))
+                
+                with col3:
+                    st.metric("Anomalies", quality_assessment.get('anomaly_count', 0))
+                    if similar_datasets:
+                        st.metric("Similar Datasets", len(similar_datasets))
+                
+                # Recommendations
+                recommendations = quality_assessment.get('recommendations', [])
+                if recommendations:
+                    st.write("**🎯 Recommendations:**")
+                    for rec in recommendations[:3]:
+                        st.write(f"• {rec}")
+                
+                # Similar datasets
+                if similar_datasets:
+                    st.write("**🔍 Similar Datasets Found:**")
+                    for sim in similar_datasets[:2]:
+                        st.write(f"• {sim['dataset']} (similarity: {sim['similarity']:.2f})")
+    
+    def _upgrade_database_schema(self):
+        """Upgrade database schema to include AI columns"""
+        conn = sqlite3.connect(DB_PATH)
+        cursor = conn.cursor()
+        
+        try:
+            # Add AI enhancement columns
+            cursor.execute('ALTER TABLE harvested_data ADD COLUMN quality_score REAL DEFAULT 0.0')
+            cursor.execute('ALTER TABLE harvested_data ADD COLUMN health_score REAL DEFAULT 0.0') 
+            cursor.execute('ALTER TABLE harvested_data ADD COLUMN similar_datasets TEXT DEFAULT "[]"')
+            conn.commit()
+        except sqlite3.OperationalError:
+            pass  # Columns already exist
         finally:
             conn.close()
     
@@ -1263,7 +1802,16 @@ st.markdown("### 🚀 Operations")
 tab1, tab2, tab3 = st.tabs(["🔍 Deep Discovery", "📊 Data Harvesting", "📈 Analytics"])
 
 with tab1:
-    st.markdown("**Discover all possible endpoints from API sources**")
+    st.markdown("**🤖 AI-Enhanced Deep Discovery - Find all endpoints with intelligent analysis**")
+    
+    # AI Settings
+    col1, col2 = st.columns(2)
+    with col1:
+        enable_ai_insights = st.checkbox("🤖 Enable AI Quality Assessment", value=True, key="enable_ai")
+    with col2:
+        show_similarity = st.checkbox("🔍 Show Semantic Similarity", value=True, key="enable_similarity")
+    
+    st.session_state['show_ai_insights'] = enable_ai_insights
     
     # API Selection for Discovery
     selected_apis_discovery = st.multiselect(
@@ -1467,13 +2015,43 @@ with tab3:
     finally:
         conn.close()
 
+# AI Enhancement Panel
+if ML_AVAILABLE:
+    st.markdown("---")
+    with st.expander("🤖 AI Enhancement Status", expanded=False):
+        col1, col2, col3 = st.columns(3)
+        
+        with col1:
+            st.markdown("**🎯 Quality Assessment**")
+            if ai_quality_assessor and ai_quality_assessor.quality_model:
+                st.success("✅ Active - DistilBERT")
+            else:
+                st.error("❌ Not Available")
+        
+        with col2:
+            st.markdown("**🔍 Semantic Analysis**")
+            if semantic_analyzer and semantic_analyzer.embeddings_model:
+                st.success("✅ Active - MiniLM-L6-v2")
+            else:
+                st.error("❌ Not Available")
+        
+        with col3:
+            st.markdown("**📊 Health Monitoring**")
+            if health_monitor:
+                st.success("✅ Active - Isolation Forest")
+            else:
+                st.error("❌ Not Available")
+        
+        if ai_quality_assessor and hasattr(ai_quality_assessor, 'quality_model'):
+            st.info("💡 AI models are loaded and ready for enhanced data analysis!")
+
 # Footer
 st.markdown("---")
 st.markdown("""
 <div style="text-align: center; padding: 1rem; opacity: 0.9;">
-    <p><strong>🚀 Ultimate Data Harvester</strong> - Deep discovery, session resumption, intelligent storage</p>
+    <p><strong>🚀 Ultimate Data Harvester with AI</strong> - Deep discovery, session resumption, intelligent storage</p>
     <p style="font-size: 0.9rem;">
-        🔍 Recursive endpoint discovery • 🎯 Session management • 💾 Smart database storage • 📊 Real-time analytics
+        🔍 Recursive endpoint discovery • 🤖 AI quality assessment • 🎯 Session management • 💾 Smart database storage • 📊 Real-time analytics
     </p>
 </div>
 """, unsafe_allow_html=True)