Python中处理CSV文件的核心方法详解

import csv

# 读取CSV文件
with open('employees.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row_num, row in enumerate(reader, 1):
        print(f"第{row_num}行: {row}")

import csv

# 准备数据
sample_data = [
    ['姓名', '年龄', '城市', '职业'],
    ['张三', '25', '北京', '工程师'],
    ['李四', '30', '上海', '设计师'],
    ['王五', '28', '广州', '产品经理']
]

# 写入CSV文件
with open('employees.csv', 'w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file)
    writer.writerows(sample_data)

print("已创建示例CSV文件 'employees.csv'")

import csv

with open('employees.csv', 'r', encoding='utf-8') as file:
    dict_reader = csv.DictReader(file)
    for row_num, row in enumerate(dict_reader, 1):
        print(f"员工{row_num}:")
        for key, value in row.items():
            print(f"  {key}: {value}")
        print()

import csv

# 准备字段名称和数据
fieldnames = ['姓名', '年龄', '城市', '职业']
new_employee = {
    '姓名': '周九',
    '年龄': '29',
    '城市': '武汉',
    '职业': '前端开发'
}

# 追加新员工数据
with open('employees.csv', 'a', newline='', encoding='utf-8') as file:
    dict_writer = csv.DictWriter(file, fieldnames=fieldnames)
    dict_writer.writerow(new_employee)

print("已使用DictWriter添加新员工")

import csv

# 创建使用分号分隔的CSV文件
custom_data = [
    ['产品名称', '价格', '库存'],
    ['苹果', '5.5', '100'],
    ['香蕉', '3.2', '200'],
    ['橙子', '4.8', '150']
]

with open('products.csv', 'w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file, delimiter=';', quoting=csv.QUOTE_MINIMAL)
    writer.writerows(custom_data)

print("已创建使用分号分隔的CSV文件 'products.csv'")

# 读取自定义分隔符的CSV文件
with open('products.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file, delimiter=';')
    for row_num, row in enumerate(reader, 1):
        print(f"第{row_num}行: {row}")

import csv

# 创建包含特殊字符的CSV数据
special_data = [
    ['姓名', '描述'],
    ['张三', '喜欢"Python"编程'],
    ['李四', '爱好: 读书, 游泳, 旅行'],
    ['王五', '地址: 北京市朝阳区\n邮编: 100000']
]

with open('special.csv', 'w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file, quoting=csv.QUOTE_ALL)
    writer.writerows(special_data)

print("已创建包含特殊字符的CSV文件 'special.csv'")

# 读取包含特殊字符的CSV文件
with open('special.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row_num, row in enumerate(reader, 1):
        print(f"第{row_num}行: {row}")

import csv

def count_employees_by_city(csv_file):
    """统计不同城市的员工数量"""
    city_count = {}
    
    with open(csv_file, 'r', encoding='utf-8') as file:
        dict_reader = csv.DictReader(file)
        for row in dict_reader:
            city = row['城市']
            if city in city_count:
                city_count[city] += 1
            else:
                city_count[city] = 1
    
    return city_count

# 示例使用
if __name__ == "__main__":
    # 假设employees.csv文件已存在
    city_counts = count_employees_by_city('employees.csv')
    print("各城市员工数量统计:")
    for city, count in city_counts.items():
        print(f"  {city}: {count}人")

import csv
import json

def csv_to_json(csv_file, json_file):
    """将CSV文件转换为JSON文件"""
    data = []
    
    with open(csv_file, 'r', encoding='utf-8') as file:
        dict_reader = csv.DictReader(file)
        for row in dict_reader:
            data.append(row)
    
    with open(json_file, 'w', encoding='utf-8') as file:
        json.dump(data, file, ensure_ascii=False, indent=2)
    
    print(f"已将 {csv_file} 转换为 {json_file}")

# 示例使用
if __name__ == "__main__":
    csv_to_json('employees.csv', 'employees.json')

import csv

def filter_employees_by_age(csv_file, min_age):
    """过滤年龄大于等于指定值的员工"""
    filtered_data = []
    
    with open(csv_file, 'r', encoding='utf-8') as file:
        dict_reader = csv.DictReader(file)
        fieldnames = dict_reader.fieldnames
        
        for row in dict_reader:
            if int(row['年龄']) >= min_age:
                filtered_data.append(row)
    
    # 保存过滤结果
    output_file = f'employees_age_{min_age}+.csv'
    with open(output_file, 'w', newline='', encoding='utf-8') as file:
        dict_writer = csv.DictWriter(file, fieldnames=fieldnames)
        dict_writer.writeheader()
        dict_writer.writerows(filtered_data)
    
    print(f"已过滤并保存结果到 {output_file}")
    return filtered_data

# 示例使用
if __name__ == "__main__":
    filtered = filter_employees_by_age('employees.csv', 30)
    print(f"过滤后共有 {len(filtered)} 名员工")

import csv
import os

def merge_csv_files(input_files, output_file):
    """合并多个CSV文件"""
    if not input_files:
        print("没有输入文件")
        return
    
    # 读取第一个文件的表头
    with open(input_files[0], 'r', encoding='utf-8') as file:
        dict_reader = csv.DictReader(file)
        fieldnames = dict_reader.fieldnames
    
    # 合并数据
    merged_data = []
    for csv_file in input_files:
        with open(csv_file, 'r', encoding='utf-8') as file:
            dict_reader = csv.DictReader(file)
            for row in dict_reader:
                merged_data.append(row)
    
    # 写入合并结果
    with open(output_file, 'w', newline='', encoding='utf-8') as file:
        dict_writer = csv.DictWriter(file, fieldnames=fieldnames)
        dict_writer.writeheader()
        dict_writer.writerows(merged_data)
    
    print(f"已合并 {len(input_files)} 个CSV文件到 {output_file}")
    print(f"总记录数: {len(merged_data)}")

# 示例使用
if __name__ == "__main__":
    input_files = ['employees1.csv', 'employees2.csv', 'employees3.csv']
    # 过滤存在的文件
    input_files = [f for f in input_files if os.path.exists(f)]
    if input_files:
        merge_csv_files(input_files, 'merged_employees.csv')
    else:
        print("没有找到输入文件")