本文目录导读：
我来详细介绍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
这些示例覆盖了常见的清理场景，根据你的具体需求选择合适的方案,并始终注意数据安全。
