You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用Python的csv库定位CSV行并更新Access Valid列后导出Excel

用CSV库替代Pandas处理大文件时间校验并导出Excel

问题背景

我原本用Pandas写了一段时间校验代码,逻辑是检查用户请求时间是否在其计划的起止时间范围内,给Access Valid列赋值"Yes"或"No",最后导出Excel:

# 功能:时间校验
def checktime(reportedtime, scheduledtimestart, scheduledtimeend):
    if scheduledtimestart <= reportedtime <= scheduledtimeend:
        return "Yes"
    else:
        return "No"

df.loc[sub_df.index, 'Access Valid'] = checktime(time_date, schedule_start_date, schedule_end_date)

df.to_excel(current_directory + '/files/updated_report.xlsx', index=False)

其中df是原始DataFrame,sub_df是过滤后的DataFrame,Access Valid是需要填充的目标列。

现在因为文件体积过大,要求改用csv库替代Pandas,但不知道如何定位行来给CSV的Access Valid列赋值,最后导出Excel。

相关文件格式

  • 用户列表(CSV,分隔符为;):
user; scheduled start date; scheduled end date;
test1; 2022-09-01 00:00:00; 2022-09-30 23:59:59;
test2; 2022-09-01 00:00:00; 2022-09-14 23:59:59;
  • 请求记录文件(CSV,分隔符为;,含1000+条数据):
time; user; Access Valid
2022-09-20 09:09:00; test1; 
2022-09-21 05:00:00; test2; 
  • 期望输出:
time; user; Access Valid
2022-09-20 09:09:00; test1; yes
2022-09-21 05:00:00; test2; no

解决方案

1. 先把用户计划时间存成字典,方便快速查询

先读取用户列表,把每个用户的计划起止时间存入字典,这样后续处理请求记录时可以直接通过用户名快速获取时间,不用反复遍历用户列表,效率更高:

import csv
from datetime import datetime

# 统一解析时间字符串的函数
def parse_time(time_str):
    return datetime.strptime(time_str.strip(), "%Y-%m-%d %H:%M:%S")

# 构建用户计划时间字典
user_schedule = {}
with open('用户列表.csv', 'r', encoding='utf-8') as f:
    # 指定分隔符为;,自动识别表头
    reader = csv.DictReader(f, delimiter=';')
    for row in reader:
        user = row['user'].strip()
        start_time = parse_time(row['scheduled start date'])
        end_time = parse_time(row['scheduled end date'])
        user_schedule[user] = (start_time, end_time)

2. 逐行处理请求记录,填充Access Valid列

逐行读取请求记录,根据用户名查字典里的计划时间,调用校验函数赋值,然后写入新的CSV文件:

# 时间校验函数
def check_time(reported_time, scheduled_start, scheduled_end):
    return "yes" if scheduled_start <= reported_time <= scheduled_end else "no"

# 输入输出文件路径
input_request_file = '请求记录.csv'
output_request_file = '更新后的请求记录.csv'

with open(input_request_file, 'r', encoding='utf-8') as in_file, \
     open(output_request_file, 'w', encoding='utf-8', newline='') as out_file:
    
    reader = csv.DictReader(in_file, delimiter=';')
    # 保留原文件的表头
    fieldnames = reader.fieldnames
    writer = csv.DictWriter(out_file, fieldnames=fieldnames, delimiter=';')
    writer.writeheader()
    
    for row in reader:
        # 解析当前请求的时间和用户名
        request_time = parse_time(row['time'].strip())
        user = row['user'].strip()
        
        # 校验并赋值
        if user in user_schedule:
            start, end = user_schedule[user]
            row['Access Valid'] = check_time(request_time, start, end)
        else:
            # 没有计划的用户默认标记为无效
            row['Access Valid'] = "no"
        
        # 写入处理后的行
        writer.writerow(row)

3. 将处理后的CSV导出为Excel

CSV库本身不能直接生成Excel,用轻量的openpyxl库实现转换,适合处理大文件:

from openpyxl import Workbook

def csv_to_excel(csv_path, excel_path):
    wb = Workbook()
    ws = wb.active
    
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f, delimiter=';')
        for row in reader:
            # 清理每个字段的多余空格
            cleaned_row = [cell.strip() for cell in row]
            ws.append(cleaned_row)
    
    wb.save(excel_path)

# 调用函数生成Excel文件
csv_to_excel('更新后的请求记录.csv', 'updated_report.xlsx')

注意事项

  • 逐行处理CSV文件,内存占用极低,完美适配大文件场景
  • 用字典存储用户计划,查询效率为O(1),避免了大量重复遍历
  • 可以根据实际需求调整用户不存在时的默认值(比如改为"unknown")
  • 确保安装openpyxl库:pip install openpyxl

内容的提问来源于stack exchange,提问作者Smelt

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.17 09:35:17