// ======================================== // BATCH PROCESSOR - SYSTÈME DE QUEUE // Responsabilité: Traitement batch des lignes Google Sheets avec pipeline modulaire // ======================================== const { logSh } = require('../ErrorReporting'); const { tracer } = require('../trace'); const { handleModularWorkflow } = require('../Main'); const { readInstructionsData } = require('../BrainConfig'); const fs = require('fs').promises; const path = require('path'); /** * BATCH PROCESSOR * Système de queue pour traiter les lignes Google Sheets une par une */ class BatchProcessor { constructor() { this.statusPath = path.join(__dirname, '../../config/batch-status.json'); this.configPath = path.join(__dirname, '../../config/batch-config.json'); this.queuePath = path.join(__dirname, '../../config/batch-queue.json'); // État du processeur this.isRunning = false; this.isPaused = false; this.currentRow = null; this.queue = []; this.errors = []; this.results = []; // Configuration par défaut this.config = { selective: 'standardEnhancement', adversarial: 'light', humanSimulation: 'none', patternBreaking: 'none', intensity: 1.0, rowRange: { start: 2, end: 10 }, saveIntermediateSteps: false }; // Métriques this.startTime = null; this.processedCount = 0; this.errorCount = 0; // Callbacks pour updates this.onStatusUpdate = null; this.onProgress = null; this.onError = null; this.onComplete = null; this.initializeProcessor(); } /** * Initialise le processeur */ async initializeProcessor() { try { // Charger la configuration await this.loadConfig(); // Initialiser la queue si elle n'existe pas await this.initializeQueue(); logSh('🎯 BatchProcessor initialisé', 'DEBUG'); } catch (error) { logSh(`❌ Erreur initialisation BatchProcessor: ${error.message}`, 'ERROR'); } } /** * Initialise les fichiers de configuration (alias pour compatibilité tests) */ async initializeFiles() { try { // Créer le dossier config s'il n'existe pas const configDir = path.dirname(this.configPath); await fs.mkdir(configDir, { recursive: true }); // Créer config par défaut si inexistant try { await fs.access(this.configPath); } catch { await fs.writeFile(this.configPath, JSON.stringify(this.config, null, 2)); logSh('📝 Configuration batch par défaut créée', 'DEBUG'); } // Créer status par défaut si inexistant const defaultStatus = { status: 'idle', currentRow: null, totalRows: 0, progress: 0, startTime: null, estimatedEnd: null, errors: [], lastResult: null, config: this.config }; try { await fs.access(this.statusPath); } catch { await fs.writeFile(this.statusPath, JSON.stringify(defaultStatus, null, 2)); logSh('📊 Status batch par défaut créé', 'DEBUG'); } } catch (error) { logSh(`❌ Erreur initialisation fichiers batch: ${error.message}`, 'ERROR'); } } /** * Charge la configuration */ async loadConfig() { try { const configData = await fs.readFile(this.configPath, 'utf8'); this.config = JSON.parse(configData); logSh(`📋 Configuration chargée: ${JSON.stringify(this.config)}`, 'DEBUG'); } catch (error) { logSh('⚠️ Configuration non trouvée, utilisation des valeurs par défaut', 'WARNING'); } } /** * Initialise la queue */ async initializeQueue() { try { // Essayer de charger la queue existante try { const queueData = await fs.readFile(this.queuePath, 'utf8'); const savedQueue = JSON.parse(queueData); if (savedQueue.queue && Array.isArray(savedQueue.queue)) { this.queue = savedQueue.queue; this.processedCount = savedQueue.processedCount || 0; logSh(`📊 Queue restaurée: ${this.queue.length} éléments`, 'DEBUG'); } } catch { // Queue n'existe pas, on la créera } // Si queue vide, la populer depuis la configuration if (this.queue.length === 0) { await this.populateQueue(); } } catch (error) { logSh(`❌ Erreur initialisation queue: ${error.message}`, 'ERROR'); } } /** * Popule la queue avec les lignes à traiter */ async populateQueue() { try { this.queue = []; const { start, end } = this.config.rowRange; for (let rowNumber = start; rowNumber <= end; rowNumber++) { this.queue.push({ rowNumber, status: 'pending', attempts: 0, maxAttempts: 3, error: null, result: null, startTime: null, endTime: null }); } await this.saveQueue(); logSh(`📋 Queue populée: ${this.queue.length} lignes (${start} à ${end})`, 'INFO'); } catch (error) { logSh(`❌ Erreur population queue: ${error.message}`, 'ERROR'); throw error; } } /** * Sauvegarde la queue */ async saveQueue() { try { const queueData = { queue: this.queue, processedCount: this.processedCount, lastUpdate: new Date().toISOString() }; await fs.writeFile(this.queuePath, JSON.stringify(queueData, null, 2)); } catch (error) { logSh(`❌ Erreur sauvegarde queue: ${error.message}`, 'ERROR'); } } // ======================================== // CONTRÔLES PRINCIPAUX // ======================================== /** * Démarre le traitement batch */ async start() { return tracer.run('BatchProcessor.start', async () => { if (this.isRunning) { throw new Error('Le traitement est déjà en cours'); } logSh('🚀 Démarrage traitement batch', 'INFO'); this.isRunning = true; this.isPaused = false; this.startTime = new Date(); this.processedCount = 0; this.errorCount = 0; // Charger la configuration la plus récente await this.loadConfig(); // Si queue vide ou configuration changée, repopuler if (this.queue.length === 0) { await this.populateQueue(); } // Mettre à jour le status await this.updateStatus(); // Démarrer le traitement asynchrone this.processQueue().catch(error => { logSh(`❌ Erreur traitement queue: ${error.message}`, 'ERROR'); this.handleError(error); }); return this.getStatus(); }); } /** * Arrête le traitement batch */ async stop() { return tracer.run('BatchProcessor.stop', async () => { logSh('🛑 Arrêt traitement batch', 'INFO'); this.isRunning = false; this.isPaused = false; this.currentRow = null; await this.updateStatus(); return this.getStatus(); }); } /** * Met en pause le traitement */ async pause() { return tracer.run('BatchProcessor.pause', async () => { if (!this.isRunning) { throw new Error('Aucun traitement en cours'); } logSh('⏸️ Mise en pause traitement batch', 'INFO'); this.isPaused = true; await this.updateStatus(); return this.getStatus(); }); } /** * Reprend le traitement */ async resume() { return tracer.run('BatchProcessor.resume', async () => { if (!this.isRunning || !this.isPaused) { throw new Error('Aucun traitement en pause'); } logSh('▶️ Reprise traitement batch', 'INFO'); this.isPaused = false; await this.updateStatus(); // Reprendre le traitement this.processQueue().catch(error => { logSh(`❌ Erreur reprise traitement: ${error.message}`, 'ERROR'); this.handleError(error); }); return this.getStatus(); }); } // ======================================== // TRAITEMENT QUEUE // ======================================== /** * Traite la queue élément par élément */ async processQueue() { return tracer.run('BatchProcessor.processQueue', async () => { while (this.isRunning && !this.isPaused) { // Chercher le prochain élément à traiter const nextItem = this.queue.find(item => item.status === 'pending' || (item.status === 'error' && item.attempts < item.maxAttempts)); if (!nextItem) { // Queue terminée logSh('✅ Traitement queue terminé', 'INFO'); await this.complete(); break; } // Traiter l'élément await this.processItem(nextItem); // Pause entre les éléments (pour éviter rate limiting) await this.sleep(1000); } }); } /** * Traite un élément de la queue */ async processItem(item) { return tracer.run('BatchProcessor.processItem', async () => { logSh(`🔄 Traitement ligne ${item.rowNumber} (tentative ${item.attempts + 1}/${item.maxAttempts})`, 'INFO'); this.currentRow = item.rowNumber; item.status = 'processing'; item.startTime = new Date().toISOString(); item.attempts++; await this.updateStatus(); await this.saveQueue(); try { // Traiter la ligne avec le pipeline modulaire const result = await this.processRow(item.rowNumber); // Succès item.status = 'completed'; item.result = result; item.endTime = new Date().toISOString(); item.error = null; this.processedCount++; logSh(`✅ Ligne ${item.rowNumber} traitée avec succès`, 'INFO'); // Callback succès if (this.onProgress) { this.onProgress(item, this.getProgress()); } } catch (error) { // Erreur item.error = { message: error.message, stack: error.stack, timestamp: new Date().toISOString() }; if (item.attempts >= item.maxAttempts) { item.status = 'failed'; this.errorCount++; logSh(`❌ Ligne ${item.rowNumber} échouée définitivement après ${item.attempts} tentatives`, 'ERROR'); } else { item.status = 'error'; logSh(`⚠️ Ligne ${item.rowNumber} échouée, retry possible`, 'WARNING'); } // Callback erreur if (this.onError) { this.onError(item, error); } } this.currentRow = null; await this.updateStatus(); await this.saveQueue(); }); } /** * Traite une ligne spécifique */ async processRow(rowNumber) { return tracer.run('BatchProcessor.processRow', { rowNumber }, async () => { // Configuration pour cette ligne const rowConfig = { rowNumber, source: 'batch_processor', selectiveStack: this.config.selective, adversarialMode: this.config.adversarial, humanSimulationMode: this.config.humanSimulation, patternBreakingMode: this.config.patternBreaking, intensity: this.config.intensity, saveIntermediateSteps: this.config.saveIntermediateSteps }; logSh(`🎯 Configuration ligne ${rowNumber}: ${JSON.stringify(rowConfig)}`, 'DEBUG'); // Exécuter le workflow modulaire const result = await handleModularWorkflow(rowConfig); logSh(`📊 Résultat ligne ${rowNumber}: ${result ? 'SUCCESS' : 'FAILED'}`, 'INFO'); return result; }); } // ======================================== // GESTION ÉTAT // ======================================== /** * Met à jour le status */ async updateStatus() { const status = this.getStatus(); try { await fs.writeFile(this.statusPath, JSON.stringify(status, null, 2)); // Callback update if (this.onStatusUpdate) { this.onStatusUpdate(status); } } catch (error) { logSh(`❌ Erreur mise à jour status: ${error.message}`, 'ERROR'); } } /** * Retourne le status actuel */ getStatus() { const now = new Date(); const completedItems = this.queue.filter(item => item.status === 'completed').length; const failedItems = this.queue.filter(item => item.status === 'failed').length; const totalItems = this.queue.length; const progress = totalItems > 0 ? ((completedItems + failedItems) / totalItems) * 100 : 0; let status = 'idle'; if (this.isRunning && this.isPaused) { status = 'paused'; } else if (this.isRunning) { status = 'running'; } else if (completedItems + failedItems === totalItems && totalItems > 0) { status = 'completed'; } return { status, currentRow: this.currentRow, totalRows: totalItems, completedRows: completedItems, failedRows: failedItems, progress: Math.round(progress), startTime: this.startTime ? this.startTime.toISOString() : null, estimatedEnd: this.estimateCompletionTime(), errors: this.queue.filter(item => item.error).map(item => ({ rowNumber: item.rowNumber, error: item.error, attempts: item.attempts })), lastResult: this.getLastResult(), config: this.config, queue: this.queue }; } /** * Retourne la progression détaillée */ getProgress() { const status = this.getStatus(); const now = new Date(); const elapsed = this.startTime ? now - this.startTime : 0; const avgTimePerRow = status.completedRows > 0 ? elapsed / status.completedRows : 0; const remainingRows = status.totalRows - status.completedRows - status.failedRows; const estimatedRemaining = avgTimePerRow * remainingRows; return { ...status, metrics: { elapsedTime: elapsed, avgTimePerRow: avgTimePerRow, estimatedRemaining: estimatedRemaining, completionPercentage: status.progress, throughput: status.completedRows > 0 && elapsed > 0 ? (status.completedRows / (elapsed / 1000 / 60)) : 0 // rows/minute } }; } /** * Estime l'heure de fin */ estimateCompletionTime() { if (!this.startTime || !this.isRunning || this.isPaused) { return null; } const progress = this.getProgress(); if (progress.metrics.estimatedRemaining > 0) { const endTime = new Date(Date.now() + progress.metrics.estimatedRemaining); return endTime.toISOString(); } return null; } /** * Retourne le dernier résultat */ getLastResult() { const completedItems = this.queue.filter(item => item.status === 'completed'); if (completedItems.length === 0) return null; const lastItem = completedItems[completedItems.length - 1]; return { rowNumber: lastItem.rowNumber, result: lastItem.result, endTime: lastItem.endTime }; } /** * Gère les erreurs critiques */ async handleError(error) { logSh(`💥 Erreur critique BatchProcessor: ${error.message}`, 'ERROR'); this.isRunning = false; this.isPaused = false; await this.updateStatus(); if (this.onError) { this.onError(null, error); } } /** * Termine le traitement */ async complete() { logSh('🏁 Traitement batch terminé', 'INFO'); this.isRunning = false; this.isPaused = false; this.currentRow = null; await this.updateStatus(); if (this.onComplete) { this.onComplete(this.getStatus()); } } // ======================================== // UTILITAIRES // ======================================== /** * Pause l'exécution */ async sleep(ms) { return new Promise(resolve => setTimeout(resolve, ms)); } /** * Reset la queue */ async resetQueue() { logSh('🔄 Reset de la queue', 'INFO'); this.queue = []; this.processedCount = 0; this.errorCount = 0; await this.populateQueue(); await this.updateStatus(); } /** * Configure les callbacks */ setCallbacks({ onStatusUpdate, onProgress, onError, onComplete }) { this.onStatusUpdate = onStatusUpdate; this.onProgress = onProgress; this.onError = onError; this.onComplete = onComplete; } } // ============= EXPORTS ============= module.exports = { BatchProcessor };