41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
import csv
|
|
from io import TextIOWrapper
|
|
from config import item_attributes
|
|
|
|
def get_csv_data(file):
|
|
"""Extract and validate CSV data."""
|
|
try:
|
|
csv_file = TextIOWrapper(file, encoding='utf-8')
|
|
|
|
# Check delimiter
|
|
sample = csv_file.read(1024)
|
|
csv_file.seek(0) # Reset the file pointer
|
|
if '|' not in sample:
|
|
raise ValueError("CSV file must use '|' as the delimiter.")
|
|
|
|
reader = csv.DictReader(csv_file, delimiter='|')
|
|
|
|
# Validate CSV headers based on config
|
|
required_headers = {attrib.display_name for attrib in item_attributes} # Use display names
|
|
csv_headers = set(reader.fieldnames) # Get headers present in the CSV file
|
|
|
|
# Check if the required headers exist
|
|
if not required_headers.issubset(csv_headers):
|
|
raise ValueError(f"CSV file must include headers: {', '.join(required_headers)}.")
|
|
|
|
# Check for invalid headers
|
|
if extra_headers := csv_headers - required_headers:
|
|
raise ValueError(f"Unexpected headers found: {', '.join(extra_headers)}")
|
|
|
|
# Map display names to attribute names
|
|
display_to_attrib = {attrib.display_name: attrib.attrib_name for attrib in item_attributes}
|
|
|
|
# Extract data dynamically based on config
|
|
return [
|
|
{display_to_attrib[header]: row[header] for header in required_headers}
|
|
for row in reader
|
|
]
|
|
except Exception as e:
|
|
# Log the error and re-raise it
|
|
print(f"Error processing CSV file: {str(e)}")
|
|
raise |