Python脚本如何清理旧结构废弃冗余数据

wen python案例 2026-06-25 256

本文目录导读：

通用的数据清理策略
基于文件模式的清理
数据库冗余数据清理
检查和处理重复数据
综合清理类
使用示例
安全注意事项

我来详细介绍Python清理旧结构废弃冗余数据的几种方法：

通用的数据清理策略

基础清理函数模板

import os
import shutil
from datetime import datetime, timedelta
from pathlib import Path
def clean_old_data(data_path, retention_days=30):
    """
    清理指定路径下超过保留期的旧数据
    Args:
        data_path: 数据路径
        retention_days: 数据保留天数
    """
    path = Path(data_path)
    cutoff_date = datetime.now() - timedelta(days=retention_days)
    cleaned_count = 0
    for item in path.iterdir():
        if item.is_file() or item.is_dir():
            try:
                # 检查文件的最后修改时间
                mtime = datetime.fromtimestamp(item.stat().st_mtime)
                if mtime < cutoff_date:
                    if item.is_file():
                        item.unlink()
                    else:
                        shutil.rmtree(item)
                    cleaned_count += 1
                    print(f"删除: {item}")
            except Exception as e:
                print(f"无法删除 {item}: {e}")
    print(f"共清理 {cleaned_count} 个项目")
    return cleaned_count

基于文件模式的清理

import glob
import re
from pathlib import Path
def clean_by_pattern(data_dir, patterns, max_age_days=90):
    """
    根据文件模式清理旧数据
    Args:
        data_dir: 数据目录
        patterns: 文件模式列表，如 ['*.tmp', '*.log', 'backup_*']
        max_age_days: 最大保留天数
    """
    import time
    cutoff_time = time.time() - (max_age_days * 86400)  # 86400秒/天
    for pattern in patterns:
        # 使用 glob 匹配文件
        for file_path in glob.glob(os.path.join(data_dir, pattern)):
            try:
                # 检查文件修改时间
                mtime = os.path.getmtime(file_path)
                if mtime < cutoff_time:
                    os.remove(file_path)
                    print(f"已删除旧文件: {file_path}")
            except Exception as e:
                print(f"处理文件 {file_path} 时出错: {e}")

数据库冗余数据清理

import sqlite3
from datetime import datetime, timedelta
def clean_database_old_data(db_path, tables_config):
    """
    清理数据库中旧的冗余数据
    Args:
        db_path: 数据库路径
        tables_config: 表配置，格式：
        {
            'table_name': {
                'date_column': 'created_at',
                'retention_days': 30,
                'additional_where': 'status="completed"'  # 可选
            }
        }
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cleaned_records = {}
    for table_name, config in tables_config.items():
        date_column = config.get('date_column', 'created_at')
        retention_days = config.get('retention_days', 30)
        cutoff_date = datetime.now() - timedelta(days=retention_days)
        # 构建删除SQL
        where_clauses = [f"{date_column} < '{cutoff_date}'"]
        if 'additional_where' in config:
            where_clauses.append(config['additional_where'])
        where_str = ' AND '.join(where_clauses)
        try:
            # 先查询受影响行数
            count_sql = f"SELECT COUNT(*) FROM {table_name} WHERE {where_str}"
            cursor.execute(count_sql)
            count = cursor.fetchone()[0]
            # 执行删除
            delete_sql = f"DELETE FROM {table_name} WHERE {where_str}"
            cursor.execute(delete_sql)
            conn.commit()
            cleaned_records[table_name] = count
            print(f"表 {table_name}: 清理了 {count} 条旧记录")
        except sqlite3.Error as e:
            print(f"清理表 {table_name} 时出错: {e}")
            conn.rollback()
    conn.close()
    return cleaned_records

检查和处理重复数据

import hashlib
from collections import defaultdict
def find_and_remove_duplicates(data_dir, recursive=True):
    """
    查找并删除重复文件
    Args:
        data_dir: 数据目录
        recursive: 是否递归搜索子目录
    """
    def get_file_hash(filepath, chunk_size=8192):
        """计算文件哈希值"""
        hasher = hashlib.md5()
        with open(filepath, 'rb') as f:
            while chunk := f.read(chunk_size):
                hasher.update(chunk)
        return hasher.hexdigest()
    # 收集所有文件信息
    hash_map = defaultdict(list)
    total_size_saved = 0
    duplicates_removed = 0
    if recursive:
        file_iterator = Path(data_dir).rglob('*')
    else:
        file_iterator = Path(data_dir).glob('*')
    for file_path in file_iterator:
        if file_path.is_file():
            try:
                file_hash = get_file_hash(file_path)
                hash_map[file_hash].append(file_path)
            except Exception as e:
                print(f"无法读取文件 {file_path}: {e}")
    # 保留第一个文件，删除其余重复文件
    for file_hash, file_list in hash_map.items():
        if len(file_list) > 1:
            # 保留最新的文件（按修改时间）
            file_list.sort(key=lambda x: x.stat().st_mtime, reverse=True)
            keep_file = file_list[0]
            for duplicate_file in file_list[1:]:
                try:
                    size = duplicate_file.stat().st_size
                    duplicate_file.unlink()
                    total_size_saved += size
                    duplicates_removed += 1
                    print(f"删除重复文件: {duplicate_file} (保留: {keep_file})")
                except Exception as e:
                    print(f"无法删除 {duplicate_file}: {e}")
    print(f"共删除 {duplicates_removed} 个重复文件，节省 {total_size_saved / 1024 / 1024:.2f} MB")
    return duplicates_removed, total_size_saved

综合清理类

import json
import logging
from datetime import datetime
class DataCleaner:
    """综合数据清理器"""
    def __init__(self, config_path=None):
        self.logger = logging.getLogger(__name__)
        self.config = self._load_config(config_path) if config_path else {}
    def _load_config(self, config_path):
        """加载清理配置"""
        try:
            with open(config_path, 'r') as f:
                return json.load(f)
        except Exception as e:
            self.logger.error(f"加载配置文件失败: {e}")
            return {}
    def run_cleanup(self, dry_run=False):
        """
        执行清理操作
        Args:
            dry_run: 是否仅模拟运行（不实际删除）
        """
        self.logger.info(f"开始清理... (dry_run={dry_run})")
        results = []
        # 1. 清理过期文件
        if 'file_cleanup' in self.config:
            for path_config in self.config['file_cleanup']:
                result = self._clean_file_path(path_config, dry_run)
                results.append(result)
        # 2. 清理数据库
        if 'database_cleanup' in self.config:
            for db_config in self.config['database_cleanup']:
                result = self._clean_database(db_config, dry_run)
                results.append(result)
        # 3. 清理重复文件
        if 'deduplication' in self.config:
            result = self._remove_duplicates(self.config['deduplication'], dry_run)
            results.append(result)
        return results
    def _clean_file_path(self, config, dry_run):
        """清理文件路径"""
        path = Path(config['path'])
        patterns = config.get('patterns', ['*'])
        retention_days = config.get('retention_days', 30)
        cleaned_count = 0
        cutoff_date = datetime.now() - timedelta(days=retention_days)
        for pattern in patterns:
            for file_path in path.glob(pattern):
                if file_path.is_file():
                    mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
                    if mtime < cutoff_date:
                        if not dry_run:
                            file_path.unlink()
                        cleaned_count += 1
        return {'type': 'file', 'cleaned': cleaned_count, 'path': str(path)}
    def _clean_database(self, config, dry_run):
        """清理数据库"""
        # 实现数据库清理
        pass
    def _remove_duplicates(self, config, dry_run):
        """删除重复文件"""
        # 实现重复文件删除
        pass
    def generate_report(self, results):
        """生成清理报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {},
            'details': results
        }
        total_cleaned = sum(r.get('cleaned', 0) for r in results)
        report['summary']['total_cleaned'] = total_cleaned
        return report

