feat: support csv preprocessing

Author: PaulKalho

cbc.yaml

@@ -89,4 +89,50 @@ entrypoints:
           normalized_overwritten_file_output_S3_SECRET_KEY: null
         description: The File Input Overwritten with the normalized output
         type: file
+  preprocess_csv_file:
+    description: Entrypoint to preprocess a column of a .csv file
+    envs:
+      FILTER_STOPWORDS: true
+      LANGUAGE: en
+      NGRAM_MAX: 3
+      NGRAM_MIN: 2
+      TXT_DOWNLOAD_PATH: /tmp/input.txt
+      UNIGRAM_NORMALIZER: lemma
+      USE_NGRAMS: true
+    inputs:
+      csv_input:
+        config:
+          csv_file_BUCKET_NAME: null
+          csv_file_FILE_EXT: txt
+          csv_file_FILE_NAME: null
+          csv_file_FILE_PATH: null
+          csv_file_S3_ACCESS_KEY: null
+          csv_file_S3_HOST: null
+          csv_file_S3_PORT: null
+          csv_file_S3_SECRET_KEY: null
+          csv_file_SELECTED_ATTRIBUTE: abstract
+          csv_file_ID_COLUMN: id
+        description: A .txt file, each line will be treated as a document
+        type: file
+    outputs:
+      normalized_docs_output:
+        config:
+          normalized_docs_DB_TABLE: null
+          normalized_docs_DB_DSN: null
+          normalized_docs_DB_SCHEMA: null
+        description: Database Output, containing bib_id aswell as the normalized text
+        type: database_table
+      normalized_overwritten_file_output:
+        config:
+          normalized_overwritten_file_output_BUCKET_NAME: null
+          normalized_overwritten_file_output_FILE_EXT: txt
+          normalized_overwritten_file_output_FILE_NAME: null
+          normalized_overwritten_file_output_FILE_PATH: null
+          normalized_overwritten_file_output_S3_ACCESS_KEY: null
+          normalized_overwritten_file_output_S3_HOST: null
+          normalized_overwritten_file_output_S3_PORT: null
+          normalized_overwritten_file_output_S3_SECRET_KEY: null
+        description: The File Input Overwritten with the normalized output
+        type: file
+
 name: Language-Preprocessing

main.py

@@ -17,7 +17,7 @@
 )
 
 from preprocessing.core import Preprocessor
