Create data_base_engineer.py

data_base_engineer.py

@@ -0,0 +1,581 @@
+#!/usr/bin/env python3
+"""
+Ingénieur Automatique de Bases de Données
+Système intelligent de correction, exécution et mise à jour automatique
+"""
+
+import sqlite3
+import mysql.connector
+import psycopg2
+import pandas as pd
+import logging
+import asyncio
+from typing import Dict, List, Any, Optional, Tuple
+from dataclasses import dataclass
+from datetime import datetime
+import re
+import json
+import hashlib
+
+@dataclass
+class DatabaseSchema:
+    """Schéma de base de données"""
+    tables: Dict[str, Any]
+    indexes: List[Dict]
+    relationships: List[Dict]
+    metadata: Dict[str, Any]
+
+@dataclass
+class QueryAnalysis:
+    """Analyse de requête SQL"""
+    query_type: str
+    tables_affected: List[str]
+    columns_affected: List[str]
+    potential_issues: List[str]
+    optimization_suggestions: List[str]
+    execution_plan: Optional[Dict]
+
+class AutomaticDatabaseEngineer:
+    """
+    Ingénieur automatique pour bases de données
+    """
+    
+    def __init__(self):
+        self.logger = logging.getLogger("db_engineer")
+        self.connections: Dict[str, Any] = {}
+        self.schemas: Dict[str, DatabaseSchema] = {}
+        
+        # Règles de correction automatique
+        self.auto_fix_rules = {
+            "syntax_error": {
+                "patterns": [
+                    r"near\s+\"([^\"]+)\"",
+                    r"syntax\s+error",
+                    r"unexpected\s+token"
+                ],
+                "fixes": self._fix_syntax_errors
+            },
+            "table_not_found": {
+                "patterns": [
+                    r"table\s+[\"']?([\w]+)[\"']?\s+not\s+found",
+                    r"no\s+such\s+table"
+                ],
+                "fixes": self._fix_table_issues
+            },
+            "column_not_found": {
+                "patterns": [
+                    r"column\s+[\"']?([\w]+)[\"']?\s+not\s+found",
+                    r"no\s+such\s+column"
+                ],
+                "fixes": self._fix_column_issues
+            },
+            "constraint_violation": {
+                "patterns": [
+                    r"constraint\s+failed",
+                    r"unique\s+constraint",
+                    r"foreign\s+key\s+constraint"
+                ],
+                "fixes": self._fix_constraint_issues
+            },
+            "performance_issue": {
+                "patterns": [
+                    r"slow\s+query",
+                    r"full\s+table\s+scan",
+                    r"index\s+missing"
+                ],
+                "fixes": self._fix_performance_issues
+            }
+        }
+        
+        # Templates de schémas optimisés
+        self.optimized_templates = {
+            "ecommerce": self._get_ecommerce_schema(),
+            "blog": self._get_blog_schema(),
+            "analytics": self._get_analytics_schema(),
+            "user_management": self._get_user_management_schema()
+        }
+    
+    async def connect_database(self, db_type: str, connection_params: Dict) -> str:
+        """Établit une connexion à la base de données"""
+        connection_id = f"{db_type}_{hashlib.md5(str(connection_params).encode()).hexdigest()[:8]}"
+        
+        try:
+            if db_type == "sqlite":
+                self.connections[connection_id] = sqlite3.connect(
+                    connection_params['database'],
+                    check_same_thread=False
+                )
+            elif db_type == "mysql":
+                self.connections[connection_id] = mysql.connector.connect(
+                    host=connection_params.get('host', 'localhost'),
+                    user=connection_params.get('user', 'root'),
+                    password=connection_params.get('password', ''),
+                    database=connection_params.get('database', '')
+                )
+            elif db_type == "postgresql":
+                self.connections[connection_id] = psycopg2.connect(
+                    host=connection_params.get('host', 'localhost'),
+                    user=connection_params.get('user', 'postgres'),
+                    password=connection_params.get('password', ''),
+                    database=connection_params.get('database', 'postgres')
+                )
+            else:
+                raise ValueError(f"Type de base de données non supporté: {db_type}")
+            
+            # Analyse du schéma existant
+            await self._analyze_schema(connection_id, db_type)
+            
+            self.logger.info(f"Connexion établie: {connection_id}")
+            return connection_id
+            
+        except Exception as e:
+            self.logger.error(f"Erreur connexion {db_type}: {e}")
+            raise
+    
+    async def execute_and_fix_query(self, connection_id: str, query: str, max_attempts: int = 3) -> Dict[str, Any]:
+        """Exécute une requête avec correction automatique en cas d'erreur"""
+        attempt = 0
+        original_query = query
+        fixes_applied = []
+        
+        while attempt < max_attempts:
+            try:
+                result = await self._execute_query(connection_id, query)
+                
+                return {
+                    "success": True,
+                    "result": result,
+                    "query_executed": query,
+                    "fixes_applied": fixes_applied,
+                    "attempts": attempt + 1
+                }
+                
+            except Exception as e:
+                error_msg = str(e)
+                self.logger.warning(f"Erreur exécution (tentative {attempt + 1}): {error_msg}")
+                
+                # Tentative de correction automatique
+                fixed_query = await self._auto_fix_query(query, error_msg, connection_id)
+                
+                if fixed_query and fixed_query != query:
+                    query = fixed_query
+                    fixes_applied.append({
+                        "original_error": error_msg,
+                        "fix_description": "Correction automatique appliquée",
+                        "fixed_query": fixed_query
+                    })
+                    attempt += 1
+                else:
+                    # Impossible de corriger automatiquement
+                    return {
+                        "success": False,
+                        "error": error_msg,
+                        "original_query": original_query,
+                        "fixes_attempted": fixes_applied,
+                        "suggested_fix": await self._suggest_manual_fix(original_query, error_msg)
+                    }
+        
+        return {
+            "success": False,
+            "error": "Échec après plusieurs tentatives de correction",
+            "original_query": original_query,
+            "fixes_applied": fixes_applied
+        }
+    
+    async def auto_optimize_database(self, connection_id: str, optimization_type: str = "full") -> Dict[str, Any]:
+        """Optimisation automatique de la base de données"""
+        optimizations_applied = []
+        
+        try:
+            # Analyse des performances
+            performance_analysis = await self._analyze_performance(connection_id)
+            
+            if optimization_type in ["full", "indexes"]:
+                # Optimisation des index
+                index_optimizations = await self._optimize_indexes(connection_id)
+                optimizations_applied.extend(index_optimizations)
+            
+            if optimization_type in ["full", "schema"]:
+                # Optimisation du schéma
+                schema_optimizations = await self._optimize_schema(connection_id)
+                optimizations_applied.extend(schema_optimizations)
+            
+            if optimization_type in ["full", "maintenance"]:
+                # Maintenance générale
+                maintenance_ops = await self._perform_maintenance(connection_id)
+                optimizations_applied.extend(maintenance_ops)
+            
+            return {
+                "success": True,
+                "optimizations_applied": optimizations_applied,
+                "performance_improvement": await self._measure_performance_improvement(connection_id),
+                "recommendations": await self._generate_recommendations(connection_id)
+            }
+            
+        except Exception as e:
+            self.logger.error(f"Erreur optimisation: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "optimizations_applied": optimizations_applied
+            }
+    
+    async def intelligent_migration(self, connection_id: str, target_schema: Dict) -> Dict[str, Any]:
+        """Migration intelligente vers un nouveau schéma"""
+        migration_steps = []
+        current_schema = self.schemas[connection_id]
+        
+        try:
+            # Analyse des différences
+            schema_diff = await self._compare_schemas(current_schema, target_schema)
+            
+            # Génération des étapes de migration
+            migration_plan = await self._generate_migration_plan(schema_diff)
+            
+            # Exécution sécurisée de la migration
+            for step in migration_plan:
+                try:
+                    result = await self.execute_and_fix_query(connection_id, step['query'])
+                    
+                    migration_steps.append({
+                        "step": step['description'],
+                        "query": step['query'],
+                        "success": result['success'],
+                        "details": result
+                    })
+                    
+                    if not result['success']:
+                        # Rollback partiel ou correction
+                        await self._handle_migration_failure(connection_id, migration_steps, step)
+                        break
+                        
+                except Exception as e:
+                    migration_steps.append({
+                        "step": step['description'],
+                        "error": str(e),
+                        "success": False
+                    })
+                    break
+            
+            # Vérification finale
+            if all(step.get('success', False) for step in migration_steps):
+                await self._verify_migration(connection_id, target_schema)
+                return {
+                    "success": True,
+                    "migration_steps": migration_steps,
+                    "message": "Migration terminée avec succès"
+                }
+            else:
+                return {
+                    "success": False,
+                    "migration_steps": migration_steps,
+                    "error": "Migration échouée partiellement"
+                }
+                
+        except Exception as e:
+            self.logger.error(f"Erreur migration: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "migration_steps": migration_steps
+            }
+    
+    async def real_time_monitoring(self, connection_id: str) -> Dict[str, Any]:
+        """Surveillance en temps réel de la base de données"""
+        monitoring_data = {
+            "timestamp": datetime.now().isoformat(),
+            "performance_metrics": await self._get_performance_metrics(connection_id),
+            "query_analysis": await self._analyze_active_queries(connection_id),
+            "resource_usage": await self._get_resource_usage(connection_id),
+            "alerts": await self._check_alerts(connection_id),
+            "recommendations": await self._generate_realtime_recommendations(connection_id)
+        }
+        
+        return monitoring_data
+    
+    async def _auto_fix_query(self, query: str, error: str, connection_id: str) -> Optional[str]:
+        """Correction automatique des requêtes basée sur les erreurs"""
+        for rule_type, rule in self.auto_fix_rules.items():
+            for pattern in rule['patterns']:
+                if re.search(pattern, error, re.IGNORECASE):
+                    fixed_query = await rule['fixes'](query, error, connection_id)
+                    if fixed_query:
+                        self.logger.info(f"Correction appliquée ({rule_type}): {fixed_query}")
+                        return fixed_query
+        
+        return None
+    
+    async def _fix_syntax_errors(self, query: str, error: str, connection_id: str) -> Optional[str]:
+        """Correction des erreurs de syntaxe"""
+        # Correction des guillemets mal fermés
+        if query.count('"') % 2 != 0:
+            fixed = query + '"'
+            return fixed
+        
+        # Correction des parenthèses mal fermées
+        if query.count('(') > query.count(')'):
+            fixed = query + ')' * (query.count('(') - query.count(')'))
+            return fixed
+        
+        # Correction des virgules en fin de SELECT
+        if re.search(r"SELECT\s+.+,\s+FROM", query, re.IGNORECASE):
+            fixed = re.sub(r",\s+FROM", " FROM", query, flags=re.IGNORECASE)
+            return fixed
+        
+        return None
+    
+    async def _fix_table_issues(self, query: str, error: str, connection_id: str) -> Optional[str]:
+        """Correction des problèmes de tables"""
+        # Extraction du nom de table de l'erreur
+        table_match = re.search(r"table\s+[\"']?([\w]+)[\"']?", error, re.IGNORECASE)
+        if table_match:
+            table_name = table_match.group(1)
+            
+            # Vérification si la table existe avec une casse différente
+            schema = self.schemas[connection_id]
+            actual_tables = list(schema.tables.keys())
+            
+            for actual_table in actual_tables:
+                if actual_table.lower() == table_name.lower():
+                    # Correction de la casse
+                    fixed = re.sub(
+                        f"\\b{table_name}\\b", 
+                        actual_table, 
+                        query, 
+                        flags=re.IGNORECASE
+                    )
+                    return fixed
+            
+            # Suggestion de création de table si approprié
+            if "CREATE" not in query.upper() and "DROP" not in query.upper():
+                create_query = await self._suggest_table_creation(table_name, query)
+                return create_query
+        
+        return None
+    
+    async def _fix_column_issues(self, query: str, error: str, connection_id: str) -> Optional[str]:
+        """Correction des problèmes de colonnes"""
+        column_match = re.search(r"column\s+[\"']?([\w]+)[\"']?", error, re.IGNORECASE)
+        if column_match:
+            column_name = column_match.group(1)
+            
+            # Recherche de la colonne correcte dans le schéma
+            schema = self.schemas[connection_id]
+            
+            for table_name, table_info in schema.tables.items():
+                if column_name in table_info.get('columns', {}):
+                    # La colonne existe dans cette table
+                    return query
+                else:
+                    # Recherche de colonnes similaires
+                    for actual_column in table_info.get('columns', {}).keys():
+                        if actual_column.lower() == column_name.lower():
+                            fixed = re.sub(
+                                f"\\b{column_name}\\b", 
+                                actual_column, 
+                                query, 
+                                flags=re.IGNORECASE
+                            )
+                            return fixed
+        
+        return None
+    
+    async def _analyze_schema(self, connection_id: str, db_type: str):
+        """Analyse complète du schéma de base de données"""
+        schema = DatabaseSchema(tables={}, indexes=[], relationships=[], metadata={})
+        
+        try:
+            # Récupération des tables
+            if db_type == "sqlite":
+                tables_query = "SELECT name FROM sqlite_master WHERE type='table';"
+            elif db_type == "mysql":
+                tables_query = "SHOW TABLES;"
+            elif db_type == "postgresql":
+                tables_query = """
+                    SELECT table_name 
+                    FROM information_schema.tables 
+                    WHERE table_schema = 'public';
+                """
+            
+            cursor = self.connections[connection_id].cursor()
+            cursor.execute(tables_query)
+            tables = cursor.fetchall()
+            
+            for table in tables:
+                table_name = table[0] if isinstance(table, (list, tuple)) else table
+                schema.tables[table_name] = await self._analyze_table(connection_id, table_name, db_type)
+            
+            # Analyse des indexes
+            schema.indexes = await self._analyze_indexes(connection_id, db_type)
+            
+            # Analyse des relations
+            schema.relationships = await self._analyze_relationships(connection_id, db_type)
+            
+            self.schemas[connection_id] = schema
+            
+        except Exception as e:
+            self.logger.error(f"Erreur analyse schéma: {e}")
+    
+    async def _analyze_table(self, connection_id: str, table_name: str, db_type: str) -> Dict[str, Any]:
+        """Analyse détaillée d'une table"""
+        table_info = {"columns": {}, "constraints": [], "indexes": []}
+        
+        try:
+            cursor = self.connections[connection_id].cursor()
+            
+            if db_type == "sqlite":
+                # Structure des colonnes
+                cursor.execute(f"PRAGMA table_info({table_name});")
+                columns = cursor.fetchall()
+                for col in columns:
+                    table_info["columns"][col[1]] = {
+                        "type": col[2],
+                        "nullable": not col[3],
+                        "default": col[4],
+                        "primary_key": col[5] == 1
+                    }
+                
+            elif db_type == "mysql":
+                cursor.execute(f"DESCRIBE {table_name};")
+                columns = cursor.fetchall()
+                for col in columns:
+                    table_info["columns"][col[0]] = {
+                        "type": col[1],
+                        "nullable": col[2] == "YES",
+                        "default": col[4],
+                        "primary_key": col[3] == "PRI"
+                    }
+            
+            # Statistiques de la table
+            cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
+            table_info["row_count"] = cursor.fetchone()[0]
+            
+        except Exception as e:
+            self.logger.error(f"Erreur analyse table {table_name}: {e}")
+        
+        return table_info
+    
+    async def _execute_query(self, connection_id: str, query: str) -> Any:
+        """Exécution sécurisée d'une requête"""
+        connection = self.connections[connection_id]
+        cursor = connection.cursor()
+        
+        try:
+            cursor.execute(query)
+            
+            if query.strip().upper().startswith(('SELECT', 'SHOW', 'DESCRIBE', 'EXPLAIN')):
+                result = cursor.fetchall()
+                columns = [desc[0] for desc in cursor.description] if cursor.description else []
+                return {"data": result, "columns": columns}
+            else:
+                connection.commit()
+                return {"rows_affected": cursor.rowcount}
+                
+        finally:
+            cursor.close()
+    
+    async def _suggest_manual_fix(self, query: str, error: str) -> Dict[str, Any]:
+        """Suggestion de correction manuelle"""
+        suggestions = {
+            "original_error": error,
+            "suggested_fixes": [],
+            "alternative_queries": [],
+            "documentation_references": []
+        }
+        
+        # Suggestions basées sur le type d'erreur
+        if "syntax" in error.lower():
+            suggestions["suggested_fixes"].append("Vérifiez la syntaxe SQL, particulièrement les guillemets et parenthèses")
+            suggestions["suggested_fixes"].append("Utilisez un outil de formatage SQL pour identifier les erreurs")
+        
+        if "table" in error.lower() and "not found" in error.lower():
+            suggestions["suggested_fixes"].append("Vérifiez le nom de la table dans le schéma de la base de données")
+            suggestions["suggested_fixes"].append("Assurez-vous que la table existe et que vous avez les permissions nécessaires")
+        
+        if "column" in error.lower() and "not found" in error.lower():
+            suggestions["suggested_fixes"].append("Vérifiez le nom des colonnes dans la structure de la table")
+            suggestions["suggested_fixes"].append("Utilisez SELECT * FROM table LIMIT 1 pour voir la structure")
+        
+        # Génération de requêtes alternatives
+        query_upper = query.upper()
+        if "SELECT" in query_upper:
+            # Suggestion d'index
+            suggestions["alternative_queries"].append(
+                "Pensez à ajouter des INDEX sur les colonnes utilisées dans WHERE et JOIN"
+            )
+        
+        return suggestions
+    
+    # Méthodes d'optimisation (implémentations simplifiées)
+    async def _optimize_indexes(self, connection_id: str) -> List[Dict]:
+        """Optimisation automatique des index"""
+        optimizations = []
+        schema = self.schemas[connection_id]
+        
+        # Analyse des colonnes fréquemment utilisées dans les WHERE
+        for table_name, table_info in schema.tables.items():
+            columns_usage = await self._analyze_column_usage(connection_id, table_name)
+            
+            for column, usage in columns_usage.items():
+                if usage['filter_usage'] > 10 and not self._has_index(table_name, column, schema):
+                    # Création d'index suggérée
+                    index_query = f"CREATE INDEX idx_{table_name}_{column} ON {table_name}({column});"
+                    try:
+                        await self._execute_query(connection_id, index_query)
+                        optimizations.append({
+                            "type": "index_creation",
+                            "table": table_name,
+                            "column": column,
+                            "query": index_query,
+                            "impact": "high"
+                        })
+                    except Exception as e:
+                        self.logger.warning(f"Impossible de créer l'index: {e}")
+        
+        return optimizations
+    
+    async def _analyze_performance(self, connection_id: str) -> Dict[str, Any]:
+        """Analyse des performances de la base de données"""
+        analysis = {
+            "slow_queries": [],
+            "missing_indexes": [],
+            "table_scans": [],
+            "lock_contention": []
+        }
+        
+        # Implémentation simplifiée
+        # Dans une version complète, on utiliserait les métriques système
+        
+        return analysis
+    
+    def _get_ecommerce_schema(self) -> Dict:
+        """Template de schéma e-commerce optimisé"""
+        return {
+            "users": {
+                "columns": {
+                    "id": "INT PRIMARY KEY AUTO_INCREMENT",
+                    "email": "VARCHAR(255) UNIQUE NOT NULL",
+                    "created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
+                },
+                "indexes": ["email"]
+            },
+            "products": {
+                "columns": {
+                    "id": "INT PRIMARY KEY AUTO_INCREMENT", 
+                    "name": "VARCHAR(255) NOT NULL",
+                    "price": "DECIMAL(10,2)",
+                    "category_id": "INT"
+                },
+                "indexes": ["name", "category_id", "price"]
+            },
+            "orders": {
+                "columns": {
+                    "id": "INT PRIMARY KEY AUTO_INCREMENT",
+                    "user_id": "INT",
+                    "status": "VARCHAR(50)",
+                    "total_amount": "DECIMAL(10,2)",
+                    "created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
+                },
+                "indexes": ["user_id", "status", "created_at"]
+            }
+        }