Решение проблемы UnicodeDecodeError в Pandas: 5 проверенных методов

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xd0 in position 4: invalid continuation byte

import pandas as pd

df = pd.read_csv('файл.csv', encoding='название_кодировки')

# Пробуем разные кодировки
encodings = ['utf-8', 'cp1251', 'latin1', 'utf-16', 'utf-8-sig']

for enc in encodings:
try:
df = pd.read_csv('проблемный_файл.csv', encoding=enc)
print(f"Успешно прочитано с кодировкой {enc}")
break # Выходим из цикла при успехе
except UnicodeDecodeError:
print(f"Ошибка при использовании кодировки {enc}")

pip install chardet

import pandas as pd
import chardet

def detect_and_read_csv(file_path, **kwargs):
# Определяем кодировку файла
with open(file_path, 'rb') as f:
result = chardet.detect(f.read())

# Получаем наиболее вероятную кодировку и уровень уверенности в результате
encoding = result['encoding']
confidence = result['confidence']

print(f"Определена кодировка: {encoding} с уверенностью {confidence:.2%}")

# Читаем файл с определенной кодировкой
return pd.read_csv(file_path, encoding=encoding, **kwargs)

# Использование функции
try:
df = detect_and_read_csv('проблемный_файл.csv')
print("Файл успешно прочитан!")
except Exception as e:
print(f"Произошла ошибка: {e}")

def smart_read_csv(file_path, **kwargs):
# Список кодировок для проверки в порядке приоритета
encodings_to_try = ['utf-8', 'cp1251', 'latin1']

# Пробуем распространенные кодировки сначала
for enc in encodings_to_try:
try:
return pd.read_csv(file_path, encoding=enc, **kwargs)
except UnicodeDecodeError:
continue

# Если не сработало, используем chardet
return detect_and_read_csv(file_path, **kwargs)

import pandas as pd
import logging
import chardet
import os

# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
filename='csv_processing.log'
)
logger = logging.getLogger('csv_reader')

def robust_read_csv(file_path, fallback_encodings=None, **kwargs):
"""
Функция для надежного чтения CSV файлов с обработкой ошибок кодировки

Parameters:
-----------
file_path : str
Путь к CSV файлу
fallback_encodings : list, optional
Список кодировок для попытки чтения в случае ошибки
**kwargs : dict
Дополнительные параметры для pd.read_csv

Returns:
--------
DataFrame или None в случае неустранимой ошибки
"""
if fallback_encodings is None:
fallback_encodings = ['utf-8', 'cp1251', 'latin1', 'utf-16', 'utf-8-sig']

# Проверяем существование файла
if not os.path.exists(file_path):
logger.error(f"Файл {file_path} не существует")
return None

# Проверяем размер файла
file_size = os.path.getsize(file_path)
if file_size == 0:
logger.warning(f"Файл {file_path} пустой")
return pd.DataFrame()

# Пробуем стандартные кодировки
for encoding in fallback_encodings:
try:
df = pd.read_csv(file_path, encoding=encoding, **kwargs)
logger.info(f"Файл {file_path} успешно прочитан с кодировкой {encoding}")
return df
except UnicodeDecodeError:
logger.debug(f"Не удалось прочитать файл с кодировкой {encoding}")
continue
except Exception as e:
logger.error(f"Неожиданная ошибка при чтении файла с кодировкой {encoding}: {str(e)}")
continue

# Если стандартные кодировки не сработали, пробуем определить кодировку автоматически
try:
# Читаем не более 1МБ для экономии памяти при больших файлах
sample_size = min(file_size, 1024*1024)
with open(file_path, 'rb') as f:
raw_data = f.read(sample_size)
result = chardet.detect(raw_data)

detected_encoding = result['encoding']
confidence = result['confidence']

if confidence > 0.7: # Достаточный уровень уверенности
try:
df = pd.read_csv(file_path, encoding=detected_encoding, **kwargs)
logger.info(f"Файл {file_path} прочитан с автоопределенной кодировкой {detected_encoding} (уверенность: {confidence:.2%})")
return df
except Exception as e:
logger.error(f"Ошибка при чтении с автоопределенной кодировкой {detected_encoding}: {str(e)}")
else:
logger.warning(f"Низкая уверенность в определении кодировки: {confidence:.2%}")

