75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
#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
|