csv-handler

name	csv-handler
description	Handle CSV files from construction software exports. Auto-detect delimiters, encodings, and clean messy data.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🏷️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":["python3"]}}}

CSV Handler for Construction Data

Overview

CSV is the universal exchange format in construction - from scheduling exports to cost databases. This skill handles encoding issues, delimiter detection, and data cleaning.

Python Implementation

import pandas as pd
import csv
from typing import Dict, Any, List, Optional, Tuple
from pathlib import Path
from dataclasses import dataclass
import chardet


@dataclass
class CSVProfile:
    
    encoding: 
    delimiter: 
    has_header: 
    row_count: 
    column_count: 
    columns: []


 :
    

    COMMON_DELIMITERS = [, , , ]
    COMMON_ENCODINGS = [, , , , ]

     ():
        .last_profile: [CSVProfile] = 

     () -> :
        
         (file_path, )  f:
            raw = f.read()
        result = chardet.detect(raw)
         result.get(, )  

     () -> :
        
         (file_path, , encoding=encoding, errors=)  f:
            sample = f.read()

        
        counts = {d: sample.count(d)  d  .COMMON_DELIMITERS}

        
         counts:
             (counts, key=counts.get)
         

     () -> CSVProfile:
        
        encoding = .detect_encoding(file_path)
        delimiter = .detect_delimiter(file_path, encoding)

        
        df = pd.read_csv(file_path, encoding=encoding, delimiter=delimiter,
                         nrows=, on_bad_lines=)

        has_header =  df.columns[].replace(, ).replace(, ).isdigit()

        
         (file_path, , encoding=encoding, errors=)  f:
            row_count = (  _  f) - (  has_header  )

        profile = CSVProfile(
            encoding=encoding,
            delimiter=delimiter,
            has_header=has_header,
            row_count=row_count,
            column_count=(df.columns),
            columns=(df.columns)
        )
        .last_profile = profile
         profile

     () -> pd.DataFrame:
        

        
         encoding  :
            encoding = .detect_encoding(file_path)
         delimiter  :
            delimiter = .detect_delimiter(file_path, encoding)

        
        df = pd.read_csv(
            file_path,
            encoding=encoding,
            delimiter=delimiter,
            on_bad_lines=,
            low_memory=
        )

         clean:
            df = .clean_dataframe(df)

         df

     () -> pd.DataFrame:
        
        
        df.columns = [._clean_column_name(c)  c  df.columns]

        
        df = df.dropna(how=)
        df = df.dropna(axis=, how=)

        
         col  df.select_dtypes(include=[]):
            df[col] = df[col]..strip()  df[col].dtype ==   df[col]

         df

     () -> :
        
          (name, ):
             (name)

        
        clean = name.strip().lower()
        clean = clean.replace(, ).replace(, )
        clean = .join(c  c  clean  c.isalnum()  c == )
         clean

     () -> pd.DataFrame:
        
        dfs = []
         path  file_paths:
            df = .read_csv(path)
            df[] = Path(path).name
            dfs.append(df)

          dfs:
             pd.DataFrame()

         on_column  on_column  dfs[].columns:
            result = dfs[]
             df  dfs[:]:
                result = pd.merge(result, df, on=on_column, how=)
             result

         pd.concat(dfs, ignore_index=)

     () -> []:
        
        output_path = Path(output_dir)
        output_path.mkdir(parents=, exist_ok=)

        files = []
         value  df[group_column].unique():
            subset = df[df[group_column] == value]
            filename = 
            filepath = output_path / filename
            subset.to_csv(filepath, index=)
            files.append((filepath))

         files

     () -> pd.DataFrame:
        
        df = df.copy()

         type_map:
             col, dtype  type_map.items():
                 col  df.columns:
                    :
                        df[col] = df[col].astype(dtype)
                    :
                        
        :
            
             col  df.columns:
                
                :
                    df[col] = pd.to_numeric(df[col])
                    
                :
                    

                
                :
                    df[col] = pd.to_datetime(df[col])
                :
                    

         df

     () -> :
        
        df.to_csv(file_path, encoding=encoding, sep=delimiter, index=)
         file_path



 ():
    

    SCHEDULE_COLUMNS = [, , , ,
                        , , ]

     () -> pd.DataFrame:
        
        df = .read_csv(file_path)

        
         col  df.columns:
               col.lower()    col.lower()    col.lower():
                :
                    df[col] = pd.to_datetime(df[col])
                :
                    

         df


 ():
    

     () -> pd.DataFrame:
        
        df = .read_csv(file_path)

        
         col  df.columns:
             (word  col.lower()  word  [, , , , , ]):
                df[col] = pd.to_numeric(df[col].replace(, , regex=), errors=)

         df