Python脚本如何快速修复项目数据问题

wen python案例 2026-06-24 30

本文目录导读：

先诊断问题类型
常见问题快速修复
一键修复脚本
特定场景修复
使用建议

针对Python脚本快速修复项目数据问题,可以按照以下步骤和常用方法操作：

先诊断问题类型

# 快速诊断数据问题
import pandas as pd
import numpy as np
def diagnose_data(df):
    """快速诊断数据问题"""
    report = {}
    # 1. 缺失值检查
    report['missing_values'] = df.isnull().sum()
    # 2. 重复值检查
    report['duplicates'] = df.duplicated().sum()
    # 3. 数据类型检查
    report['dtypes'] = df.dtypes
    # 4. 异常值检查（数值列）
    num_cols = df.select_dtypes(include=[np.number]).columns
    for col in num_cols:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        outliers = df[(df[col] < Q1 - 1.5*IQR) | (df[col] > Q3 + 1.5*IQR)]
        if len(outliers) > 0:
            report[f'{col}_outliers'] = len(outliers)
    return report

常见问题快速修复

缺失值处理

def fix_missing_values(df, strategy='auto'):
    """快速修复缺失值"""
    if strategy == 'auto':
        for col in df.columns:
            if df[col].dtype in ['object', 'category']:
                df[col].fillna(df[col].mode()[0] if not df[col].mode().empty else 'Unknown', inplace=True)
            else:
                df[col].fillna(df[col].median(), inplace=True)
    elif strategy == 'drop':
        df.dropna(inplace=True)
    elif strategy == 'interpolate':
        df.interpolate(method='linear', inplace=True)
    return df

重复数据处理

def fix_duplicates(df, subset=None, keep='first'):
    """快速修复重复数据"""
    # 找出重复行
    duplicates = df[df.duplicated(subset=subset, keep=False)]
    if len(duplicates) > 0:
        print(f"发现 {len(duplicates)} 行重复数据")
        # 去重
        df_clean = df.drop_duplicates(subset=subset, keep=keep)
        print(f"去重后剩余 {len(df_clean)} 行")
        return df_clean
    return df

数据类型修正

def fix_dtypes(df):
    """快速修正数据类型"""
    for col in df.columns:
        # 尝试转换为数值类型
        if df[col].dtype == 'object':
            try:
                df[col] = pd.to_numeric(df[col], errors='coerce')
                print(f"列 {col} 转换为数值类型")
            except:
                pass
        # 日期格式转换
        if 'date' in col.lower() or 'time' in col.lower():
            try:
                df[col] = pd.to_datetime(df[col], errors='coerce')
                print(f"列 {col} 转换为日期类型")
            except:
                pass
    return df

异常值处理

def fix_outliers(df, method='clip'):
    """快速处理异常值"""
    num_cols = df.select_dtypes(include=[np.number]).columns
    for col in num_cols:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5*IQR
        upper_bound = Q3 + 1.5*IQR
        if method == 'clip':
            df[col] = df[col].clip(lower_bound, upper_bound)
        elif method == 'remove':
            df = df[(df[col] >= lower_bound) & (df[col] <= upper_bound)]
        elif method == 'median':
            median_val = df[col].median()
            df.loc[(df[col] < lower_bound) | (df[col] > upper_bound), col] = median_val
    return df

一键修复脚本

def quick_fix_data(data_path, output_path=None):
    """一键修复数据问题"""
    # 读取数据
    if data_path.endswith('.csv'):
        df = pd.read_csv(data_path)
    elif data_path.endswith('.xlsx'):
        df = pd.read_excel(data_path)
    else:
        raise ValueError("不支持的格式")
    print("="*50)
    print("数据修复开始...")
    print(f"原始数据: {df.shape}")
    # 1. 诊断问题
    report = diagnose_data(df)
    print("\n诊断报告:", report)
    # 2. 修复缺失值
    df = fix_missing_values(df)
    print("缺失值已修复")
    # 3. 修复重复值
    df = fix_duplicates(df)
    print("重复值已处理")
    # 4. 修正数据类型
    df = fix_dtypes(df)
    print("数据类型已修正")
    # 5. 处理异常值
    df = fix_outliers(df)
    print("异常值已处理")
    print(f"\n修复后数据: {df.shape}")
    print("="*50)
    # 保存结果
    if output_path:
        if output_path.endswith('.csv'):
            df.to_csv(output_path, index=False)
        elif output_path.endswith('.xlsx'):
            df.to_excel(output_path, index=False)
        print(f"数据已保存到: {output_path}")
    return df
# 使用示例
# df = quick_fix_data('data.csv', 'fixed_data.csv')

特定场景修复

# 文本数据清洗
def clean_text_data(df, text_columns):
    """清洗文本列"""
    for col in text_columns:
        if col in df.columns:
            # 去除多余空格
            df[col] = df[col].str.strip()
            # 统一大小写
            df[col] = df[col].str.lower()
            # 去除特殊字符
            df[col] = df[col].str.replace(r'[^\w\s]', '', regex=True)
    return df
# ID/关联键修复
def fix_primary_key(df, key_column):
    """修复主键问题"""
    # 检查空值
    null_keys = df[key_column].isnull().sum()
    if null_keys > 0:
        print(f"发现 {null_keys} 个空主键")
        # 生成唯一标识
        max_id = df[key_column].max()
        for idx in df[df[key_column].isnull()].index:
            max_id += 1
            df.at[idx, key_column] = max_id
    return df

使用建议

备份数据：修改前先备份原始数据
增量修复：每次只修复特定类型的问题
日志记录：记录修复操作和影响的行数
单元测试：编写测试验证修复逻辑
批量处理：使用chunksize处理大文件

# 安全修复示例
def safe_fix(data_path):
    import shutil
    # 1. 创建备份
    backup_path = data_path.replace('.', '_backup.')
    shutil.copy2(data_path, backup_path)
    print(f"备份已创建: {backup_path}")
    # 2. 分步修复
    df = pd.read_csv(data_path, chunksize=10000)
    # 3. 记录日志
    log_entries = []
    for i, chunk in enumerate(df):
        original_shape = chunk.shape
        chunk = fix_missing_values(chunk)
        log_entries.append(f"Chunk {i}: {original_shape} -> {chunk.shape}")
    # 4. 保存日志
    with open('fix_log.txt', 'w') as f:
        f.write('\n'.join(log_entries))
    return log_entries

根据具体的数据问题和项目需求,选择合适的修复方法组合使用。