|
|
import csv |
|
|
import io |
|
|
from pathlib import Path |
|
|
|
|
|
from langflow.custom import Component |
|
|
from langflow.io import FileInput, MessageTextInput, MultilineInput, Output |
|
|
from langflow.schema import Data |
|
|
|
|
|
|
|
|
class CSVToDataComponent(Component): |
|
|
display_name = "Load CSV" |
|
|
description = "Load a CSV file, CSV from a file path, or a valid CSV string and convert it to a list of Data" |
|
|
icon = "file-spreadsheet" |
|
|
name = "CSVtoData" |
|
|
legacy = True |
|
|
|
|
|
inputs = [ |
|
|
FileInput( |
|
|
name="csv_file", |
|
|
display_name="CSV File", |
|
|
file_types=["csv"], |
|
|
info="Upload a CSV file to convert to a list of Data objects", |
|
|
), |
|
|
MessageTextInput( |
|
|
name="csv_path", |
|
|
display_name="CSV File Path", |
|
|
info="Provide the path to the CSV file as pure text", |
|
|
), |
|
|
MultilineInput( |
|
|
name="csv_string", |
|
|
display_name="CSV String", |
|
|
info="Paste a CSV string directly to convert to a list of Data objects", |
|
|
), |
|
|
] |
|
|
|
|
|
outputs = [ |
|
|
Output(name="data_list", display_name="Data List", method="load_csv_to_data"), |
|
|
] |
|
|
|
|
|
def load_csv_to_data(self) -> list[Data]: |
|
|
if sum(bool(field) for field in [self.csv_file, self.csv_path, self.csv_string]) != 1: |
|
|
msg = "Please provide exactly one of: CSV file, file path, or CSV string." |
|
|
raise ValueError(msg) |
|
|
|
|
|
csv_data = None |
|
|
try: |
|
|
if self.csv_file: |
|
|
resolved_path = self.resolve_path(self.csv_file) |
|
|
file_path = Path(resolved_path) |
|
|
if file_path.suffix.lower() != ".csv": |
|
|
self.status = "The provided file must be a CSV file." |
|
|
else: |
|
|
with file_path.open(newline="", encoding="utf-8") as csvfile: |
|
|
csv_data = csvfile.read() |
|
|
|
|
|
elif self.csv_path: |
|
|
file_path = Path(self.csv_path) |
|
|
if file_path.suffix.lower() != ".csv": |
|
|
self.status = "The provided file must be a CSV file." |
|
|
else: |
|
|
with file_path.open(newline="", encoding="utf-8") as csvfile: |
|
|
csv_data = csvfile.read() |
|
|
|
|
|
else: |
|
|
csv_data = self.csv_string |
|
|
|
|
|
if csv_data: |
|
|
csv_reader = csv.DictReader(io.StringIO(csv_data)) |
|
|
result = [Data(data=row) for row in csv_reader] |
|
|
|
|
|
if not result: |
|
|
self.status = "The CSV data is empty." |
|
|
return [] |
|
|
|
|
|
self.status = result |
|
|
return result |
|
|
|
|
|
except csv.Error as e: |
|
|
error_message = f"CSV parsing error: {e}" |
|
|
self.status = error_message |
|
|
raise ValueError(error_message) from e |
|
|
|
|
|
except Exception as e: |
|
|
error_message = f"An error occurred: {e}" |
|
|
self.status = error_message |
|
|
raise ValueError(error_message) from e |
|
|
|
|
|
|
|
|
raise ValueError(self.status) |
|
|
|
