第12讲 · 清洗并分析成绩数据

清洗并分析成绩数据

把虚构成绩文本清洗成结构化记录，再完成统计与异常检查。

1读取原始成绩文件

2清洗并保存有效记录

3完成统计并生成报告

处理前后对比

处理前：空行、字段缺失、分数不是数字 · 处理后：保留4条有效记录，识别2条异常记录

处理前 · scores.txt

小林,Python,88

格式错误
小郑,AI协作,优秀
小周,数据整理,76
小吴,Python,59
小陈,Python,94

处理后 · score_report.txt

成绩分析报告
有效记录：4
异常记录：2
平均分：79.25
最高分：小陈 94
不及格：小吴 59

同一份数据经过读取、清洗、结构化和统计，最终生成可以直接验收的分析报告。

查看并逐行读取成绩文件

read_score_lines.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

path = prepare_scores()
line_number = 0
with open(path, "r", encoding="utf-8") as file:
    for line in file:
        line = line.strip()
        if not line:
            continue
        line_number += 1
        print(f"第{line_number}条：{line}")

with open(path, "r", encoding="utf-8") as file:: 以UTF-8编码读取文本
for line in file:: 逐行遍历文件
line = line.strip(): 清理每行首尾空白
if not line: continue: 跳过空行

1先观察原始文件中的正常与异常记录

2open以只读模式打开文件

3for按顺序取得每一行

4strip清理空白并跳过空行

拆分一行成绩数据

parse_score_row.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

line = "小林,Python,88"
parts = line.split(",")
name, subject, score = parts
score = int(score)
print("字段列表：", parts)
print("姓名：", name)
print("科目：", subject)
print("分数：", score)
print("分数类型：", type(score).__name__)

parts = line.split(","): 按英文逗号拆分
name, subject, score = parts: 解包三个字段
score = int(score): 转换分数类型

1split返回字符串列表

2解包按顺序接收三个字段

3int把分数字符串转换为整数

检查成绩字段数量

check_score_fields.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

rows = ["小林,Python,88", "格式错误", "小周,数据整理,76"]
invalid_count = 0
for line in rows:
    parts = line.split(",")
    if len(parts) != 3:
        invalid_count += 1
        print("跳过字段不完整：", line)
        continue
    print("字段完整：", parts)
print("字段异常：", invalid_count)

if len(parts) != 3:: 判断字段数量是否完整
invalid_count += 1: 累计异常记录
continue: 跳过当前记录

1len取得字段数量

2三个字段才是完整记录

3continue跳过当前异常行

检查分数是否有效

check_score_value.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

rows = ["小林,Python,88", "小郑,AI协作,优秀", "小周,数据整理,76"]
valid_scores = []
for line in rows:
    name, subject, score = line.split(",")
    try:
        score = int(score)
    except ValueError:
        print("跳过无效分数：", line)
        continue
    valid_scores.append(score)
    print("有效分数：", name, score)
print("有效分数数量：", len(valid_scores))

try:: 执行可能失败的类型转换
score = int(score): 尝试得到整数分数
except ValueError:: 捕获转换失败

1try尝试转换分数

2ValueError表示内容不能转成整数

3except只处理对应的转换失败

保存结构化成绩记录

build_score_records.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

name, subject, score = "小林", "Python", 88
record = {"name": name, "subject": subject, "score": score}
records = []
records.append(record)
print("记录数量：", len(records))
print("第一条记录：", records[0])
print("姓名与分数：", records[0]["name"], records[0]["score"])

record = {"name": name, "subject": subject, "score": score}: 创建一条命名清楚的记录
records.append(record): 加入有效记录列表
record["score"]: 按键读取整数分数

1字典保存一条记录

2列表保存多条记录

3append只加入通过校验的数据

计算总分、平均分与最高分

calculate_score_statistics.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

records = [
    {"name": "小林", "score": 88},
    {"name": "小周", "score": 76},
    {"name": "小吴", "score": 59},
    {"name": "小陈", "score": 94},
]
total = sum(record["score"] for record in records)
average = total / len(records)
highest = max(records, key=lambda record: record["score"])
print("总分：", total)
print(f"平均分：{average:.2f}")
print("最高分：", highest["name"], highest["score"])

sum(record["score"] for record in records): 汇总全部分数
total / len(records): 计算平均分
max(records, key=lambda record: record["score"]): 取得最高分记录

1sum计算总分

2len提供有效记录数量

3max返回最高分整条记录

4格式化平均分保留两位小数

统计各科目记录数

count_score_subjects.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

records = [
    {"subject": "Python"}, {"subject": "数据整理"},
    {"subject": "Python"}, {"subject": "Python"},
]
subject_counts = {}
for record in records:
    subject = record["subject"]
    subject_counts[subject] = subject_counts.get(subject, 0) + 1
for subject in sorted(subject_counts):
    print(subject, subject_counts[subject])

subject_counts = {}: 准备空计数字典
subject_counts.get(subject, 0) + 1: 读取旧次数并加一
sorted(subject_counts): 稳定输出科目顺序

1科目名作为字典键

2出现次数作为字典值

3get为第一次出现的科目提供零起点

筛选不及格成绩记录

filter_failing_scores.py

import os

WORKSPACE = "/tmp/lesson12_workspace"
SOURCE = "小林,Python,88\n\n格式错误\n小郑,AI协作,优秀\n小周,数据整理,76\n小吴,Python,59\n小陈,Python,94"

def prepare_scores():
    os.makedirs(WORKSPACE, exist_ok=True)
    path = os.path.join(WORKSPACE, "scores.txt")
    with open(path, "w", encoding="utf-8") as fixture:
        fixture.write(SOURCE)
    return path

records = [
    {"name": "小林", "subject": "Python", "score": 88},
    {"name": "小周", "subject": "数据整理", "score": 76},
    {"name": "小吴", "subject": "Python", "score": 59},
]
failing = []
for record in records:
    if record["score"] < 60:
        failing.append(record)
for record in failing:
    print(record["name"], record["subject"], record["score"])

if record["score"] < 60:: 判断是否低于及格线
failing.append(record): 保存命中的完整记录

1遍历全部有效记录

2分数低于60时命中

3保留完整记录用于报告

生成成绩分析报告

score_analyzer.py

records, invalid_count = load_records(path)
report_path = os.path.join(WORKSPACE, "score_report.txt")
with open(report_path, "w", encoding="utf-8") as file:
    file.write(report + "\n")

总结成绩分析项目

1输入：读取UTF-8文本文件

2清洗：跳过空行，检查字段与分数

3结构化：用字典列表保存有效记录

4分析：完成总体、分类与条件统计

5输出：生成并打开成绩分析报告

语法拆解

read → split: 读取并拆分原始数据
validate → append: 校验后保存统一记录
aggregate → write: 统计并写出报告文件