#helpers/csv_mapper from typing import List, Dict, Any, Optional from app.modules.edos.schema import EdosUpload from datetime import date, datetime class CSVMapper: """ Mapper: filter and mapping only necesary data, based on a schema""" FIELD_MAPPING = { 'RFC': 'numero', 'Nombre del Contribuyente': 'razon_social', 'Situación del contribuyente': 'situacion', 'Número y fecha de oficio global definitivo SAT': 'numero_definitivo', 'Publicación página SAT definitivo': 'fecha_definitivo', 'Número y fecha de oficio global definitivo DOF': 'numero_def_dof', 'Publicación DOF definitivo': 'fecha_def_dof', 'Número y fecha de oficio global de sentencia favorable SAT': 'numero_fav_sat', 'Publicación página SAT sentencia favorable': 'publicacion_sat', 'Publicación DOF sentencia favorable': 'publicacion_dof', } REQUIRED_FIELD = [ 'numero', 'razon_social', 'situacion'] #Campos obligatorios @staticmethod def map_rows(csv_rows: List[Dict]) -> List[Dict]: mapped_rows = [] for row in csv_rows: mapped = {} for csv_field, bd_field in CSVMapper.FIELD_MAPPING.items(): value = row.get(csv_field, '') if bd_field == 'razon_social': value = value.replace('\n', ' ').replace('\r', ' ') if bd_field == 'situacion': value = CSVMapper._normalize_situacion(value) if 'fecha' in bd_field and value: value = CSVMapper._parse_date(value) mapped[bd_field] = value if value else None missing = [f for f in CSVMapper.REQUIRED_FIELD if not mapped.get(f)] if missing: print(f"Fila omitida - faltan: {missing}") continue mapped_rows.append(mapped) return mapped_rows @staticmethod def _normalize_situacion(value: str) -> str: """ converts text of CVS to format ENUM""" if not value: return None if 'Sentencia Favorable' in value: return 'sentencia_favorable' if 'Definitivo' in value: return 'definitivo' return value.lower() @staticmethod def _parse_date(value: str) -> Optional[date]: """Converts DD/MM/YYYY to date""" if not value: return None try: return datetime.strptime(value.strip(), '%dd%m%Y').date() except: return None