#!/usr/bin/env python3 """ πŸ›’οΈ DIESEL CFDI CONSOLIDATION SCRIPT Extracts diesel line items from CFDI (Mexican tax invoice) XLSX exports, consolidates them by invoice, calculates taxes, and produces formatted reports. ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ FEATURES: βœ“ Phase 1: Diesel consolidation with IVA calculations βœ“ Phase 2: Stimulus summary with tax rate reductions βœ“ Automatic stimulus schedule loading (CSV or XLSX) βœ“ Majority-vote fallback for null/inconsistent RFC/Anio/Mes columns βœ“ Multi-language support (Spanish/English) βœ“ Detailed logging with DEBUG mode βœ“ Windows 10 & macOS compatible ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ BASIC USAGE: python diesel.py --input-file report.xlsx FULL CLI PARAMETERS: --input-file FILE Path to CFDI XLSX file (REQUIRED) --output DIR Output directory (default: ./output) Example: --output /path/to/reports -y, --yes Skip confirmation prompt (for scripts/CI) --generate-csv Also generate the flat CSV output file (default: False, only XLSX is generated) --log-level LEVEL Logging verbosity (default: INFO) Options: DEBUG, INFO, WARNING, ERROR Use DEBUG to see stimulus loading details --cuota-ieps RATE Base IEPS tax rate per liter (default: 7.0946) Can also use env var: DIESEL_COUTA_IEPS=8.5 Example: --cuota-ieps 8.5 --lang LANGUAGE Output language (default: es) Options: es (Spanish), en (English) Affects: sheet names, headers, messages ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ EXAMPLES: # macOS/Linux basic python3 diesel.py --input-file ~/reports/cfdi_2026_07.xlsx # Windows 10 with output directory python diesel.py --input-file C:\\Reports\\diesel.xlsx --output C:\\Output # Skip confirmation (CI/automation) python diesel.py --input-file report.xlsx -y # Debug mode with custom tax rate python diesel.py --input-file report.xlsx --log-level DEBUG --cuota-ieps 8.5 # English output with custom output directory python diesel.py --input-file report.xlsx --lang en --output ./reports # Also generate the flat CSV output file python diesel.py --input-file report.xlsx --generate-csv # Full example with all parameters python diesel.py \\ --input-file /data/cfdi.xlsx \\ --output ./consolidated \\ --cuota-ieps 7.50 \\ --lang es \\ --log-level DEBUG \\ --generate-csv \\ -y # Using environment variable for tax rate export DIESEL_COUTA_IEPS=8.0 python diesel.py --input-file report.xlsx -y ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ STIMULUS SCHEDULE: The script looks for estimulos.xlsx (or estimulos.csv as fallback) in the same directory as the script. Format: fecha_inicio | fecha_fin | porcentaje 2026-07-05 | 2026-07-12 | 17.85 2026-07-13 | 2026-07-20 | 26.26 ... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ OUTPUT FILES: {RFC}_{YEAR}_{MONTH}_{TIMESTAMP}.xlsx - Formatted report with 2 sheets {RFC}_{YEAR}_{MONTH}_{TIMESTAMP}.csv - Flat CSV for import (only with --generate-csv) logs/diesel_YYYYMMDD_HHMMSS.log - Detailed log file for this execution logs/diesel_history.log - Persistent log, appended across all executions EXIT CODES: 0: Success 1: Validation error or user declined 2: Unexpected error (see log file) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ """ import sys import argparse import logging import logging.handlers import re import csv import json import os from collections import Counter from datetime import datetime, date from pathlib import Path from typing import List, Dict, Optional, Tuple try: from openpyxl import Workbook, load_workbook from openpyxl.styles import Alignment, Font, PatternFill, Border, Side except ImportError: print("ERROR: openpyxl is not installed.") print("Install it with: pip install -r requirements.txt") print("Or: pip install openpyxl==3.1.5") sys.exit(2) def get_base_dir() -> Path: """Directory to resolve relative files (estimulos.csv) from. Returns the .exe's own directory when frozen by PyInstaller (sys.executable), since __file__ would otherwise point into a temp extraction folder. Returns the script's directory otherwise. """ if getattr(sys, 'frozen', False): return Path(sys.executable).parent return Path(__file__).parent class Colors: """ANSI color codes for terminal output (cross-platform).""" GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' CYAN = '\033[96m' BLUE = '\033[94m' MAGENTA = '\033[95m' WHITE = '\033[97m' BOLD = '\033[1m' RESET = '\033[0m' @staticmethod def disable_on_windows(): """Disable colors on Windows if needed (usually not necessary on Win10+).""" import platform if platform.system() == 'Windows': os.system('color') # Enable ANSI on Windows 10+ # Language translations TRANSLATIONS = { 'es': { 'phase1_title': 'DATOS CFDI POR CONSUMO DE DIESEL', 'phase2_title': 'RESUMEN ESTIMULO FISCAL - DIESEL', 'summary_header': '================================================================================\n πŸ›’οΈ DIESEL CFDI CONSOLIDACION - RESUMEN\n================================================================================', 'no_items_found': 'No se encontraron facturas con items de diesel. Saliendo.', 'input_file': 'Archivo de entrada', 'rows_scanned': 'Filas escaneadas', 'invoices_found': 'Facturas con items de diesel', 'total_items': 'Total de items de diesel', 'total_litros': 'Total de Litros (Cantidad)', 'subtotal_importe': 'Subtotal (Importe)', 'total_iva': 'IVA (16%)', 'total': 'TOTAL', 'output_files': 'Archivos de salida (a escribirse en', 'preview': 'Vista previa', 'first_invoices': 'primeras facturas', 'line_items': 'Items de linea', 'subtotal': 'Subtotal', 'files_written': 'Archivos escritos exitosamente', 'logs': 'Registros', 'proceed_prompt': 'Proceder con la escritura de archivos de salida? [s/N]: ', 'aborted': 'Abortado.', 'phase1_headers': ['FECHA', 'FOLIO FISCAL', 'LITROS', 'CUOTA', 'IMPORTE'], 'phase2_headers': ['FECHA', 'FOLIO FISCAL', 'MES', 'LITROS', 'CUOTA IEPS (ART 2 FRACC I INCISO D LIEPS)', '% ESTIMULO (ART PRIMERO)', 'MONTO ESTIMULO FISCAL (ART SEGUNDO)', 'CUOTA DISMINUIDA VIGENTE (ART TERCERO)', 'MONTO DEL ESTIMULO FISCAL'], 'january': 'ENERO', 'february': 'FEBRERO', 'march': 'MARZO', 'april': 'ABRIL', 'may': 'MAYO', 'june': 'JUNIO', 'july': 'JULIO', 'august': 'AGOSTO', 'september': 'SEPTIEMBRE', 'october': 'OCTUBRE', 'november': 'NOVIEMBRE', 'december': 'DICIEMBRE', }, 'en': { 'phase1_title': 'DIESEL CFDI CONSUMPTION DATA', 'phase2_title': 'DIESEL FISCAL STIMULUS SUMMARY', 'summary_header': '================================================================================\n πŸ›’οΈ DIESEL CFDI CONSOLIDATION - SUMMARY\n================================================================================', 'no_items_found': 'No invoices with diesel items found. Exiting.', 'input_file': 'Input file', 'rows_scanned': 'Rows scanned', 'invoices_found': 'Invoices with diesel items', 'total_items': 'Total diesel line items', 'total_litros': 'Total Liters (Quantity)', 'subtotal_importe': 'Subtotal (Amount)', 'total_iva': 'IVA (16%)', 'total': 'TOTAL', 'output_files': 'Output files (to be written to', 'preview': 'Preview', 'first_invoices': 'first invoices', 'line_items': 'Line items', 'subtotal': 'Subtotal', 'files_written': 'Files written successfully', 'logs': 'Logs', 'proceed_prompt': 'Proceed with writing output files? [y/N]: ', 'aborted': 'Aborted.', 'phase1_headers': ['FECHA', 'FOLIO FISCAL', 'LITERS', 'RATE', 'AMOUNT'], 'phase2_headers': ['FECHA', 'FOLIO FISCAL', 'MONTH', 'LITERS', 'IEPS RATE (ART 2 FRACC I INCISO D LIEPS)', '% STIMULUS (ART FIRST)', 'FISCAL STIMULUS AMOUNT (ART SECOND)', 'REDUCED EFFECTIVE RATE (ART THIRD)', 'TOTAL FISCAL STIMULUS'], 'january': 'JANUARY', 'february': 'FEBRUARY', 'march': 'MARCH', 'april': 'APRIL', 'may': 'MAY', 'june': 'JUNE', 'july': 'JULY', 'august': 'AUGUST', 'september': 'SEPTEMBER', 'october': 'OCTOBER', 'november': 'NOVEMBER', 'december': 'DECEMBER', } } class DieselConsolidationProcessor: DIESEL_PRODUCT_CODE = "15101505" CURRENCY_FORMAT = '#,##0.00' IVA_RATE = 0.16 DEFAULT_CUOTA_IEPS = 7.0946 MONTH_NAMES_ES = { 1: 'ENERO', 2: 'FEBRERO', 3: 'MARZO', 4: 'ABRIL', 5: 'MAYO', 6: 'JUNIO', 7: 'JULIO', 8: 'AGOSTO', 9: 'SEPTIEMBRE', 10: 'OCTUBRE', 11: 'NOVIEMBRE', 12: 'DICIEMBRE' } MONTH_NAMES_EN = { 1: 'JANUARY', 2: 'FEBRUARY', 3: 'MARCH', 4: 'APRIL', 5: 'MAY', 6: 'JUNE', 7: 'JULY', 8: 'AUGUST', 9: 'SEPTEMBER', 10: 'OCTOBER', 11: 'NOVEMBER', 12: 'DECEMBER' } def __init__(self, logger: logging.Logger, language: str = 'es'): self.logger = logger self.invoices: List[Dict] = [] self.warnings: List[str] = [] self.rfc_receptor = None self.anio = None self.mes = None self.stimulus_schedule: List[Dict] = [] self.generate_csv = False self.language = language if language in TRANSLATIONS else 'es' self.t = TRANSLATIONS[self.language] @staticmethod def _safe_int(value) -> Optional[int]: """Parse value as int, returning None if null/blank/unparsable.""" if value is None or value == '': return None try: return int(value) except (ValueError, TypeError): return None def load_workbook(self, input_path: str) -> Tuple[object, List[str]]: """Load workbook and validate required columns.""" try: wb = load_workbook(input_path, data_only=True) ws = wb.active self.logger.info(f"Loaded workbook: {input_path}") except Exception as e: self.logger.error(f"Failed to load workbook: {e}") raise # Parse header row (first row) header = [cell.value for cell in ws[1]] self.logger.debug(f"Header row: {header}") # Resolve column indexes (case-insensitive) header_lower = [h.lower() if h else "" for h in header] required_cols = { 'fecha': None, 'uuid': None, 'conceptos': None, 'rfc receptor': None, 'anio': None, 'mes': None, } for idx, h_lower in enumerate(header_lower): for key in required_cols: if key in h_lower: required_cols[key] = idx break missing = [k for k, v in required_cols.items() if v is None] if missing: raise ValueError(f"Missing required columns: {', '.join(missing)}") self.logger.info(f"Column indexes resolved: {required_cols}") return wb, required_cols def extract_diesel_lines(self, conceptos_text: str, row_num: int) -> List[Dict]: """Extract diesel line items from Conceptos cell.""" if not conceptos_text or not isinstance(conceptos_text, str): return [] items = [] lines = conceptos_text.split('\n') for line_idx, line in enumerate(lines): line = line.strip() if not line: continue # Only process lines starting with the diesel product code if not line.startswith(f"ClaveProdServ: {self.DIESEL_PRODUCT_CODE}"): # Non-diesel lines are silently ignored continue # Parse the line pattern = ( r'ClaveProdServ:\s*15101505\s+' r'Cantidad:\s*([\d.]+)\s+' r'ValorUnitario:\s*([\d.]+)\s+' r'Importe:\s*([\d.]+)\s+' r'Descripcion:\s*(.+)$' ) match = re.match(pattern, line) if not match: msg = f"Row {row_num}, Line {line_idx + 1}: Malformed diesel line: {line[:100]}" self.logger.warning(msg) self.warnings.append(msg) continue try: items.append({ 'cantidad': float(match.group(1)), 'valor_unitario': float(match.group(2)), 'importe': float(match.group(3)), 'descripcion': match.group(4), }) except (ValueError, IndexError) as e: msg = f"Row {row_num}, Line {line_idx + 1}: Failed to parse values: {e}" self.logger.warning(msg) self.warnings.append(msg) continue return items def process_workbook(self, wb: object, col_map: Dict[str, int]): """Extract all invoices with diesel line items.""" ws = wb.active rows_scanned = 0 invoices_found = 0 # First pass: determine majority RFC/Anio/Mes across all rows, tolerating nulls. rfc_counter = Counter() anio_counter = Counter() mes_counter = Counter() for row in ws.iter_rows(min_row=2, values_only=False): rfc_receptor = row[col_map['rfc receptor']].value anio = self._safe_int(row[col_map['anio']].value) mes = self._safe_int(row[col_map['mes']].value) if rfc_receptor: rfc_counter[str(rfc_receptor).strip()] += 1 if anio is not None: anio_counter[anio] += 1 if mes is not None: mes_counter[mes] += 1 self.rfc_receptor = rfc_counter.most_common(1)[0][0] if rfc_counter else "UNKNOWN" self.anio = anio_counter.most_common(1)[0][0] if anio_counter else 0 self.mes = mes_counter.most_common(1)[0][0] if mes_counter else 0 if not anio_counter: self.logger.warning("Column 'anio' is null/unparsable for every row. Defaulting to 0.") if not mes_counter: self.logger.warning("Column 'mes' is null/unparsable for every row. Defaulting to 0.") self.logger.info(f"Filename metadata (majority vote): RFC={self.rfc_receptor}, Anio={self.anio}, Mes={self.mes}") # Second pass: extract diesel line items, warning on per-row deviations from the majority. for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=False), start=2): rows_scanned += 1 fecha = row[col_map['fecha']].value uuid = row[col_map['uuid']].value conceptos = row[col_map['conceptos']].value rfc_receptor = row[col_map['rfc receptor']].value anio = self._safe_int(row[col_map['anio']].value) mes = self._safe_int(row[col_map['mes']].value) rfc_value = str(rfc_receptor).strip() if rfc_receptor else None if (rfc_value is None or rfc_value != self.rfc_receptor or anio is None or anio != self.anio or mes is None or mes != self.mes): msg = ( f"Row {row_idx}: RFC/Anio/Mes differs from or is missing the majority value " f"(RFC: {rfc_receptor}, Anio: {anio}, Mes: {mes}). " f"Using majority values for output filename." ) self.logger.warning(msg) self.warnings.append(msg) items = self.extract_diesel_lines(conceptos, row_idx) if not items: self.logger.debug(f"Row {row_idx}: No diesel items found") continue # Calculate subtotals subtotal_litros = sum(item['cantidad'] for item in items) subtotal_importe = sum(item['importe'] for item in items) iva = round(subtotal_importe * self.IVA_RATE, 2) total = subtotal_importe + iva invoice = { 'fecha': fecha, 'uuid': uuid, 'items': items, 'subtotal_litros': subtotal_litros, 'subtotal_importe': subtotal_importe, 'iva': iva, 'total': total, } self.invoices.append(invoice) invoices_found += 1 self.logger.debug( f"Row {row_idx}: Extracted {len(items)} items, " f"subtotal_litros={subtotal_litros:.2f}, subtotal_importe={subtotal_importe:.2f}" ) self.logger.info(f"Processing complete: {rows_scanned} rows, {invoices_found} invoices with diesel items") return rows_scanned def get_totals(self) -> Dict: """Compute grand totals (litros/importe/iva/total) across all extracted invoices.""" grand_total_litros = sum(inv['subtotal_litros'] for inv in self.invoices) grand_total_importe = sum(inv['subtotal_importe'] for inv in self.invoices) grand_total_iva = sum(inv['iva'] for inv in self.invoices) return { 'litros': grand_total_litros, 'importe': grand_total_importe, 'iva': grand_total_iva, 'total': grand_total_importe + grand_total_iva, 'invoices_found': len(self.invoices), 'total_items': sum(len(inv['items']) for inv in self.invoices), } def print_summary(self, input_file: str, rows_scanned: int, output_dir: str): """Print a summary and preview of extracted data.""" if not self.invoices: print("\n" + "="*80) print(f"{Colors.RED}⚠️ RESUMEN - NO SE ENCONTRARON ITEMS DE DIESEL{Colors.RESET}" if self.language == 'es' else f"{Colors.RED}⚠️ SUMMARY - NO DIESEL ITEMS FOUND{Colors.RESET}") print("="*80) print(f"{self.t['input_file']}: {input_file}") print(f"{self.t['rows_scanned']}: {rows_scanned}") print(f"Resultado: {self.t['no_items_found']}") print("="*80 + "\n") return totals = self.get_totals() grand_total_litros = totals['litros'] grand_total_importe = totals['importe'] grand_total_iva = totals['iva'] grand_total = totals['total'] timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename_base = f"{self.rfc_receptor}_{self.anio}_{self.mes:02d}_{timestamp}" xlsx_file = f"{filename_base}.xlsx" csv_file = f"{filename_base}.csv" if self.generate_csv else None print("\n" + Colors.CYAN + "="*80) print(f"{self.t['summary_header']}") print("="*80 + Colors.RESET) print(f"{Colors.BLUE}πŸ“„ {self.t['input_file']}:{Colors.RESET} {input_file}") print(f"{Colors.BLUE}πŸ“Š {self.t['rows_scanned']}:{Colors.RESET} {rows_scanned}") print(f"{Colors.BLUE}πŸ“‹ {self.t['invoices_found']}:{Colors.RESET} {len(self.invoices)}") print(f"{Colors.BLUE}πŸ”’ {self.t['total_items']}:{Colors.RESET} {sum(len(inv['items']) for inv in self.invoices)}") print(f"\n{Colors.MAGENTA}πŸ’° {'Totales Generales' if self.language == 'es' else 'Grand Totals'}:{Colors.RESET}") print(f" {Colors.GREEN}βœ“{Colors.RESET} {self.t['total_litros']}: {Colors.BOLD}{grand_total_litros:>15.2f}{Colors.RESET} L") print(f" {Colors.GREEN}βœ“{Colors.RESET} {self.t['subtotal_importe']}: {Colors.BOLD}${grand_total_importe:>14,.2f}{Colors.RESET}") print(f" {Colors.GREEN}βœ“{Colors.RESET} {self.t['total_iva']}: {Colors.BOLD}${grand_total_iva:>14,.2f}{Colors.RESET}") print(f" {Colors.GREEN}βœ“{Colors.RESET} {self.t['total']}: {Colors.BOLD}${grand_total:>14,.2f}{Colors.RESET}") # Display loaded stimulus schedule (total count + latest 12 only, to avoid flooding the console) total_stimuli = len(self.stimulus_schedule) print(f"\n{Colors.YELLOW}πŸ“… {'EstΓ­mulos Cargados' if self.language == 'es' else 'Loaded Stimuli'} ({total_stimuli}):{Colors.RESET}") if self.stimulus_schedule: STIMULUS_PREVIEW_COUNT = 12 preview = self.stimulus_schedule[-STIMULUS_PREVIEW_COUNT:] start_idx = total_stimuli - len(preview) + 1 if total_stimuli > STIMULUS_PREVIEW_COUNT: omitted = total_stimuli - len(preview) omitted_text = (f"mostrando los ΓΊltimos {len(preview)}, {omitted} omitidos" if self.language == 'es' else f"showing latest {len(preview)}, {omitted} omitted") print(f" {Colors.CYAN}({omitted_text}){Colors.RESET}") for offset, period in enumerate(preview): idx = start_idx + offset start = period['start'] end = period['end'] pct = period['percent'] * 100 print(f" {Colors.BOLD}[{idx}]{Colors.RESET} {start} β†’ {end}: {Colors.YELLOW}{pct:.2f}%{Colors.RESET}") else: print(f" {Colors.RED}{'Ninguno' if self.language == 'es' else 'None'}{Colors.RESET}") print(f"\n{Colors.CYAN}πŸ“ {self.t['output_files']} {output_dir}/):{Colors.RESET}") print(f" {Colors.GREEN}↳{Colors.RESET} {xlsx_file}") if csv_file: print(f" {Colors.GREEN}↳{Colors.RESET} {csv_file}") print(f"\n{Colors.CYAN}πŸ‘οΈ {self.t['preview']} ({min(2, len(self.invoices))} {self.t['first_invoices']}):{Colors.RESET}") print("-" * 80) for inv_idx, inv in enumerate(self.invoices[:2], 1): print(f"\n{Colors.BOLD}{'Factura' if self.language == 'es' else 'Invoice'} {inv_idx}:{Colors.RESET} {inv['fecha']} | {Colors.MAGENTA}{inv['uuid']}{Colors.RESET}") print(f" {Colors.BLUE}{self.t['line_items']}:{Colors.RESET}") for item in inv['items'][:3]: # Show first 3 items print(f" {Colors.GREEN}{item['cantidad']:>10.2f}{Colors.RESET} L @ {Colors.YELLOW}{item['valor_unitario']:>10.6f}{Colors.RESET} = {Colors.BOLD}${item['importe']:>11,.2f}{Colors.RESET}") if len(inv['items']) > 3: more_text = f"mΓ‘s {len(inv['items']) - 3} items" if self.language == 'es' else f"+ {len(inv['items']) - 3} more items" print(f" {Colors.CYAN}... {more_text}{Colors.RESET}") print(f" {Colors.BOLD}{self.t['subtotal']}:{Colors.RESET} {inv['subtotal_litros']:.2f} L, ${inv['subtotal_importe']:,.2f}") iva_text = f"IVA (16%): ${inv['iva']:,.2f} | {self.t['total']}: ${inv['total']:,.2f}" print(f" {Colors.GREEN}{iva_text}{Colors.RESET}") print("\n" + "="*80) if self.warnings: print(f"\n{Colors.YELLOW}⚠️ Warnings ({len(self.warnings)}):{Colors.RESET}") for w in self.warnings[:5]: # Show first 5 warnings print(f" {Colors.YELLOW}⚠{Colors.RESET} {w}") if len(self.warnings) > 5: print(f" {Colors.YELLOW}...{Colors.RESET} and {len(self.warnings) - 5} more warnings (see log file for details)") print() def confirm_proceed(self, skip_confirmation: bool) -> bool: """Ask user for confirmation to proceed.""" if skip_confirmation: self.logger.info("Skipping confirmation (--yes flag)") return True yes_options = ('s', 'si', 'sΓ­') if self.language == 'es' else ('y', 'yes') no_options = ('n', 'no', '') while True: response = input(self.t['proceed_prompt']).strip().lower() if response in yes_options: return True elif response in no_options: return False def load_stimulus_schedule(self, file_path: str) -> List[Dict]: """Load stimulus schedule from XLSX or CSV file.""" path = Path(file_path) # Try XLSX first xlsx_path = path.with_suffix('.xlsx') if xlsx_path.exists(): return self._load_stimulus_from_xlsx(str(xlsx_path)) # Fall back to CSV if path.exists(): return self._load_stimulus_from_csv(str(path)) self.logger.warning(f"Stimulus schedule not found: {file_path}. All stimuli will be 0%") return [] def _load_stimulus_from_xlsx(self, xlsx_path: str) -> List[Dict]: """Load stimulus schedule from XLSX file.""" schedule = [] try: from openpyxl import load_workbook self.logger.debug(f"Opening XLSX file: {xlsx_path}") wb = load_workbook(xlsx_path, data_only=True) ws = wb.active self.logger.debug(f"Sheet name: {ws.title}") # Skip header row, read data for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), 1): try: if not row or len(row) < 3: self.logger.debug(f"Stimulus row {row_idx}: Skipped (empty or incomplete)") continue start_val = row[0] end_val = row[1] percent_val = row[2] self.logger.debug(f"Stimulus row {row_idx}: start_val={start_val}, end_val={end_val}, percent_val={percent_val}") if not start_val or not end_val or percent_val is None: self.logger.warning(f"Skipping incomplete stimulus row {row_idx}") continue # Handle date values (can be date objects or strings) if isinstance(start_val, str): start_date = datetime.strptime(start_val.strip(), '%Y-%m-%d').date() else: start_date = start_val.date() if hasattr(start_val, 'date') else start_val if isinstance(end_val, str): end_date = datetime.strptime(end_val.strip(), '%Y-%m-%d').date() else: end_date = end_val.date() if hasattr(end_val, 'date') else end_val percent = float(percent_val) / 100.0 schedule.append({ 'start': start_date, 'end': end_date, 'percent': percent, }) self.logger.debug(f"Stimulus row {row_idx}: Loaded successfully - {start_date} to {end_date}: {percent*100:.2f}%") except (ValueError, KeyError, TypeError, AttributeError) as e: self.logger.warning(f"Skipping malformed stimulus row {row_idx}: {e}") self.logger.debug(f"Row data: {row}") continue self.logger.info(f"Loaded {len(schedule)} stimulus periods from {xlsx_path}") for idx, period in enumerate(schedule): self.logger.debug(f"Stimulus period {idx}: {period['start']} to {period['end']}: {period['percent']*100:.2f}%") return schedule except Exception as e: self.logger.error(f"Failed to load stimulus schedule from XLSX: {e}") self.logger.debug(f"Exception details:", exc_info=True) return [] def _load_stimulus_from_csv(self, csv_path: str) -> List[Dict]: """Load stimulus schedule from CSV file.""" schedule = [] try: self.logger.debug(f"Opening CSV file: {csv_path}") with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row_idx, row in enumerate(reader, 1): try: start_str = str(row.get('fecha_inicio', '')).strip() end_str = str(row.get('fecha_fin', '')).strip() percent_str = str(row.get('porcentaje', '')).strip() self.logger.debug(f"Stimulus row {row_idx}: start_str={start_str}, end_str={end_str}, percent_str={percent_str}") if not start_str or not end_str or not percent_str: self.logger.warning(f"Skipping incomplete stimulus row {row_idx}") continue start_date = datetime.strptime(start_str, '%Y-%m-%d').date() end_date = datetime.strptime(end_str, '%Y-%m-%d').date() percent = float(percent_str) / 100.0 schedule.append({ 'start': start_date, 'end': end_date, 'percent': percent, }) self.logger.debug(f"Stimulus row {row_idx}: Loaded successfully - {start_date} to {end_date}: {percent*100:.2f}%") except (ValueError, KeyError, TypeError) as e: self.logger.warning(f"Skipping malformed stimulus row {row_idx}: {e}") continue self.logger.info(f"Loaded {len(schedule)} stimulus periods from {csv_path}") for idx, period in enumerate(schedule): self.logger.debug(f"Stimulus period {idx}: {period['start']} to {period['end']}: {period['percent']*100:.2f}%") return schedule except Exception as e: self.logger.error(f"Failed to load stimulus schedule from CSV: {e}") self.logger.debug(f"Exception details:", exc_info=True) return [] def get_stimulus_percentage(self, fecha: datetime, schedule: List[Dict]) -> float: """Get stimulus percentage for a given date.""" if not fecha: return 0.0 # Extract date part only - handle multiple formats fecha_date = None if isinstance(fecha, datetime): fecha_date = fecha.date() elif isinstance(fecha, date): fecha_date = fecha elif isinstance(fecha, str): # Handle ISO format strings (e.g., "2026-07-31T14:57:07" or "2026-07-31") try: # Try ISO format with time if 'T' in fecha: fecha_date = datetime.fromisoformat(fecha).date() else: # Try simple date format fecha_date = datetime.strptime(fecha, '%Y-%m-%d').date() except (ValueError, TypeError): self.logger.warning(f"Could not parse fecha as date: {fecha}") return 0.0 else: self.logger.warning(f"Unexpected fecha type: {type(fecha)}") return 0.0 if not fecha_date: return 0.0 for period in schedule: if not period or 'start' not in period or 'end' not in period: continue try: start_d = period['start'] if isinstance(period['start'], date) else datetime.strptime(str(period['start']), '%Y-%m-%d').date() end_d = period['end'] if isinstance(period['end'], date) else datetime.strptime(str(period['end']), '%Y-%m-%d').date() if start_d <= fecha_date <= end_d: self.logger.debug(f"Date match: {fecha_date} in range {start_d}β†’{end_d}, stimulus={period.get('percent', 0.0)*100:.2f}%") return period.get('percent', 0.0) except (ValueError, TypeError) as e: self.logger.warning(f"Error comparing dates in stimulus period: {e}") continue self.logger.debug(f"No stimulus period matched for date: {fecha_date}") return 0.0 def generate_summary_sheet(self, wb: object, cuota_ieps: float): """Generate summary sheet with stimulus calculations.""" if not self.invoices: self.logger.info("No invoices to summarize") return # Group by FOLIO (UUID) folio_groups = {} for inv in self.invoices: uuid = inv['uuid'] if uuid not in folio_groups: folio_groups[uuid] = { 'fecha': inv['fecha'], 'litros': 0.0, } folio_groups[uuid]['litros'] += inv['subtotal_litros'] # Create summary sheet ws = wb.create_sheet('Resumen') # Title title_row = 1 ws[f'A{title_row}'] = self.t['phase2_title'] ws[f'A{title_row}'].font = Font(bold=True, size=14) # Header row header_row = 3 headers = self.t['phase2_headers'] for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=header_row, column=col_idx) cell.value = header cell.font = Font(bold=True, size=10) cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) cell.fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid") # Set column widths ws.column_dimensions['A'].width = 20 ws.column_dimensions['B'].width = 38 ws.column_dimensions['C'].width = 12 ws.column_dimensions['D'].width = 12 ws.column_dimensions['E'].width = 16 ws.column_dimensions['F'].width = 16 ws.column_dimensions['G'].width = 16 ws.column_dimensions['H'].width = 16 ws.column_dimensions['I'].width = 18 current_row = header_row + 1 total_litros = 0.0 total_monto = 0.0 self.logger.debug(f"Stimulus schedule loaded: {len(self.stimulus_schedule)} periods") # Write rows for each FOLIO for uuid in sorted(folio_groups.keys()): group = folio_groups[uuid] fecha = group['fecha'] litros = group['litros'] # Extract month month_names = self.MONTH_NAMES_ES if self.language == 'es' else self.MONTH_NAMES_EN if isinstance(fecha, datetime): mes_num = fecha.month mes_name = month_names.get(mes_num, 'UNKNOWN') elif isinstance(fecha, str): try: fecha_obj = datetime.strptime(str(fecha), '%Y-%m-%d %H:%M:%S') mes_num = fecha_obj.month mes_name = month_names.get(mes_num, 'UNKNOWN') except: mes_name = 'UNKNOWN' else: mes_name = 'UNKNOWN' # Get stimulus percentage (parse fecha if it's a string) if isinstance(fecha, str): try: fecha_obj = datetime.strptime(fecha, '%Y-%m-%d %H:%M:%S') except: fecha_obj = fecha else: fecha_obj = fecha stimulus_pct = self.get_stimulus_percentage(fecha_obj, self.stimulus_schedule) # Calculate amounts monto_estimulo = cuota_ieps * stimulus_pct # Per liter cuota_disminuida = cuota_ieps * (1.0 - stimulus_pct) # Per liter monto_total = litros * cuota_ieps * (1.0 - stimulus_pct) # Total # Write row ws.cell(row=current_row, column=1).value = fecha ws.cell(row=current_row, column=2).value = uuid ws.cell(row=current_row, column=3).value = mes_name ws.cell(row=current_row, column=4).value = litros ws.cell(row=current_row, column=5).value = cuota_ieps ws.cell(row=current_row, column=6).value = stimulus_pct ws.cell(row=current_row, column=7).value = monto_estimulo ws.cell(row=current_row, column=8).value = cuota_disminuida ws.cell(row=current_row, column=9).value = monto_total # Format numbers ws.cell(row=current_row, column=4).number_format = '#,##0.00' ws.cell(row=current_row, column=5).number_format = '0.0000' ws.cell(row=current_row, column=6).number_format = '0.00%' ws.cell(row=current_row, column=7).number_format = '0.0000' ws.cell(row=current_row, column=8).number_format = '0.0000' ws.cell(row=current_row, column=9).number_format = '#,##0.00' # Debug logging self.logger.debug(f"Resumen row {current_row - header_row}: UUID={uuid}, Fecha={fecha}, Litros={litros:.2f}, " f"Stimulus%={stimulus_pct*100:.2f}%, Monto=${monto_total:,.2f}") total_litros += litros total_monto += monto_total current_row += 1 # TOTAL row ws.cell(row=current_row, column=1).value = "TOTAL" ws.cell(row=current_row, column=1).font = Font(bold=True) ws.cell(row=current_row, column=4).value = total_litros ws.cell(row=current_row, column=4).font = Font(bold=True) ws.cell(row=current_row, column=4).number_format = '#,##0.00' ws.cell(row=current_row, column=9).value = total_monto ws.cell(row=current_row, column=9).font = Font(bold=True) ws.cell(row=current_row, column=9).number_format = '#,##0.00' ws.cell(row=current_row, column=9).fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid") self.logger.info(f"Summary sheet generated with {current_row - header_row - 1} invoices") def write_xlsx(self, output_path: Path, cuota_ieps: float = None): """Write consolidated data to XLSX file.""" if cuota_ieps is None: cuota_ieps = self.DEFAULT_CUOTA_IEPS wb = Workbook() ws = wb.active ws.title = "Diesel Consolidation" # Title row title_row = 1 ws[f'A{title_row}'] = self.t['phase1_title'] ws[f'A{title_row}'].font = Font(bold=True, size=14) # Header row header_row = 3 headers = self.t['phase1_headers'] for col_idx, header in enumerate(headers, 1): cell = ws.cell(row=header_row, column=col_idx) cell.value = header cell.font = Font(bold=True) cell.alignment = Alignment(horizontal='center', vertical='center') cell.fill = PatternFill(start_color="D3D3D3", end_color="D3D3D3", fill_type="solid") # Set column widths ws.column_dimensions['A'].width = 20 ws.column_dimensions['B'].width = 38 ws.column_dimensions['C'].width = 12 ws.column_dimensions['D'].width = 14 ws.column_dimensions['E'].width = 14 current_row = header_row + 1 # Write invoices for inv in self.invoices: fecha_value = inv['fecha'] uuid_value = inv['uuid'] for item_idx, item in enumerate(inv['items']): # First item gets fecha and uuid if item_idx == 0: ws.cell(row=current_row, column=1).value = fecha_value ws.cell(row=current_row, column=2).value = uuid_value # Subsequent items have blank fecha and uuid ws.cell(row=current_row, column=3).value = item['cantidad'] ws.cell(row=current_row, column=4).value = item['valor_unitario'] ws.cell(row=current_row, column=5).value = item['importe'] # Format numbers ws.cell(row=current_row, column=3).number_format = '0.00' ws.cell(row=current_row, column=4).number_format = '0.000000' ws.cell(row=current_row, column=5).number_format = self.CURRENCY_FORMAT current_row += 1 # SUBTOTAL row ws.cell(row=current_row, column=1).value = "SUBTOTAL" ws.cell(row=current_row, column=1).font = Font(bold=True) ws.cell(row=current_row, column=3).value = inv['subtotal_litros'] ws.cell(row=current_row, column=3).number_format = '0.00' ws.cell(row=current_row, column=3).font = Font(bold=True) ws.cell(row=current_row, column=5).value = inv['subtotal_importe'] ws.cell(row=current_row, column=5).number_format = self.CURRENCY_FORMAT ws.cell(row=current_row, column=5).font = Font(bold=True) current_row += 1 # IVA row ws.cell(row=current_row, column=1).value = "IVA" ws.cell(row=current_row, column=4).value = self.IVA_RATE ws.cell(row=current_row, column=4).number_format = '0.00%' ws.cell(row=current_row, column=5).value = inv['iva'] ws.cell(row=current_row, column=5).number_format = self.CURRENCY_FORMAT ws.cell(row=current_row, column=5).font = Font(bold=True) current_row += 1 # TOTAL row ws.cell(row=current_row, column=1).value = "TOTAL" ws.cell(row=current_row, column=1).font = Font(bold=True) ws.cell(row=current_row, column=5).value = inv['total'] ws.cell(row=current_row, column=5).number_format = self.CURRENCY_FORMAT ws.cell(row=current_row, column=5).fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid") ws.cell(row=current_row, column=5).font = Font(bold=True) current_row += 1 # Generate summary sheet (phase 2) self.generate_summary_sheet(wb, cuota_ieps) wb.save(str(output_path)) self.logger.info(f"XLSX file written: {output_path}") def write_csv(self, output_path: Path): """Write consolidated data to CSV file.""" with open(output_path, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['FECHA', 'FOLIO FISCAL', 'LITROS', 'CUOTA', 'IMPORTE']) for inv in self.invoices: for item_idx, item in enumerate(inv['items']): writer.writerow([ inv['fecha'], inv['uuid'], f"{item['cantidad']:.2f}", f"{item['valor_unitario']:.6f}", f"{item['importe']:.2f}", ]) # SUBTOTAL row writer.writerow(['SUBTOTAL', '', f"{inv['subtotal_litros']:.2f}", '', f"{inv['subtotal_importe']:.2f}"]) # IVA row writer.writerow(['IVA', '', '', f"{self.IVA_RATE:.2f}", f"{inv['iva']:.2f}"]) # TOTAL row writer.writerow(['TOTAL', '', '', '', f"{inv['total']:.2f}"]) self.logger.info(f"CSV file written: {output_path}") def setup_logging(output_dir: Path, log_level: str) -> logging.Logger: """Configure logging with console, per-run file, and persistent history file handlers.""" logger = logging.getLogger('diesel_consolidation') logger.setLevel(getattr(logging, log_level.upper(), logging.INFO)) # Console handler console_handler = logging.StreamHandler() console_handler.setLevel(getattr(logging, log_level.upper(), logging.INFO)) console_formatter = logging.Formatter('%(levelname)-8s | %(message)s') console_handler.setFormatter(console_formatter) logger.addHandler(console_handler) output_dir.mkdir(parents=True, exist_ok=True) logs_dir = output_dir / 'logs' logs_dir.mkdir(exist_ok=True) file_formatter = logging.Formatter( '%(asctime)s | %(levelname)-8s | %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # Per-run file handler (one file per execution) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") log_file = logs_dir / f"diesel_{timestamp}.log" file_handler = logging.handlers.RotatingFileHandler( log_file, maxBytes=10 * 1024 * 1024, # 10 MB backupCount=5 ) file_handler.setLevel(logging.DEBUG) # Always log DEBUG to file file_handler.setFormatter(file_formatter) logger.addHandler(file_handler) # Persistent history file handler (appends across every execution) history_file = logs_dir / "diesel_history.log" history_handler = logging.handlers.RotatingFileHandler( history_file, maxBytes=50 * 1024 * 1024, # 50 MB backupCount=5, mode='a' ) history_handler.setLevel(logging.DEBUG) history_handler.setFormatter(file_formatter) separator = ( "\n" + "=" * 80 + f"\nEXECUTION START: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | " f"CLI args: {' '.join(sys.argv[1:])}\n" + "=" * 80 ) with open(history_file, 'a', encoding='utf-8') as f: f.write(separator + "\n") logger.addHandler(history_handler) logger.info(f"Logging initialized (per-run log: {log_file.name}, history log: {history_file.name})") return logger def main(): parser = argparse.ArgumentParser( description="Extract and consolidate diesel line items from CFDI XLSX exports" ) parser.add_argument( '--input-file', required=True, help='Path to the CFDI XLSX file' ) parser.add_argument( '--output', default='./output', help='Output directory (default: ./output)' ) parser.add_argument( '-y', '--yes', action='store_true', help='Skip confirmation prompt' ) parser.add_argument( '--generate-csv', action='store_true', default=False, help='Also generate the flat CSV output file (default: False, only XLSX is generated)' ) parser.add_argument( '--log-level', default='INFO', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], help='Logging level (default: INFO)' ) parser.add_argument( '--cuota-ieps', type=float, default=None, help='Base IEPS tax rate per liter (default: 7.0946, can be overridden by DIESEL_COUTA_IEPS env var)' ) parser.add_argument( '--lang', choices=['es', 'en'], default='es', help='Language for output (es: Spanish, en: English, default: es)' ) args = parser.parse_args() # Setup logging output_dir = Path(args.output) logger = setup_logging(output_dir, args.log_level) logger.info(f"Starting diesel consolidation") logger.info(f"Input file: {args.input_file}") logger.info(f"Output directory: {args.output}") # Resolve cuota-ieps (env var < CLI flag) cuota_ieps = DieselConsolidationProcessor.DEFAULT_CUOTA_IEPS if 'DIESEL_COUTA_IEPS' in os.environ: try: cuota_ieps = float(os.environ['DIESEL_COUTA_IEPS']) logger.info(f"Using cuota_ieps from environment: {cuota_ieps}") except ValueError: logger.warning(f"Invalid DIESEL_COUTA_IEPS env var, using default") if args.cuota_ieps is not None: cuota_ieps = args.cuota_ieps logger.info(f"Using cuota_ieps from CLI argument: {cuota_ieps}") try: # Validate input file input_path = Path(args.input_file) if not input_path.exists(): logger.error(f"Input file not found: {input_path}") print(f"ERROR: Input file not found: {input_path}") return 1 # Process processor = DieselConsolidationProcessor(logger, args.lang) processor.generate_csv = args.generate_csv # Load stimulus schedule for phase 2 stimulus_csv = get_base_dir() / 'estimulos.csv' processor.stimulus_schedule = processor.load_stimulus_schedule(str(stimulus_csv)) wb, col_map = processor.load_workbook(str(input_path)) rows_scanned = processor.process_workbook(wb, col_map) # Summary and confirmation processor.print_summary(str(input_path), rows_scanned, args.output) if not processor.invoices: logger.info(processor.t['no_items_found']) print(processor.t['no_items_found']) return 0 if not processor.confirm_proceed(args.yes): logger.info("User declined to proceed") print(f"\n{Colors.YELLOW}⏸️ {processor.t['aborted']}{Colors.RESET}\n") return 1 # Generate output files output_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename_base = f"{processor.rfc_receptor}_{processor.anio}_{processor.mes:02d}_{timestamp}" xlsx_path = output_dir / f"{filename_base}.xlsx" processor.write_xlsx(xlsx_path, cuota_ieps) print(f"\n{Colors.GREEN}{Colors.BOLD}βœ… {processor.t['files_written']}:{Colors.RESET}") print(f" {Colors.GREEN}↳{Colors.RESET} {xlsx_path}") if args.generate_csv: csv_path = output_dir / f"{filename_base}.csv" processor.write_csv(csv_path) print(f" {Colors.GREEN}↳{Colors.RESET} {csv_path}") print(f"\n{Colors.BLUE}πŸ“‹ {processor.t['logs']}:{Colors.RESET} {output_dir / 'logs'}") logger.info("Consolidation completed successfully") return 0 except Exception as e: logger.exception(f"Unexpected error: {e}") print(f"\n{Colors.RED}{Colors.BOLD}❌ ERROR: {e}{Colors.RESET}") print(f"{Colors.YELLOW}πŸ“‹ See log file for details: {output_dir / 'logs'}{Colors.RESET}\n") return 2 if __name__ == '__main__': sys.exit(main())