#helper/csv_mapper import csv import requests from io import StringIO # from typing import List, Dict, Any import os from dotenv import load_dotenv # import pandas as pd import math load_dotenv() class Settings: """configuration""" CSV_DOWNLOAD_TIMEOUT = int(os.getenv('CSV_DOWNLOAD_TIMEOUT', 30)) CSV_MAX_SIZE_MB = int(os.getenv('CSV_MAX_SIZE_MB', 10)) settings = Settings() def _find_header_row(self, rows, delimiter): """ Search row with RFC an name""" for idx, row in enumerate(rows): #Normalice: Convert to String, clean space, mayus clean_row = [str(cell).strip().upper() for cell in row] has_rfc = any('RFC' in cell for cell in clean_row) has_nombre =any('NOMBRE' in cell for cell in clean_row) if has_rfc and has_nombre: return idx return None class CSVExtractor: """Extract raw data of CSV""" @staticmethod def download_csv(url: str) -> str: """ Download CSV of URL""" try: response = requests.get(url, timeout=30) response.raise_for_status() encoding = response.encoding or 'utf-8' content = response.text return content, encoding except Exception as e: raise Exception(f"Error Download CSV: {str(e)}") @staticmethod def read_csv(content): """ Converts CSV to list of dictionary""" for delimiter in[',', ';', '\t']: try: lines = content.split('\n') header_row_idx = None for idx, line in enumerate(lines): if 'RFC' in line.upper(): header_row_idx = idx break if header_row_idx is None: continue df = pd.read_csv( StringIO(content), delimiter=delimiter, encoding='utf-8', skiprows=header_row_idx, dtype=str, keep_default_na=False, na_filter=False) df.columns = df.columns.str.strip() df = df.dropna(how='all') result = df.to_dict('records') if result: return result except Exception as e: print(f"Error with delimiter '{delimiter}': {e}") continue raise ValueError("Don't read CSV with pandas")