CSV数据预处理逻辑错误：无法输出符合条件参与者的全部数据

问题描述 #

我正在开展CSV数据预处理工作，需求为：除满足多项筛选条件外，仅保留至少拥有3条符合条件数据的参与者的所有符合条件行。但当前代码（第9-16行及第37-40行）存在逻辑错误，无法输出此类参与者的第一条符合条件数据，需要排查修复。

原代码 #

import csv

def preprocessing(raw_filename, output_filename):
    data=open(raw_filename)
    new=open(output_filename, "a")
    health_write=csv.writer(new)
    reader=csv.DictReader(data)
    health_write.writerow(reader.fieldnames) 
    counting_part={}
    for row in reader:
        checker = False
        participant_id=row['participant_ID']
        # 错误点1：对所有行计数，而非仅符合条件的行
        if participant_id in counting_part: 
            counting_part[participant_id]+=1
        else: 
            counting_part[participant_id]=1
        x=row['timestamp'].split()[0]
        if x.split('/')[-1] == '2023':
            if float(row['accuracy']) <= 30:
                if len(participant_id.split('-'))==5:
                    match=[]
                    for each in participant_id.split('-'):
                        match.append(len(each))
                        # 错误点：ID分段长度判断放在循环内，逻辑混乱
                        if match == [8, 4, 4, 4, 12]:
                            for elm in each:
                                if elm.islower() or elm.isdigit():
                                    value_lat=float(row['double_latitude'])
                                    value_long=float(row['double_longitude'])
                                    if (value_lat >= -180 and value_lat
                                        <= 180 and value_long >= -180
                                        and value_long <=180):
                                        if 0<=float(row
                                                    ['double_altitude'])<=1000:
                                            if row['provider'] in ['gps',
                                                                   'network']:
                                                checker = True
                                                
        # 错误点2：仅当前行计数>=3时才写入，无法回溯早期符合条件的行
        if checker and counting_part[participant_id] >= 3: 
            for participant_key, count in counting_part.items():
                if participant_id == participant_key:
                    health_write.writerow(row.values())

    data.close()
    new.close()

preprocessing("example_raw.csv", "my_output.csv")

部分测试数据 #

row_id,timestamp,participant_ID,double_latitude,double_longitude,double_altitude,provider,accuracy
1,24/1/2023 1:05,2c9a7613-84f7-4c6d-af82,-37.80014035,144.963727,171.117,gps,41
2,24/1/2023 3:38,2c9a7613-84f7-4c6d-af82-9f54c8be2c1a,-37.80389128,144.9634838,143.159,gps,12.7445
3,24/1/2022 3:44,2c9a7613-84f7-4c6d-af82-9f54c8be2c1a,-37.80438867,144.9645387,19.95410156,gps,57.0303
4,24/1/2023 4:02,2c9a7613-84f7-4c6d-af82-9f54c8be2c1a,-37.80366035,144.9679344,56.83,gps,6.76879
5,24/1/2023 4:05,2c9a7613-84f7-4c6d-af82-9f54c8be2c1a,-37.80139125,144.9683195,55.13,gps,4.98383
6,15/7/2022 4:08,a7f3e14c-5db5-47e7-86c9-81f60e727d61,-37.80512032,144.9676758,4.723449707,gps,6.10221
7,15/7/2023 7:30,a7f3e14c-5db5-47e7-86c9-81f60e727d61,-37.80480437,144.9675091,95.83,gps,15

逻辑错误分析 #

1. 计数逻辑错误：
   原代码对所有行（包括不符合筛选条件的行）计数，导致counting_part存储的是参与者的总数据量，而非符合条件的数据量，判断是否保留时用总数量>=3，完全偏离需求。

2. 写入逻辑错误：
   原代码仅在处理当前行时，若计数>=3才写入。但参与者的第一条、第二条符合条件行在处理时计数分别为1、2，不满足>=3的条件，因此无法被写入，后续计数达标时也无法回溯写入之前的行。

3. 参与者ID校验逻辑错误：
   原代码在遍历ID分段时，每次添加一个长度就判断是否等于目标列表，导致只有第一个分段长度为8时才会进入后续判断，逻辑完全混乱，应该收集完所有分段长度后再统一判断。

修复后的代码 #

import csv

def preprocessing(raw_filename, output_filename):
    # 第一遍遍历：收集所有符合条件的行，并统计每个参与者的有效行数
    valid_rows = []
    participant_valid_count = {}
    
    with open(raw_filename, 'r') as data:
        reader = csv.DictReader(data)
        fieldnames = reader.fieldnames
        
        for row in reader:
            checker = False
            participant_id = row['participant_ID']
            
            # 执行所有筛选条件
            timestamp_year = row['timestamp'].split()[0].split('/')[-1]
            if timestamp_year == '2023':
                if float(row['accuracy']) <= 30:
                    id_segments = participant_id.split('-')
                    if len(id_segments) == 5:
                        # 检查ID分段长度是否符合要求
                        if [len(seg) for seg in id_segments] == [8,4,4,4,12]:
                            # 检查每个字符是否为小写字母或数字
                            is_valid_char = True
                            for seg in id_segments:
                                for char in seg:
                                    if not (char.islower() or char.isdigit()):
                                        is_valid_char = False
                                        break
                                if not is_valid_char:
                                    break
                            if is_valid_char:
                                lat = float(row['double_latitude'])
                                lon = float(row['double_longitude'])
                                if -180 <= lat <= 180 and -180 <= lon <= 180:
                                    altitude = float(row['double_altitude'])
                                    if 0 <= altitude <= 1000:
                                        if row['provider'] in ['gps', 'network']:
                                            checker = True
            
            if checker:
                valid_rows.append(row)
                # 仅对符合条件的行计数
                if participant_id in participant_valid_count:
                    participant_valid_count[participant_id] += 1
                else:
                    participant_valid_count[participant_id] = 1
    
    # 第二遍遍历：写入有效行数>=3的参与者的所有符合条件行
    with open(output_filename, 'w', newline='') as new:
        writer = csv.writer(new)
        writer.writerow(fieldnames)
        
        for row in valid_rows:
            pid = row['participant_ID']
            if participant_valid_count.get(pid, 0) >= 3:
                writer.writerow(row.values())

preprocessing("example_raw.csv", "my_output.csv")

修复说明 #

• 分两次遍历：第一次收集所有符合筛选条件的行并统计有效行数，第二次只写入有效行数>=3的参与者的所有行，确保不会遗漏早期符合条件的行。
• 修正计数逻辑：仅对符合筛选条件的行计数，保证统计的是有效数据量。
• 修复参与者ID校验逻辑：收集完所有分段长度后再统一判断，同时优化字符校验的循环结构，避免不必要的冗余判断。

内容的提问来源于stack exchange，提问作者Bluma Seline Cornett