import os as PI_OS import polars as PI_POLARS # Import def FC_IMPORT_TEXT( ZVFCI_ST_SOURCE_FILE, ZVFCI_ST_FOLDER, ZVFCI_ST_DELIMITER, ZVFCI_ST_PREFIX, ZVFCI_LI_NUM_COLS ): ZV_ST_FILE_PATH = PI_OS.path.join(ZVFCI_ST_FOLDER, ZVFCI_ST_SOURCE_FILE) ZVFCI_LI_ENCODINGS = [ 'utf8', 'utf-8-sig', 'utf8-lossy', 'windows-1252', 'latin1', 'iso-8859-1', 'iso-8859-15', 'utf-16', 'utf-16-le', 'utf-16-be', 'cp1250', 'cp1251', 'cp1253', 'cp1254', 'cp1257' ] ZV_ER_ALL = None ZV_DF_HEADER = None for ZV_ST_ENCODING in ZVFCI_LI_ENCODINGS: try: ZV_DF_HEADER = PI_POLARS.read_csv( ZV_ST_FILE_PATH, separator=ZVFCI_ST_DELIMITER, encoding=ZV_ST_ENCODING, has_header=True, n_rows=1, infer_schema_length=0, quote_char=None ) ZV_LI_COLS = ZV_DF_HEADER.columns break except Exception as ZV_ER: print(f'Cannot encode using: {ZV_ER}') if not ZV_DF_HEADER is None: ZV_DI_SCHEMA_OVERRIDES = {ZV_COL: PI_POLARS.Utf8 for ZV_COL in ZV_LI_COLS} ZV_DF = PI_POLARS.read_csv( ZV_ST_FILE_PATH, separator=ZVFCI_ST_DELIMITER, encoding=ZV_ST_ENCODING, ignore_errors=True, has_header=True, schema_overrides=ZV_DI_SCHEMA_OVERRIDES, infer_schema_length=0, null_values=[''], quote_char=None ).fill_null('') for ZV_COL in ZV_DF.columns: ZV_DF = ZV_DF.rename( { ZV_COL: f'{ZVFCI_ST_PREFIX}{ZV_COL}' } ) for ZV_COL in ZVFCI_LI_NUM_COLS: ZV_DF = ZV_DF.with_columns( [ PI_POLARS.col(f'{ZV_COL}') .cast(PI_POLARS.Float64, strict=False) .alias(f'{ZV_COL}') ] ) return ZV_DF