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