data_base_engineer.py

#!/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"]
            }
        }