except Exception as e:
logger.error(f"Ошибка при определении кодировки: {str(e)}")

# Последняя попытка – чтение с игнорированием ошибок
try:
df = pd.read_csv(file_path, encoding='latin1', errors='replace', **kwargs)
logger.warning(f"Файл {file_path} прочитан с заменой проблемных символов (кодировка: latin1)")
return df
except Exception as e:
logger.critical(f"Все попытки чтения файла {file_path} не удались. Последняя ошибка: {str(e)}")
return None

# Пример использования
try:
df = robust_read_csv('проблемный_файл.csv')
if df is not None:
print(f"Успешно прочитано {len(df)} строк")
else:
print("Не удалось прочитать файл")
except Exception as e:
print(f"Критическая ошибка: {str(e)}")

def process_csv_directory(directory_path, output_dir=None, **kwargs):
"""Обработка всех CSV файлов в указанной директории"""
if output_dir and not os.path.exists(output_dir):
os.makedirs(output_dir)

results = {}
failed_files = []

for filename in os.listdir(directory_path):
if filename.lower().endswith('.csv'):
file_path = os.path.join(directory_path, filename)

df = robust_read_csv(file_path, **kwargs)

if df is not None:
results[filename] = df

# Сохраняем результат, если указан выходной каталог
if output_dir:
output_path = os.path.join(output_dir, filename)
df.to_csv(output_path, index=False, encoding='utf-8')
logger.info(f"Файл {filename} обработан и сохранен в {output_path}")
else:
failed_files.append(filename)
logger.error(f"Не удалось обработать файл {filename}")

logger.info(f"Обработано файлов: {len(results)}, не удалось обработать: {len(failed_files)}")
return results, failed_files

import codecs
import os

def convert_file_encoding(input_file, output_file, source_encoding, target_encoding='utf-8'):
"""
Конвертирует файл из одной кодировки в другую

Parameters:
-----------
input_file : str
Путь к исходному файлу
output_file : str
Путь для сохранения преобразованного файла
source_encoding : str
Исходная кодировка файла
target_encoding : str, optional
Целевая кодировка (по умолчанию 'utf-8')

Returns:
--------
bool
True если конвертация прошла успешно, False в противном случае
"""
try:
# Создаем директорию для выходного файла, если она не существует
output_dir = os.path.dirname(output_file)
if output_dir and not os.path.exists(output_dir):
os.makedirs(output_dir)

# Открываем исходный файл в указанной кодировке
with codecs.open(input_file, 'r', encoding=source_encoding) as f_input:
# Открываем выходной файл для записи в целевой кодировке
with codecs.open(output_file, 'w', encoding=target_encoding) as f_output:
# Копируем содержимое строка за строкой
for line in f_input:
f_output.write(line)

return True
except Exception as e:
print(f"Ошибка при конвертации файла: {str(e)}")
return False

# Пример использования
if convert_file_encoding('проблемный_файл.csv', 'utf8_файл.csv', 'cp1251'):
# Теперь можно читать сконвертированный файл без проблем с кодировкой
df = pd.read_csv('utf8_файл.csv')
print("Файл успешно сконвертирован и прочитан")

def convert_to_utf8(input_file, output_file=None):
"""
Автоматически определяет кодировку файла и конвертирует его в UTF-8

Parameters:
-----------
input_file : str
Путь к исходному файлу
output_file : str, optional
Путь для сохранения преобразованного файла. Если не указан, 
создается файл с префиксом 'utf8_' в том же каталоге

Returns:
--------
str или None
Путь к сконвертированному файлу в случае успеха, None в случае ошибки
"""
if output_file is None:
dir_name = os.path.dirname(input_file)
file_name = os.path.basename(input_file)
output_file = os.path.join(dir_name, 'utf8_' + file_name)

# Определяем кодировку файла
with open(input_file, 'rb') as f:
raw_data = f.read()
result = chardet.detect(raw_data)