-from preprocessing.loader import TxtLoader, BibLoader
+from preprocessing.loader import CSVLoader, TxtLoader, BibLoader
 from preprocessing.models import DocumentRecord, PreprocessedDocument
 
 logging.basicConfig(
@@ -48,11 +48,24 @@ class BIBFileInput(FileSettings, InputSettings):
     SELECTED_ATTRIBUTE: str = "Abstract"
 
 
+class CSVFileInput(FileSettings, InputSettings):
+    __identifier__ = "csv_file"
+    FILE_EXT: str = "csv"
+
+    SELECTED_ATTRIBUTE: str = "abstract"
+    ID_COLUMN: str = "id"
+
+
 class NormalizedBIBOutput(FileSettings, OutputSettings):
     __identifier__ = "normalized_overwritten_file_output"
     FILE_EXT: str = "bib"
 
 
+class NormalizedCSVOutput(FileSettings, OutputSettings):
+    __identifier__ = "normalized_overwritten_file_output"
+    FILE_EXT: str = "csv"
+
+
 class PreprocessTXT(EnvSettings):
     LANGUAGE: str = "en"
     FILTER_STOPWORDS: bool = True
@@ -83,6 +96,21 @@ class PreprocessBIB(EnvSettings):
     normalized_overwritten_file_output: NormalizedBIBOutput
 
 
+class PreprocessCSV(EnvSettings):
+    LANGUAGE: str = "en"
+    FILTER_STOPWORDS: bool = True
+    UNIGRAM_NORMALIZER: str = "lemma"
+    USE_NGRAMS: bool = True
+    NGRAM_MIN: int = 2
+    NGRAM_MAX: int = 3
+
+    CSV_DOWNLOAD_PATH: str = "/tmp/input.csv"
+
+    csv_input: CSVFileInput
+    normalized_docs_output: NormalizedDocsOutput
+    normalized_overwritten_file_output: NormalizedCSVOutput
+
+
 def _write_preprocessed_docs_to_postgres(
     preprocessed_ouput: list[PreprocessedDocument],
     settings: DatabaseSettings,
@@ -172,3 +200,21 @@ def preprocess_bib_file(settings):
         overwrite_callback=loader.overwrite_with_results,
         settings=settings,
     )
+
+
+@entrypoint(PreprocessCSV)
+def preprocess_csv_file(settings):
+    logger.info("Downloading CSV file...")
+    S3Operations.download(settings.csv_input, settings.CSV_DOWNLOAD_PATH)
+
+    loader = CSVLoader(
+        file_path=settings.CSV_DOWNLOAD_PATH,
+        attribute=settings.csv_input.SELECTED_ATTRIBUTE,
+        id_column=settings.csv_input.ID_COLUMN,
+    )
+
+    _preprocess_and_store(
+        documents=loader.document_records,
+        overwrite_callback=loader.overwrite_with_results,
+        settings=settings,
+    )

preprocessing/loader.py

@@ -1,6 +1,7 @@
 import logging
 import re
 import bibtexparser
+import pandas as pd
 from typing import List
 from pathlib import Path
 
@@ -111,3 +112,86 @@ def overwrite_with_results(
             bibtexparser.dump(self.bib_db, f)
 
         logger.info(f"BIB file successfully written to: {output_path}")
+
+
+class CSVLoader:
+    def __init__(self, file_path: str, attribute: str, id_column: str = "id"):
+        logger.info(
+            f"Loading CSV file (attribute={attribute}, id_column={id_column})."
+        )
+
+        self.file_path = file_path
+        self.attribute = attribute
+        self.id_column = id_column
+
+        self.df = pd.read_csv(file_path)
+
+        if self.attribute not in self.df.columns:
+            raise ValueError(
+                f"Column '{self.attribute}' not found in CSV file."
+            )
+
+        if self.id_column not in self.df.columns:
+            raise ValueError(
+                f"ID column '{self.id_column}' not found in CSV file."
+            )
+
+        self.document_records = self._build_document_records()
+
+    @staticmethod
+    def _extract_doc_id(row: pd.Series, id_column: str) -> str:
+        value = row.get(id_column)
+
+        if pd.isna(value):
+            return "UNKNOWN_ID"
+
+        return str(value)
+
+    def _build_document_records(self) -> List[DocumentRecord]:
+        records = []
+
+        for _, row in self.df.iterrows():
+            doc_id = self._extract_doc_id(row, self.id_column)
+
+            raw_value = row.get(self.attribute, "")
+
+            if pd.isna(raw_value):
+                raw_value = ""
+
+            normalized = normalize_text(str(raw_value))
+
+            records.append(
+                DocumentRecord(
+                    doc_id=doc_id,
+                    text=normalized,
+                )
+            )
+
+        return records
+
+    def overwrite_with_results(
+        self,
+        preprocessed_docs: List[PreprocessedDocument],
+        export_path: Path,
+    ) -> None:
+        logger.info("Overwriting CSV documents with preprocessed text...")
+
+        output_path = Path.cwd() / export_path.name
+
+        preprocessed_dict = {doc.doc_id: doc for doc in preprocessed_docs}
+
+        updated_df = self.df.copy()
+
+        for idx, row in updated_df.iterrows():
+            doc_id = self._extract_doc_id(row, self.id_column)
+
+            preprocessed = preprocessed_dict.get(doc_id)
+
+            if not preprocessed:
+                continue
+
+            updated_df.at[idx, self.attribute] = " ".join(preprocessed.tokens)
+
+        updated_df.to_csv(output_path, index=False)
+
+        logger.info(f"CSV file successfully written to: {output_path}")

test/files/input.csv

@@ -0,0 +1,6 @@
+id,title,abstract
+1,Neural Network Optimization,"This invention relates to optimizing neural network training using adaptive gradient techniques and distributed computation."
+2,Battery Management System,"A system for monitoring lithium ion battery health, temperature regulation, and charge balancing in electric vehicles."
+3,Medical Imaging Analysis,"Methods for automated tumor detection in MRI scans using convolutional neural networks and image segmentation."
+4,Autonomous Navigation,"A navigation framework for autonomous mobile robots using sensor fusion and probabilistic path planning."
+5,Natural Language Processing,"Techniques for multilingual text classification and semantic similarity detection using transformer architectures."

test/test_full.py

@@ -4,7 +4,7 @@
 import pandas as pd
 
 from pathlib import Path
-from main import preprocess_bib_file, preprocess_txt_file
+from main import preprocess_bib_file, preprocess_txt_file, preprocess_csv_file
 from botocore.exceptions import ClientError
 from sqlalchemy import create_engine
 
@@ -14,6 +14,7 @@
 
 PG_USER = "postgres"
 PG_PASS = "postgres"
+DB_SCHEMA = ""
 
 INPUT_FILE_NAME = "input"
 OUTPUT_FILE_NAME = "output"
@@ -82,6 +83,7 @@ def test_full_bib(s3_minio):
         # PostgreSQL output
         "normalized_docs_DB_DSN": f"postgresql://{PG_USER}:{PG_PASS}@127.0.0.1:5432/postgres",
         "normalized_docs_DB_TABLE": "normalized_docs_bib",
+        "normalized_docs_DB_SCHEMA": DB_SCHEMA,
         "normalized_overwritten_file_output_S3_HOST": "http://127.0.0.1",
         "normalized_overwritten_file_output_S3_PORT": "9000",
         "normalized_overwritten_file_output_S3_ACCESS_KEY": MINIO_USER,
@@ -145,6 +147,7 @@ def test_full_txt(s3_minio):
         # Postgres output
         "normalized_docs_DB_DSN": f"postgresql://{PG_USER}:{PG_PASS}@127.0.0.1:5432/postgres",
         "normalized_docs_DB_TABLE": "normalized_docs_txt",
+        "normalized_docs_DB_SCHEMA": DB_SCHEMA,
         "normalized_overwritten_file_output_S3_HOST": "http://127.0.0.1",
         "normalized_overwritten_file_output_S3_PORT": "9000",
         "normalized_overwritten_file_output_S3_ACCESS_KEY": MINIO_USER,
@@ -177,3 +180,75 @@ def test_full_txt(s3_minio):
     assert all(isinstance(t, str) for t in df.iloc[0]["tokens"])
 
     # TODO: Test overwritten file upload
+
+
+def test_full_csv(s3_minio):
+    csv_path = Path(__file__).parent / "files" / f"{INPUT_FILE_NAME}.csv"
+    csv_bytes = csv_path.read_bytes()
+
+    # Upload input to MinIO
+    s3_minio.put_object(
+        Bucket=BUCKET_NAME,
+        Key=f"{INPUT_FILE_NAME}.csv",
+        Body=csv_bytes,
+    )
+
+    env = {
+        "UNIGRAM_NORMALIZER": "porter",
+        # CSV input S3
+        "csv_file_S3_HOST": "http://127.0.0.1",
+        "csv_file_S3_PORT": "9000",
+        "csv_file_S3_ACCESS_KEY": MINIO_USER,
+        "csv_file_S3_SECRET_KEY": MINIO_PWD,
+        "csv_file_BUCKET_NAME": BUCKET_NAME,
+        "csv_file_FILE_PATH": "",
+        "csv_file_FILE_NAME": INPUT_FILE_NAME,
+        "csv_file_SELECTED_ATTRIBUTE": "abstract",
+        "csv_file_ID_COLUMN": "id",
+        # PostgreSQL output
+        "normalized_docs_DB_DSN": (
+            f"postgresql://{PG_USER}:{PG_PASS}@127.0.0.1:5432/postgres"
+        ),
+        "normalized_docs_DB_TABLE": "normalized_docs_csv",
+        "normalized_docs_DB_SCHEMA": DB_SCHEMA,
+        # overwritten CSV output
+        "normalized_overwritten_file_output_S3_HOST": "http://127.0.0.1",
+        "normalized_overwritten_file_output_S3_PORT": "9000",
+        "normalized_overwritten_file_output_S3_ACCESS_KEY": MINIO_USER,
+        "normalized_overwritten_file_output_S3_SECRET_KEY": MINIO_PWD,
+        "normalized_overwritten_file_output_BUCKET_NAME": BUCKET_NAME,
+        "normalized_overwritten_file_output_FILE_PATH": "",
+        "normalized_overwritten_file_output_FILE_NAME": OUTPUT_FILE_NAME,
+    }
+
+    for k, v in env.items():
+        os.environ[k] = v
+
+    # Run block
+    preprocess_csv_file()
+
+    # Query PostgreSQL
+    engine = create_engine(
+        f"postgresql+psycopg2://{PG_USER}:{PG_PASS}@localhost:5432/"
+    )
+
+    df = pd.read_sql_table("normalized_docs_csv", engine)
+
+    # Assertions
+    assert len(df) > 0
+    assert "doc_id" in df.columns
+    assert "tokens" in df.columns
+
+    # IDs
+    assert len(df["doc_id"]) == len(df)
+    assert df["doc_id"].is_unique
+    assert all(isinstance(x, str) for x in df["doc_id"])
+
+    # Convert PG arrays
+    df["tokens"] = df["tokens"].apply(parse_pg_array)
+
+    # Token structure
+    assert isinstance(df.iloc[0]["tokens"], list)
+    assert all(isinstance(t, str) for t in df.iloc[0]["tokens"])
+
+    # TODO: Test overwritten file upload