import csv import requests from io import StringIO # from typing import List, Dict, Any import os from dotenv import load_dotenv 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() 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: str) -> List[Dict[str, Any]]: """ Converts CSV to list of dictionary""" csv_file = StringIO(content) #detect delimiter auto sample = csv_file.read(1024) csv_file.seek(0) try: dialect = csv.Sniffer().sniff(sample) reader =csv.DictReader(csv_file, dialect=dialect) except: csv_file.seek(0) reader = csv.DictReader(csv_file) rows = [row for row in reader] if rows: clean_row = {} for key, value in rows[0].items(): if key is not None: clean_key = key.strip().replace('"', '').replace("'", "").replace('\ufeff', '') else: clean_key = '' clean_row[clean_key] = value for i, row in enumerate(rows): new_row = {} for old_key, value in row.items(): if old_key is not None: new_key = old_key.strip().replace('"', '').replace("'", "").replace('\ufeff', '') else: new_key = '' #FIX: manage value None for avoid error strip if value is not None and isinstance(value, str): new_value = value.strip() else: new_value = value new_row[new_key] = value rows[i] = new_row return rows