96 lines
2.5 KiB
Python
96 lines
2.5 KiB
Python
#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")
|
|
|