使用示例

# 1. 基础使用 - 清理旧文件
clean_old_data('/path/to/data', retention_days=30)
# 2. 按模式清理
clean_by_pattern('/tmp', ['*.tmp', '*.log', 'cache_*'], max_age_days=7)
# 3. 清理数据库
db_config = {
    'logs': {
        'date_column': 'created_at',
        'retention_days': 90
    },
    'temp_data': {
        'date_column': 'expire_date',
        'retention_days': 7,
        'additional_where': 'status="expired"'
    }
}
clean_database_old_data('my_database.db', db_config)
# 4. 删除重复文件
find_and_remove_duplicates('/path/to/duplicate/dir')
# 5. 综合清理
cleaner = DataCleaner('cleanup_config.json')
results = cleaner.run_cleanup(dry_run=False)
report = cleaner.generate_report(results)
print(json.dumps(report, indent=2))

安全注意事项

def safe_cleanup(paths, dry_run=True, backup=False):
    """
    安全的清理函数，包含保护机制
    Args:
        paths: 要清理的路径列表
        dry_run: 是否模拟运行
        backup: 是否在删除前备份
    """
    # 保护目录列表（绝对不能清理的目录）
    PROTECTED_DIRS = [
        '/', '/etc', '/usr', '/bin', '/sbin',
        '/boot', '/dev', '/proc', '/sys'
    ]
    for path_str in paths:
        path = Path(path_str).resolve()
        # 安全检查：防止删除系统关键目录
        for protected in PROTECTED_DIRS:
            if str(path).startswith(protected):
                print(f"⛔ 安全警告：跳过保护目录 {path}")
                continue
        # 确认操作
        if not dry_run:
            response = input(f"确认清理 {path}? (yes/no): ")
            if response.lower() not in ['yes', 'y']:
                print("跳过清理")
                continue
        # 清理前备份
        if backup and not dry_run:
            backup_path = path.parent / f"{path.name}.backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            shutil.copytree(path, backup_path)
            print(f"备份创建于: {backup_path}")
        # 执行清理
        if not dry_run:
            # 清理逻辑
            pass

这些示例覆盖了常见的清理场景，根据你的具体需求选择合适的方案,并始终注意数据安全。

本文地址： https://www.bmjup.com/post/8605.html