source_encoding = result['encoding']
confidence = result['confidence']

print(f"Определена кодировка: {source_encoding} с уверенностью {confidence:.2%}")

if confidence < 0.7:
print("Внимание: низкий уровень уверенности в определении кодировки")

# Конвертируем файл
if convert_file_encoding(input_file, output_file, source_encoding, 'utf-8'):
return output_file
else:
return None

def stream_convert_encoding(input_file, output_file, source_encoding, target_encoding='utf-8', buffer_size=1024*1024):
"""
Потоковая конвертация файла из одной кодировки в другую с минимальным использованием памяти

Parameters:
-----------
buffer_size : int, optional
Размер буфера для чтения/записи (по умолчанию 1MB)
"""
try:
with codecs.open(input_file, 'r', encoding=source_encoding, buffering=buffer_size) as f_input:
with codecs.open(output_file, 'w', encoding=target_encoding, buffering=buffer_size) as f_output:
while True:
chunk = f_input.read(buffer_size)
if not chunk:
break
f_output.write(chunk)
return True
except Exception as e:
print(f"Ошибка при потоковой конвертации: {str(e)}")
return False

import pandas as pd

try:
# Используем Python-движок с заменой проблемных символов и пропуском проблемных строк
df = pd.read_csv(
'проблемный_файл.csv',
encoding='utf-8',
encoding_errors='replace', # заменяет непонятные символы на 
engine='python', # более медленный, но более гибкий движок
on_bad_lines='skip', # пропускает строки, которые нельзя распарсить
quoting=3, # QUOTE_NONE – отключает специальную обработку кавычек
escapechar='\\' # указывает символ экранирования
)
print(f"Файл прочитан, получено {len(df)} строк")
except Exception as e:
print(f"Ошибка: {str(e)}")

def read_csv_manually(file_path, encoding, delimiter=',', header=True):
"""
Ручное чтение CSV файла с полным контролем над процессом

Parameters:
-----------
file_path : str
Путь к CSV файлу
encoding : str
Кодировка файла
delimiter : str, optional
Разделитель полей (по умолчанию ',')
header : bool, optional
Имеет ли файл заголовок (по умолчанию True)

Returns:
--------
DataFrame
"""
with open(file_path, 'r', encoding=encoding, errors='replace') as f:
lines = []
for line in f:
# Удаляем символы новой строки и разделяем по delimiter
values = line.rstrip('\r\n').split(delimiter)
lines.append(values)

if not lines:
return pd.DataFrame()

if header:
# Первая строка – заголовки
headers = lines[0]
data = lines[1:]
return pd.DataFrame(data, columns=headers)
else:
return pd.DataFrame(lines)

# Пример использования
df = read_csv_manually('проблемный_файл.csv', encoding='latin1')
print(df.head())

import csv
import pandas as pd

def csv_to_dataframe(file_path, encoding):
"""
Чтение CSV через стандартную библиотеку CSV с последующим преобразованием в DataFrame
"""
data = []
with open(file_path, 'r', encoding=encoding, newline='', errors='replace') as f:
reader = csv.reader(f)
for row in reader:
data.append(row)

if not data:
return pd.DataFrame()

# Первая строка становится заголовком
headers = data[0]
rows = data[1:]

return pd.DataFrame(rows, columns=headers)

def process_large_csv(file_path, encoding, chunksize=10000):
"""
Обработка большого CSV файла частями

Parameters:
-----------
chunksize : int, optional
Количество строк для обработки за раз (по умолчанию 10000)
"""
chunks = pd.read_csv(
file_path, 
encoding=encoding,
encoding_errors='replace',
chunksize=chunksize, # Читаем файл порциями
low_memory=True # Оптимизация для больших файлов
)

# Здесь можно обрабатывать каждую часть отдельно
result = pd.DataFrame()
for i, chunk in enumerate(chunks):
print(f"Обработка части {i+1}, {len(chunk)} строк")

# Пример обработки: фильтрация данных
filtered_chunk = chunk[chunk['some_column'] > 0]

# Объединяем с общим результатом
result = pd.concat([result, filtered_chunk])

return result