如何在Python数据查看器中实现CSV文件导入及去重功能

需求说明 #

• 导入包含销售额、年、月、日字段的CSV文件，将数据整合到程序的销售数据列表中
• 维护一个文本记录文件，记录所有已导入的文件名，防止重复导入

现有代码 #

import sales
import csv

# 多余的初始化读取可移除
# with open('all_sales.csv', newline='') as f:
#     data = list(csv.reader(f))

def display_welcome_message():
    print("Welcome to data viewer program")
    
def display_menu():
    print("enter sales data")
    print()
    print("COMMAND MENU")
    print("View  - View All Sales")
    print("add   - Add sales")
    print("Menu  - Show menu")
    print("Import - Import file")
    print("Exit  - Exit Program")
    print()
 
total_amount=0
FILENAME = "all_sales.csv"
# 新增：记录已导入文件的日志文件名
IMPORTED_FILES_LOG = "imported_files.txt"
 
def read_sales():
    all_sales = []
    with open(FILENAME, newline="") as file:
        reader = csv.reader(file)
        for row in reader:
            all_sales.append(row)
    return all_sales
 
def write_sales(all_sales):
    with open(FILENAME, "w", newline="") as file:
        writer = csv.writer(file)
        writer.writerows(all_sales)
            
def add(all_sales):
    amount1 = sales.get_amount()
    year1 = sales.get_year()
    month1 = sales.get_month()
    day1 = sales.get_day()
 
    add_sales = []
    add_sales.append(month1)
    add_sales.append(day1)
    add_sales.append(year1)
    add_sales.append(amount1)
    all_sales.append(add_sales)
    write_sales(all_sales)
    print(f"{month1} - {day1} - {year1} was added")
 
# 待完善的导入函数
def importfile(all_sales):
    print("Enter a file name to import: ")
 
def view(all_sales):
    i = 0
    total = 0
    if len(all_sales) == 0:
        print("there are no sales data in the list. \n")
        return
    print(f"              Date               Quarter     Amount")
    print("---------------------------------------------------------")
    while i < len(all_sales):
        # 实时计算季度
        month = int(all_sales[i][0])
        quarter = (month - 1) // 3 + 1
        print(f"{i+1}.          {all_sales[i][2]}-{all_sales[i][0]}-{all_sales[i][1]}     Q{quarter}          ${all_sales[i][3]}")
        total += float(all_sales[i][3])
        i += 1            
    print("---------------------------------------------------------")
    print(f"TOTAL:                                       ${total:.2f}")
 
def main():
    display_welcome_message()
    display_menu()
    # 初始化读取已有销售数据
    sales_list = read_sales()
 
    while True:
        command = input('enter command: ').strip().lower()
        if command == 'view':
            view(sales_list)
        elif command == 'add':
            add(sales_list)
        elif command == 'menu':
            display_menu()
        elif command == 'import':
            importfile(sales_list)
        elif command == 'exit':
            break
        else:
            print("not a valid command. Please try again. \n")
    print("Bye!")
 
if __name__ == "__main__":
    main()

完善后的导入功能实现 #

步骤1：添加已导入文件日志的读写函数 #

def get_imported_files():
    """读取已导入的文件列表"""
    imported = set()
    try:
        with open(IMPORTED_FILES_LOG, 'r') as f:
            for line in f:
                filename = line.strip()
                if filename:
                    imported.add(filename)
    except FileNotFoundError:
        # 日志文件不存在则返回空集合
        pass
    return imported

def log_imported_file(filename):
    """将已导入文件名写入日志"""
    with open(IMPORTED_FILES_LOG, 'a') as f:
        f.write(f"{filename}\n")

步骤2：完善importfile核心函数 #

def importfile(all_sales):
    filename = input("Import file name:  ").strip()
    if not filename:
        print("文件名不能为空")
        return
    
    # 校验是否重复导入
    imported_files = get_imported_files()
    if filename in imported_files:
        print(f"文件 {filename} 已导入过，无需重复操作")
        return
    
    # 读取并解析待导入CSV
    try:
        with open(filename, newline='') as f:
            reader = csv.reader(f)
            # 转换CSV格式：原CSV为「销售额、年、月、日」→ 程序存储格式「月、日、年、销售额」
            valid_data = []
            for row in reader:
                if len(row) != 4:
                    print(f"跳过无效行：{row}")
                    continue
                try:
                    amount, year, month, day = [item.strip() for item in row]
                    valid_data.append([month, day, year, amount])
                except ValueError:
                    print(f"数据格式错误，跳过行：{row}")
                    continue
            
            if not valid_data:
                print("未读取到有效销售数据")
                return
            
            # 合并数据并持久化
            all_sales.extend(valid_data)
            write_sales(all_sales)
            # 记录导入日志
            log_imported_file(filename)
            print(f"成功导入 {len(valid_data)} 条销售数据")
    except FileNotFoundError:
        print(f"未找到目标文件：{filename}")
    except Exception as e:
        print(f"导入失败：{str(e)}")

关键逻辑说明 #

1. 重复校验：用集合存储已导入文件名，实现O(1)时间复杂度的重复判断
2. 格式兼容：将导入CSV的字段顺序转换为程序统一存储格式，保证数据一致性
3. 异常处理：覆盖文件不存在、数据格式错误、空行等场景，提升程序鲁棒性
4. 数据合并：用extend批量添加导入数据，避免多次IO操作

内容的提问来源于stack exchange，提问作者kste27