loans.csv

plumb.rb

require "csv"
require "plumb"
require "date"
require "active_support/core_ext/string"

module Types
  include Plumb::Types

  NumericString = Types::String.transform(Float) do |str|
    str.to_s.delete(",").to_f
  end

  DateString = Types::String.build(::Date) do |str|
    ::Date.strptime(str, "%d/%m/%Y")
  rescue Date::Error
    nil
  end

  LoanRecord = Types::Hash.schema(
    loan_no: Types::String.present,
    rgn: NumericString.present,
    account_name: Types::String.present,
    principal_bal: NumericString.present,
    interest_bal: NumericString.present,
    total: NumericString.present,
    disbursed: DateString.present,
    end_date: DateString.present,
    int_rate: NumericString.present,
    tenure: NumericString.present,
    current_inst: NumericString.present,
    rem_inst: NumericString.present,
    prinpal_pyt_mth: NumericString.present,
    int_pyt_mth: NumericString.present,
    total_pyt_mth: NumericString.present,
    cagd_payment: NumericString.present,
    disbursed_amount: NumericString.present,
    acc_int: NumericString.present
  )
end

def normalize_header(header)
  header
    .parameterize
    .underscore
    .to_sym
end

def process_loans(filepath)
  valid_loans = []
  invalid_loans = []

  CSV.foreach(filepath, headers: true) do |row|
    normalized_row = row.to_h.transform_keys { |k| normalize_header(k) }

    result = Types::LoanRecord.resolve(normalized_row)

    if result.valid?
      valid_loans << result.value
    else
      invalid_loans << {
        loan_no: row["Loan No."],
        row: normalized_row,
        errors: result.errors
      }
    end
  end

  {
    valid: valid_loans,
    invalid: invalid_loans,
    total_count: valid_loans.length + invalid_loans.length,
    valid_count: valid_loans.length,
    invalid_count: invalid_loans.length
  }
end

# Process and analyze the data
results = process_loans("loans.csv")

puts "\nProcessing Summary:"
puts "Total records: #{results[:total_count]}"
puts "Valid records: #{results[:valid_count]}"
puts "Invalid records: #{results[:invalid_count]}"

if results[:valid].any?
  puts "\nValid Records Details:"
  results[:valid].each do |valid|
    puts "\nLoan No. #{valid[:loan_no]}"
    puts "Row: #{valid}"
  end
end

if results[:invalid].any?
  puts "\nInvalid Records Details:"
  results[:invalid].each do |invalid|
    puts "\nLoan No. #{invalid[:loan_no]}"
    puts "Available keys: #{invalid[:row].keys.join(", ")}"
    puts "Errors: #{invalid[:errors]}"
  end
end

plumb_v2.rb

require "csv"
require "plumb"
require "date"
require "active_support/core_ext/string"

module Importer
  include Plumb::Types

  # FileHandle type checks for file existence and creates a File object
  # Example usage:
  #   file = FileHandle.parse('./files/data.csv') # => File
  # Raises error if file doesn't exist
  FileHandle = Plumb::Types::String
    .check("File does not exist") { |s| ::File.exist?(s) }
    .build(::File)

  # CSVStream converts a File object into a CSV enumerator with headers
  # Example usage:
  #   csv_enum = CSVStream.parse(file) #=> Enumerator
  CSVStream = Plumb::Types::Any[::File]
    .transform(::CSV) { ::CSV.new(it, headers: true) }
    .transform(::Enumerator, &:each)

  # Combines FileHandle and CSVStream to create a pipeline that:
  # 1. Takes a file path string
  # 2. Validates file existence
  # 3. Creates a CSV enumerator
  CSVFileStream = FileHandle >> CSVStream
end

module LoanImporter
  include Plumb::Types

  # Header type for normalizing CSV column headers:
  # 1. Parameterizes the string (handles special characters)
  # 2. Converts to underscore format
  # 3. Transforms to symbol
  # Example: "Account Name" -> :account_name
  Header = Plumb::Types::String
    .invoke(%i[parameterize underscore])
    .transform(::Symbol, &:to_sym)

  # Row type that enforces normalized headers while accepting any value type
  # This creates a hash with symbolized, normalized keys
  Row = Plumb::Types::Hash[Header, Plumb::Types::Any]

  # Schema definition for a loan record
  # Each field is marked as .present to ensure no missing values
  # Lax::Decimal allows flexible parsing of decimal numbers
  # Forms::Date handles date string parsing
  Record = Plumb::Types::Hash[
    loan_no: Plumb::Types::Lax::String.present,
    rgn: Plumb::Types::Lax::Decimal.present,
    account_name: Plumb::Types::String.present,
    principal_bal: Plumb::Types::Lax::Decimal.present,
    interest_bal: Plumb::Types::Lax::Decimal.present,
    total: Plumb::Types::Lax::Decimal.present,
    disbursed: Forms::Date.present,
    end_date: Forms::Date.present,
    int_rate: Plumb::Types::Lax::Decimal.present,
    tenure: Plumb::Types::Lax::Decimal.present,
    current_inst: Plumb::Types::Lax::Decimal.present,
    rem_inst: Plumb::Types::Lax::Decimal.present,
    prinpal_pyt_mth: Plumb::Types::Lax::Decimal.present,
    int_pyt_mth: Plumb::Types::Lax::Decimal.present,
    total_pyt_mth: Plumb::Types::Lax::Decimal.present,
    cagd_payment: Plumb::Types::Lax::Decimal.present,
    disbursed_amount: Plumb::Types::Lax::Decimal.present,
    acc_int: Plumb::Types::Lax::Decimal.present
  ]

  # Pipeline for normalizing and validating CSV rows:
  # 1. Converts CSV::Row to Hash
  # 2. Normalizes keys using Row type
  # 3. Validates against Record schema
  NormalizeCSVRow = Plumb::Types::Any.pipeline do |pl|
    pl.step Any.transform(::Hash, &:to_h)
    pl.step Row
    pl.step Record
  end

  # Complete pipeline that:
  # 1. Opens and streams CSV file
  # 2. Processes each row through NormalizeCSVRow
  # 3. Returns a Stream of validated Records
  CSVFileStream = Importer::CSVFileStream >> Plumb::Types::Stream[NormalizeCSVRow]
end

# Process the CSV file and output results
# For each row:
# - If valid: displays the parsed record
# - If invalid: displays validation errors
LoanImporter::CSVFileStream.parse("loans.csv").each.with_index do |row, index|
  if row.valid?
    puts "Row #{index}: #{row.value.inspect}"
  else
    puts "Error in row #{index}: #{row.errors.inspect}"
  end
  puts "~~~"
end

Nice! This is another version, a bit more "plumbisized", with comments.

https://gist.github.com/ismasan/6f27dbdf680bdc35a88c5bd0dc0864a4