处理CSV文件时：如何优雅应对编码错误、数据类型转换异常？

import chardet

def detect_encoding(file_path):
    with open(file_path, 'rb') as f:
        raw_data = f.read()
    result = chardet.detect(raw_data)
    return result['encoding']

file_path = 'your_file.csv'
encoding = detect_encoding(file_path)

# 再次尝试，如果chardet失败，则使用备选方案
if not encoding:
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            f.read()
        encoding = 'utf-8'
    except UnicodeDecodeError:
        encoding = 'latin1' # 备选编码，通常latin1能处理大部分情况

print(f"Detected encoding: {encoding}")

import csv
with open(file_path, 'r', encoding=encoding) as csvfile:
    reader = csv.reader(csvfile)
    for row in reader:
        print(row)

import csv
from datetime import datetime

def convert_row(row, column_types):
    converted_row = []
    for i, value in enumerate(row):
        converter = column_types[i]
        try:
            converted_value = converter(value)
        except (ValueError, TypeError):
            converted_value = None  # 或者使用默认值，记录错误等
        converted_row.append(converted_value)
    return converted_row

# 定义列的类型转换器
column_types = [int, float, str, lambda x: datetime.strptime(x, '%Y-%m-%d')]

file_path = 'your_file.csv'
with open(file_path, 'r') as csvfile:
    reader = csv.reader(csvfile)
    header = next(reader) # 跳过标题行
    for row in reader:
        converted_row = convert_row(row, column_types)
        print(converted_row)

import csv

def process_large_csv(file_path, chunk_size=1000):
    with open(file_path, 'r') as csvfile:
        reader = csv.reader(csvfile)
        header = next(reader) # 跳过标题行

        chunk = []
        for i, row in enumerate(reader):
            chunk.append(row)
            if (i + 1) % chunk_size == 0:
                yield chunk
                chunk = []

        if chunk: # 处理剩余的行
            yield chunk

file_path = 'your_large_file.csv'
for chunk in process_large_csv(file_path):
    # 在这里处理每个chunk
    for row in chunk:
        print(row)

import csv
import logging

# 配置日志
logging.basicConfig(filename='csv_processing.log', level=logging.ERROR,
                    format='%(asctime)s - %(levelname)s - %(message)s')

def process_row(row):
    try:
        # 模拟一些可能出错的操作
        value1 = int(row[0])
        value2 = float(row[1])
        result = value1 / value2
        return result
    except (ValueError, ZeroDivisionError) as e:
        logging.error(f"Error processing row: {row}. Error: {e}")
        return None

file_path = 'your_file.csv'
with open(file_path, 'r') as csvfile:
    reader = csv.reader(csvfile)
    header = next(reader)
    for row in reader:
        result = process_row(row)
        if result is not None:
            print(f"Result: {result}")