blob: b9dccdb4d4f6c7b600b84694b8502268cf55a0a7 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
import csv
import io
from typing import List, Dict
from obfuscator.logger import get_logger
logger = get_logger("CSVReader")
class CSVReader:
@staticmethod
def read_local(path) -> List[Dict[str, str]]:
logger.debug(f"Reading local CSV from: {path}")
try:
with open(path, mode="r", encoding="utf-8") as f:
reader = csv.DictReader(f)
return [dict(row) for row in reader]
except FileNotFoundError:
logger.error(f"File not found: {path}")
except Exception as e:
logger.error(f"Error reading file: {e}")
@staticmethod
def read_s3(path) -> List[Dict[str, str]]:
return []
@staticmethod
def read_string(content: str) -> List[Dict[str, str]]:
if not content.strip():
return []
f = io.StringIO(content)
reader = csv.DictReader(f)
return [dict(row) for row in reader]
|
