#!/usr/bin/env python3 """ ═══════════════════════════════════════════════════════════════ 🔄 CONSISTENCY CHECKER - ICHIGRIDEA PIPELINE ═══════════════════════════════════════════════════════════════ Vérifie la cohérence entre STATE.json, MODULES_REGISTRY.json et les fichiers réels Usage: python consistency_checker.py [--fix] [--report] """ import json import os import re from pathlib import Path from datetime import datetime class ConsistencyChecker: def __init__(self, base_dir: str = "."): self.base_dir = Path(base_dir) self.errors = [] self.warnings = [] self.info = [] def load_json(self, filename: str) -> dict: """Charge un fichier JSON""" filepath = self.base_dir / filename if filepath.exists(): with open(filepath, "r", encoding="utf-8") as f: return json.load(f) return {} def check_state_json(self): """Vérifie STATE.json vs réalité""" print("🔍 Vérification STATE.json...") state = self.load_json("STATE.json") if not state: self.errors.append("STATE.json manquant ou vide") return # Compter les modules réels include_dir = self.base_dir / "Include" real_modules = list(include_dir.glob("*.mqh")) if include_dir.exists() else [] real_count = len(real_modules) # Comparer avec STATE.json state_count = state.get("statistics", {}).get("total_modules", 0) if real_count != state_count: self.warnings.append( f"Décalage modules: STATE.json={state_count}, réel={real_count}" ) else: self.info.append(f"✅ Modules cohérents: {real_count}") def check_modules_registry(self): """Vérifie MODULES_REGISTRY.json vs fichiers réels""" print("🔍 Vérification MODULES_REGISTRY.json...") registry = self.load_json("MODULES_REGISTRY.json") if not registry: self.errors.append("MODULES_REGISTRY.json manquant ou vide") return include_dir = self.base_dir / "Include" real_modules = {f.stem for f in include_dir.glob("*.mqh")} if include_dir.exists() else set() # Modules dans le registre registry_modules = set(registry.get("modules", {}).keys()) # Modules manquants dans le registre missing_in_registry = real_modules - registry_modules if missing_in_registry: self.warnings.append(f"Modules non enregistrés: {missing_in_registry}") # Modules fantômes dans le registre ghost_modules = registry_modules - real_modules if ghost_modules: self.warnings.append(f"Modules fantômes (dans registre mais pas sur disque): {ghost_modules}") def check_dependencies(self): """Vérifie le graphe des dépendances""" print("🔍 Vérification DEPENDENCIES_GRAPH.json...") deps = self.load_json("DEPENDENCIES_GRAPH.json") if not deps: self.errors.append("DEPENDENCIES_GRAPH.json manquant ou vide") return include_dir = self.base_dir / "Include" real_modules = {f.stem for f in include_dir.glob("*.mqh")} if include_dir.exists() else set() # Vérifier que chaque dépendance existe for module, dependencies in deps.get("module_dependencies", {}).items(): for dep in dependencies: if dep not in real_modules: # Vérifier dans core_dependencies if dep not in deps.get("core_dependencies", {}): self.warnings.append(f"{module} dépend de {dep} qui n'existe pas") def check_inputs_registry(self): """Vérifie que tous les inputs sont documentés""" print("🔍 Vérification INPUTS_REGISTRY.json...") inputs_reg = self.load_json("INPUTS_REGISTRY.json") if not inputs_reg: self.errors.append("INPUTS_REGISTRY.json manquant ou vide") return # Extraire tous les inputs déclarés dans IchiGridEA.mq5 mq5_file = self.base_dir / "IchiGridEA.mq5" if mq5_file.exists(): with open(mq5_file, "r", encoding="utf-8", errors="ignore") as f: content = f.read() # Pattern pour les inputs input_pattern = r"input\s+\w+\s+(\w+)\s*=" real_inputs = set(re.findall(input_pattern, content)) # Inputs dans le registre registry_inputs = set() for category in inputs_reg.get("categories", {}).values(): if isinstance(category, dict): registry_inputs.update(category.keys()) # Inputs non documentés undocumented = real_inputs - registry_inputs if undocumented: self.warnings.append(f"Inputs non documentés: {undocumented}") def check_mq5_includes(self): """Vérifie que tous les #include existent""" print("🔍 Vérification des #include...") mq5_file = self.base_dir / "IchiGridEA.mq5" if not mq5_file.exists(): self.errors.append("IchiGridEA.mq5 manquant") return with open(mq5_file, "r", encoding="utf-8", errors="ignore") as f: content = f.read() # Pattern pour les includes include_pattern = r'#include\s+"([^"]+)"' includes = re.findall(include_pattern, content) for inc in includes: inc_path = self.base_dir / inc if not inc_path.exists(): self.errors.append(f"Include manquant: {inc}") def check_headers(self): """Vérifie les headers de traçabilité""" print("🔍 Vérification des headers de traçabilité...") include_dir = self.base_dir / "Include" if not include_dir.exists(): return modules_without_header = [] for mqh in include_dir.glob("*.mqh"): with open(mqh, "r", encoding="utf-8", errors="ignore") as f: content = f.read(2000) # Lire le début seulement if "@section" not in content: modules_without_header.append(mqh.name) if modules_without_header: self.warnings.append(f"Modules sans header @section: {len(modules_without_header)}") def run_all_checks(self) -> dict: """Exécute toutes les vérifications""" self.check_state_json() self.check_modules_registry() self.check_dependencies() self.check_inputs_registry() self.check_mq5_includes() self.check_headers() return { "timestamp": datetime.utcnow().isoformat() + "Z", "errors": self.errors, "warnings": self.warnings, "info": self.info, "status": "FAIL" if self.errors else ("WARN" if self.warnings else "PASS") } def generate_report(self) -> str: """Génère un rapport markdown""" result = self.run_all_checks() report = f"""# 📊 Rapport de Cohérence IchiGridEA **Date**: {result['timestamp']} **Statut**: {result['status']} ## ❌ Erreurs ({len(result['errors'])}) """ for e in result['errors']: report += f"- {e}\n" report += f""" ## ⚠️ Warnings ({len(result['warnings'])}) """ for w in result['warnings']: report += f"- {w}\n" report += f""" ## ✅ Info ({len(result['info'])}) """ for i in result['info']: report += f"- {i}\n" return report if __name__ == "__main__": import sys checker = ConsistencyChecker() if "--report" in sys.argv: report = checker.generate_report() print(report) # Sauvegarder le rapport with open("pipeline/reports/consistency_report.md", "w") as f: f.write(report) print("\n📄 Rapport sauvegardé: pipeline/reports/consistency_report.md") else: result = checker.run_all_checks() print("\n" + "="*60) print(f"📊 RÉSULTAT: {result['status']}") print("="*60) if result['errors']: print(f"\n❌ ERREURS ({len(result['errors'])}):") for e in result['errors']: print(f" - {e}") if result['warnings']: print(f"\n⚠️ WARNINGS ({len(result['warnings'])}):") for w in result['warnings']: print(f" - {w}") sys.exit(0 if result['status'] == "PASS" else 1)