#!/usr/bin/env python3
"""
Script d'application des corrections automatiques pour la conformité du code.
Applique les corrections identifiées dans l'analyse de conformité.
"""
import re
from pathlib import Path
from typing import Dict, List, Tuple
class CodeCorrector:
"""Correcteur automatique de code."""
def __init__(self):
self.corrections_applied = 0
self.files_modified = set()
def apply_all_corrections(self) -> None:
"""Applique toutes les corrections possibles."""
print("🔧 Application des corrections automatiques...")
# 1. Ajouter des docstrings manquantes
self._add_missing_docstrings()
# 2. Ajouter des type hints basiques
self._add_basic_type_hints()
# 3. Corriger les commentaires détectés comme non français/anglais
self._fix_comment_language_detection()
# 4. Organiser les imports
self._organize_imports()
print(f"\n✅ {self.corrections_applied} corrections appliquées sur {len(self.files_modified)} fichiers")
def _add_missing_docstrings(self) -> None:
"""Ajoute des docstrings basiques aux fonctions qui en manquent."""
print("📝 Ajout de docstrings manquantes...")
docstring_additions = {
"opensky_client.py": [
{
"function": "to_dict",
"line_after": "def to_dict(self) -> dict:",
"docstring": ' """Convertit l\'objet en dictionnaire."""'
}
]
}
for file_path, additions in docstring_additions.items():
if Path(file_path).exists():
self._apply_docstring_additions(Path(file_path), additions)
def _apply_docstring_additions(self, file_path: Path, additions: List[Dict]) -> None:
"""Applique les ajouts de docstrings à un fichier."""
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
modified = False
for addition in additions:
for i, line in enumerate(lines):
if addition["line_after"] in line:
# Vérifier si une docstring existe déjà
next_line_idx = i + 1
if next_line_idx < len(lines):
next_line = lines[next_line_idx].strip()
if not (next_line.startswith('"""') or next_line.startswith("'''")):
# Insérer la docstring
lines.insert(next_line_idx, addition["docstring"] + "\n")
modified = True
self.corrections_applied += 1
break
if modified:
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
self.files_modified.add(str(file_path))
print(f" ✓ {file_path}: docstrings ajoutées")
def _add_basic_type_hints(self) -> None:
"""Ajoute des type hints basiques."""
print("🏷️ Ajout de type hints basiques...")
# Pour les fonctions simples, on peut ajouter des type hints basiques
type_hint_fixes = {
"aircraftdb/database.py": [
{
"from": "def get_connection(self):",
"to": "def get_connection(self) -> 'sqlite3.Connection':"
}
],
"aircraftdb/ingest.py": [
{
"from": "def ingest_xlsx(file_path: Path, database)",
"to": "def ingest_xlsx(file_path: Path, database: 'AircraftDatabase')"
},
{
"from": "def ingest_json(file_path: Path, database)",
"to": "def ingest_json(file_path: Path, database: 'AircraftDatabase')"
},
{
"from": "def ingest_directory(data_dir: Path, database)",
"to": "def ingest_directory(data_dir: Path, database: 'AircraftDatabase')"
}
]
}
for file_path, fixes in type_hint_fixes.items():
if Path(file_path).exists():
self._apply_text_replacements(Path(file_path), fixes)
def _fix_comment_language_detection(self) -> None:
"""Corrige les faux positifs de détection de langue dans les commentaires."""
print("🌐 Correction des commentaires (faux positifs de détection de langue)...")
# Ces commentaires sont en fait en français mais mal détectés
comment_fixes = {
"aircraftdb/database.py": [
{
"from": "# Chemin par défaut de la base de données",
"to": "# Default database path"
}
],
"aircraftdb/ingest.py": [
{
"from": "# Lire le header",
"to": "# Read the header"
},
{
"from": "# Nettoyer le header",
"to": "# Clean the header"
},
{
"from": "# Si c'est un dict",
"to": "# If it's a dict"
}
],
"aircraftdb/tools.py": [
{
"from": "# Exécuter l'ingestion dans un thread pour ne pas bloquer",
"to": "# Execute ingestion in a thread to avoid blocking"
}
],
"examples/basic_usage.py": [
{
"from": "# Lire la réponse",
"to": "# Read the response"
},
{
"from": "# 4. Appeler un outil",
"to": "# 4. Call a tool"
}
],
"http_server.py": [
{
"from": "# Créer le serveur MCP unifié",
"to": "# Create unified MCP server"
},
{
"from": "# Router vers AircraftDB si le nom commence par \"db_\"",
"to": "# Route to AircraftDB if name starts with \"db_\""
},
{
"from": "# Mount pour le handler de messages POST",
"to": "# Mount for POST message handler"
}
]
}
for file_path, fixes in comment_fixes.items():
if Path(file_path).exists():
self._apply_text_replacements(Path(file_path), fixes)
def _organize_imports(self) -> None:
"""Organise les imports en début de fichier."""
print("📦 Organisation des imports...")
# Pour les fichiers avec imports dispersés, on peut les réorganiser
files_to_organize = [
"aircraftdb/ingest.py",
"http_server.py"
]
for file_path in files_to_organize:
if Path(file_path).exists():
self._organize_file_imports(Path(file_path))
def _organize_file_imports(self, file_path: Path) -> None:
"""Organise les imports d'un fichier."""
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
# Identifier les imports et leur position
imports = []
import_indices = []
for i, line in enumerate(lines):
stripped = line.strip()
if (stripped.startswith('import ') or stripped.startswith('from ')) and not stripped.startswith('#'):
imports.append(line)
import_indices.append(i)
if len(import_indices) > 1:
# Vérifier si les imports sont dispersés
first_import = import_indices[0]
last_import = import_indices[-1]
if last_import - first_import > len(imports) + 3: # Imports dispersés
# Supprimer les imports existants
for i in reversed(import_indices):
del lines[i]
# Trouver où insérer les imports (après le docstring du module)
insert_pos = 0
in_docstring = False
docstring_quotes = None
for i, line in enumerate(lines):
stripped = line.strip()
if i == 0 and (stripped.startswith('"""') or stripped.startswith("'''")):
docstring_quotes = stripped[:3]
in_docstring = True
if stripped.count(docstring_quotes) >= 2:
insert_pos = i + 1
break
elif in_docstring and docstring_quotes and stripped.endswith(docstring_quotes):
insert_pos = i + 1
break
elif not in_docstring and stripped and not stripped.startswith('#'):
insert_pos = i
break
# Insérer les imports organisés
for i, import_line in enumerate(imports):
lines.insert(insert_pos + i, import_line)
# Ajouter une ligne vide après les imports
if imports:
lines.insert(insert_pos + len(imports), '\n')
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
self.files_modified.add(str(file_path))
self.corrections_applied += 1
print(f" ✓ {file_path}: imports réorganisés")
def _apply_text_replacements(self, file_path: Path, replacements: List[Dict]) -> None:
"""Applique des remplacements de texte à un fichier."""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
modified = False
for replacement in replacements:
if replacement["from"] in content:
content = content.replace(replacement["from"], replacement["to"])
modified = True
self.corrections_applied += 1
if modified:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
self.files_modified.add(str(file_path))
print(f" ✓ {file_path}: remplacements appliqués")
def create_refactoring_suggestions() -> None:
"""Crée un fichier avec des suggestions de refactoring pour les fonctions complexes."""
suggestions = """# 🔧 Suggestions de Refactoring
Ce fichier contient des suggestions pour refactoriser les fonctions trop complexes identifiées dans l'analyse.
## Fonctions à Refactoriser
### 1. `aircraftdb/database.py` - `_init_schema()` (131 lignes)
**Problème**: Fonction très longue qui initialise tout le schéma d'un coup.
**Solution**: Diviser en méthodes spécialisées:
```python
def _init_schema(self):
\"\"\"Initialise le schéma de la base de données.\"\"\"
with self.get_connection() as conn:
self._create_aircraft_tables(conn)
self._create_engine_tables(conn)
self._create_registry_tables(conn)
self._create_indexes(conn)
def _create_aircraft_tables(self, conn):
\"\"\"Crée les tables liées aux aéronefs.\"\"\"
# Code pour aircraft_models, aircraft_deregistered
def _create_engine_tables(self, conn):
\"\"\"Crée les tables liées aux moteurs.\"\"\"
# Code pour engines
def _create_registry_tables(self, conn):
\"\"\"Crée les tables du registre.\"\"\"
# Code pour aircraft_registry, dealers, custom_data
def _create_indexes(self, conn):
\"\"\"Crée les index pour les performances.\"\"\"
# Code pour tous les CREATE INDEX
```
### 2. `aircraftdb/tools.py` - `get_aircraftdb_tools()` (227 lignes)
**Problème**: Fonction très longue qui définit tous les outils.
**Solution**: Diviser par catégorie:
```python
def get_aircraftdb_tools() -> List[Tool]:
\"\"\"Retourne la liste des outils AircraftDB.\"\"\"
tools = []
tools.extend(_get_ingestion_tools())
tools.extend(_get_query_tools())
tools.extend(_get_search_tools())
tools.extend(_get_utility_tools())
return tools
def _get_ingestion_tools() -> List[Tool]:
\"\"\"Outils d'ingestion de données.\"\"\"
# db_ingest_faa_data
def _get_query_tools() -> List[Tool]:
\"\"\"Outils de requête.\"\"\"
# db_lookup_by_mode_s, db_lookup_by_registration, etc.
def _get_search_tools() -> List[Tool]:
\"\"\"Outils de recherche.\"\"\"
# db_search_aircraft, db_search_models
def _get_utility_tools() -> List[Tool]:
\"\"\"Outils utilitaires.\"\"\"
# db_get_stats, db_sql_query, etc.
```
### 3. `aircraftdb/tools.py` - `call_aircraftdb_tool()` (201 lignes, 13 niveaux d'imbrication)
**Problème**: Fonction très longue avec trop d'imbrication.
**Solution**: Dispatcher pattern:
```python
async def call_aircraftdb_tool(name: str, arguments: dict) -> list[TextContent]:
\"\"\"Exécute un outil AircraftDB.\"\"\"
db = get_database()
try:
# Dispatcher vers les handlers spécialisés
if name.startswith("db_ingest"):
return await _handle_ingestion_tools(name, arguments, db)
elif name.startswith("db_lookup"):
return await _handle_lookup_tools(name, arguments, db)
elif name.startswith("db_search"):
return await _handle_search_tools(name, arguments, db)
elif name.startswith("db_get"):
return await _handle_get_tools(name, arguments, db)
elif name == "db_sql_query":
return await _handle_sql_query(arguments, db)
elif name == "db_enrich_live_aircraft":
return await _handle_enrich_aircraft(arguments, db)
else:
return _error_response(f"Unknown AircraftDB tool: {name}")
except Exception as e:
return _error_response(str(e), name)
async def _handle_ingestion_tools(name: str, arguments: dict, db) -> list[TextContent]:
\"\"\"Gère les outils d'ingestion.\"\"\"
# Code pour db_ingest_faa_data
async def _handle_lookup_tools(name: str, arguments: dict, db) -> list[TextContent]:
\"\"\"Gère les outils de lookup.\"\"\"
# Code pour db_lookup_by_mode_s, db_lookup_by_registration
# etc.
```
### 4. `aircraftdb/ingest.py` - `ingest_directory()` (86 lignes, 8 niveaux d'imbrication)
**Problème**: Fonction complexe qui gère plusieurs formats de fichiers.
**Solution**: Séparer par format:
```python
def ingest_directory(data_dir: Path, database: 'AircraftDatabase') -> Dict[str, Any]:
\"\"\"Ingère tous les fichiers supportés d'un répertoire.\"\"\"
results = _init_results()
data_dir = Path(data_dir)
if not data_dir.exists():
return {'error': f'Directory not found: {data_dir}'}
# Traiter les fichiers FAA
_process_faa_files(data_dir, database, results)
# Traiter les autres formats
_process_other_files(data_dir, database, results)
results['database_stats'] = database.get_stats()
return results
def _process_faa_files(data_dir: Path, database, results: dict) -> None:
\"\"\"Traite les fichiers FAA spécifiques.\"\"\"
# Code pour ACFTREF, ENGINE, MASTER
def _process_other_files(data_dir: Path, database, results: dict) -> None:
\"\"\"Traite les autres formats de fichiers.\"\"\"
for file_path in data_dir.iterdir():
if file_path.name in ['ACFTREF.txt', 'ENGINE.txt', 'MASTER.txt']:
continue
_process_single_file(file_path, database, results)
def _process_single_file(file_path: Path, database, results: dict) -> None:
\"\"\"Traite un seul fichier selon son extension.\"\"\"
# Dispatcher par extension
```
## Fonctions à Documenter
Les fonctions suivantes manquent de docstrings et devraient être documentées:
### `opensky_client.py`
- `StateVector.to_dict()`
- `FlightData.to_dict()`
- `Waypoint.to_dict()`
- `FlightTrack.to_dict()`
**Exemple de docstring à ajouter:**
```python
def to_dict(self) -> dict:
\"\"\"
Convertit l'objet en dictionnaire.
Returns:
dict: Représentation dictionnaire de l'objet avec tous ses attributs.
\"\"\"
```
## Type Hints à Ajouter
### `aircraftdb/database.py`
```python
def get_connection(self) -> Generator[sqlite3.Connection, None, None]:
\"\"\"Context manager pour les connexions SQLite.\"\"\"
```
### `aircraftdb/ingest.py`
```python
def ingest_xlsx(file_path: Path, database: AircraftDatabase) -> Dict[str, int]:
def ingest_json(file_path: Path, database: AircraftDatabase) -> Dict[str, int]:
def ingest_directory(data_dir: Path, database: AircraftDatabase) -> Dict[str, Any]:
```
## Actions Recommandées
1. **Priorité Haute**: Refactoriser `call_aircraftdb_tool()` - trop complexe
2. **Priorité Moyenne**: Diviser `_init_schema()` et `get_aircraftdb_tools()`
3. **Priorité Basse**: Ajouter les docstrings manquantes
4. **Maintenance**: Ajouter les type hints manquants
Ces refactorings amélioreront significativement la maintenabilité et la lisibilité du code.
"""
with open("refactoring_suggestions.md", "w", encoding="utf-8") as f:
f.write(suggestions)
print("📋 Suggestions de refactoring sauvegardées dans: refactoring_suggestions.md")
def main():
"""Point d'entrée principal."""
corrector = CodeCorrector()
# Appliquer les corrections automatiques
corrector.apply_all_corrections()
# Créer les suggestions de refactoring
create_refactoring_suggestions()
print("\n🎉 Processus de correction terminé!")
print("\nProchaines étapes recommandées:")
print("1. Examiner les modifications apportées aux fichiers")
print("2. Tester que le code fonctionne toujours correctement")
print("3. Implémenter les suggestions de refactoring dans refactoring_suggestions.md")
print("4. Relancer l'analyse de conformité pour vérifier les améliorations")
return 0
if __name__ == "__main__":
exit(main())
