Python实现智能磁盘监控并自动清理垃圾文件

import os
import shutil
from pathlib import Path
from collections import defaultdict
def get_disk_usage():
"""获取所有磁盘分区的使用情况"""
print("=" * 60)
print(f"{'盘符':<8}{'总容量':<12}{'已用':<12}{'可用':<12}{'使用率':<8}")
print("=" * 60)
results = []
for partition in os.popen("wmic logicaldisk get caption,size,freespace").readlines()[1:]:
parts = partition.strip().split()
if len(parts) < 3:
continue
drive = parts[0]
total = int(parts[1])  # 字节
free = int(parts[2])
used = total - free
usage_percent = (used / total * 100) if total > 0 else 0
# 格式化显示
total_gb = total / 1024**3
used_gb = used / 1024**3
free_gb = free / 1024**3
# 根据使用率选择显示标记
if usage_percent >= 90:
icon = "🔴"
elif usage_percent >= 80:
icon = "🟡"
else:
icon = "🟢"
print(f"{icon} {drive:<6}{total_gb:>8.1f} GB{used_gb:>8.1f} GB"
f"{free_gb:>8.1f} GB{usage_percent:>6.1f}%")
results.append({
"drive": drive,
"total_gb": total_gb,
"used_gb": used_gb,
"free_gb": free_gb,
"usage_percent": usage_percent,
})
return results
# 使用
# disks = get_disk_usage()

def scan_large_files(
directory,
min_size_mb=100,
top_n=20,
exclude_dirs=None
):
"""
扫描大文件
directory: 扫描目录
min_size_mb: 最小文件大小（MB）
top_n: 返回前 N 个
exclude_dirs: 排除的目录列表
"""
if exclude_dirs is None:
exclude_dirs = [
r"C:\Windows", r"C:\Program Files",
r"C:\Program Files (x86)",
]
large_files = []
min_size_bytes = min_size_mb * 1024 * 1024
print(f"正在扫描 {directory}（最小 {min_size_mb} MB）...")
for root, dirs, files in os.walk(directory):
# 排除目录
dirs[:] = [
d for d in dirs
if not any(
os.path.normpath(os.path.join(root, d)).startswith(
os.path.normpath(excl)
)
for excl in exclude_dirs
)
]
for file in files:
try:
filepath = os.path.join(root, file)
size = os.path.getsize(filepath)
if size >= min_size_bytes:
mtime = os.path.getmtime(filepath)
large_files.append({
"path": filepath,
"size_mb": round(size / 1024**2, 1),
"size_bytes": size,
"modified": mtime,
})
except (PermissionError, OSError):
continue
# 按大小排序
large_files.sort(key=lambda x: x["size_bytes"], reverse=True)
print(f"\n发现 {len(large_files)} 个大文件，TOP {top_n}：\n")
for f in large_files[:top_n]:
from datetime import datetime
mod_time = datetime.fromtimestamp(f["modified"]).strftime("%Y-%m-%d")
print(f"  {f['size_mb']:>8.1f} MB  {mod_time}  {f['path']}")
return large_files
# 使用：扫描 D 盘超过 500MB 的文件
# large_files = scan_large_files(r"D:\", min_size_mb=500, top_n=30)

def get_directory_sizes(
root_dir,
max_depth=2,
top_n=15,
exclude_dirs=None
):
"""
分析目录大小
max_depth: 递归深度
"""
if exclude_dirs is None:
exclude_dirs = []
dir_sizes = defaultdict(int)
print(f"正在分析 {root_dir} 的目录大小...")
for root, dirs, files in os.walk(root_dir):
# 计算当前深度
rel_path = os.path.relpath(root, root_dir)
depth = rel_path.count(os.sep) + (1 if rel_path != "." else 0)
if depth > max_depth:
dirs.clear()  # 不再深入
continue
# 排除目录
dirs[:] = [
d for d in dirs
if not any(os.path.join(root, d).startswith(excl) for excl in exclude_dirs)
]
# 统计当前目录下文件大小
current_size = 0
for file in files:
try:
current_size += os.path.getsize(os.path.join(root, file))
except (PermissionError, OSError):
pass
# 累加到各级父目录
if depth <= max_depth:
dir_sizes[root] += current_size
# 排序
sorted_dirs = sorted(dir_sizes.items(), key=lambda x: x[1], reverse=True)
print(f"\nTOP {top_n} 目录：\n")
total = sum(s for _, s in sorted_dirs)
for path, size in sorted_dirs[:top_n]:
size_gb = size / 1024**3
percent = (size / total * 100) if total > 0 else 0
# 进度条可视化
bar_len = 30
filled = int(bar_len * percent / 100)
bar = "█" * filled + "░" * (bar_len - filled)
print(f"  {size_gb:>8.2f} GB  {bar} {percent:>5.1f}%")
print(f"             {path}")
return sorted_dirs
# 使用
# dirs = get_directory_sizes(r"C:\Users", max_depth=2, exclude_dirs=[r"C:\Users\Default"])

import hashlib
def get_file_hash(filepath, block_size=65536):
"""计算文件的 MD5 哈希"""
hasher = hashlib.md5()
try:
with open(filepath, "rb") as f:
for block in iter(lambda: f.read(block_size), b""):
hasher.update(block)
return hasher.hexdigest()
except (PermissionError, OSError):
return None
def find_duplicate_files(directory, min_size_kb=100):
"""
找出重复文件（基于文件大小+MD5双重校验）
min_size_kb: 只检测大于此大小的文件
"""
# 第一步：按文件大小分组
print("第一步：按文件大小分组...")
size_groups = defaultdict(list)
for root, dirs, files in os.walk(directory):
for file in files:
try:
filepath = os.path.join(root, file)
size = os.path.getsize(filepath)
if size >= min_size_kb * 1024:
size_groups[size].append(filepath)
except (PermissionError, OSError):
continue
# 只保留有多个文件的组
potential_duplicates = {
size: files for size, files in size_groups.items()
if len(files) >= 2
}
print(f"发现 {len(potential_duplicates)} 组大小相同的文件")
# 第二步：对大小相同的文件计算 MD5
print("第二步：计算 MD5 确认重复...")
duplicates = {}  # hash -> [filepath, ...]
total_to_check = sum(len(files) for files in potential_duplicates.values())
checked = 0
for size, files in potential_duplicates.items():
for filepath in files:
file_hash = get_file_hash(filepath)
checked += 1
if file_hash:
if file_hash not in duplicates:
duplicates[file_hash] = []
duplicates[file_hash].append(filepath)
# 只保留真正重复的
true_duplicates = {
h: files for h, files in duplicates.items()
if len(files) >= 2
}
# 统计可释放空间
total_wasted = 0
for h, files in true_duplicates.items():
# 保留一个，其余都是浪费
size = os.path.getsize(files[0])
total_wasted += size * (len(files) - 1)
print(f"\n发现 {len(true_duplicates)} 组重复文件")
print(f"可释放空间: {total_wasted / 1024**3:.2f} GB\n")
# 按浪费空间排序
dup_sorted = sorted(
true_duplicates.items(),
key=lambda x: os.path.getsize(x[1][0]) * (len(x[1]) - 1),
reverse=True
)
for h, files in dup_sorted[:10]:
size = os.path.getsize(files[0])
wasted = size * (len(files) - 1)
print(f"  重复 {len(files)} 份, 各 {size/1024**2:.1f} MB, "
f"浪费 {wasted/1024**2:.1f} MB")
for f in files:
print(f"    {f}")
print()
return true_duplicates
def cleanup_duplicates(duplicate_map, dry_run=True):
"""
清理重复文件（保留每组中的第一个）
dry_run: True=只报告不删除
"""
total_freed = 0
for h, files in duplicate_map.items():
# 保留第一个，删除其余
keep = files[0]
remove = files[1:]
for filepath in remove:
if dry_run:
size = os.path.getsize(filepath)
total_freed += size
print(f"  [模拟删除] {filepath} ({size/1024**2:.1f} MB)")
else:
try:
size = os.path.getsize(filepath)
os.remove(filepath)
total_freed += size
print(f"  [已删除] {filepath} ({size/1024**2:.1f} MB)")
except Exception as e:
print(f"  [失败] {filepath}: {e}")
action = "可释放" if dry_run else "已释放"
print(f"\n{action}空间: {total_freed / 1024**3:.2f} GB")
if dry_run:
print("\n⚠️ 这是模拟运行，没有实际删除文件。")
print("确认无误后，设置 dry_run=False 重新运行。")
return total_freed
# 使用
# dups = find_duplicate_files(r"D:\Data", min_size_kb=1024)
# cleanup_duplicates(dups, dry_run=True)  # 先模拟
# cleanup_duplicates(dups, dry_run=False)  # 确认后删除

from datetime import datetime, timedelta
import re
class DiskCleaner:
"""智能磁盘清理器"""
# 常见可清理的目录和模式
CLEANUP_TARGETS = {
"Windows 临时文件": r"C:\Windows\Temp",
"用户临时文件": os.path.join(os.environ.get("TEMP", ""), ""),
"缩略图缓存": os.path.join(
os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Windows", "Explorer"
),
"Windows 更新缓存": r"C:\Windows\SoftwareDistribution\Download",
"Prefetch 缓存": r"C:\Windows\Prefetch",
}
# 可清理的文件模式
CLEANUP_PATTERNS = {
"日志文件": [r"\.log$"],
"临时文件": [r"\.tmp$", r"\.temp$", r"~\$"],
"缓存文件": [r"\.cache$", r"\.bak$"],
"缩略图数据库": [r"thumbcache_.*\.db$"],
}
# 可清理的目录模式
CLEANUP_DIR_PATTERNS = {
"node_modules": r"node_modules$",
"__pycache__": r"__pycache__$",
".pytest_cache": r"\.pytest_cache$",
"pip cache": r"pip[\\/]cache$",
}
def __init__(self, dry_run=True):
self.dry_run = dry_run
self.report = defaultdict(lambda: {"count": 0, "size": 0})
def clean_temp_directories(self):
"""清理系统临时目录"""
print("\n=== 清理临时目录 ===")
for name, path in self.CLEANUP_TARGETS.items():
if not os.path.exists(path):
continue
total_size = 0
file_count = 0
for item in os.listdir(path):
item_path = os.path.join(path, item)
try:
if os.path.isfile(item_path):
size = os.path.getsize(item_path)
total_size += size
file_count += 1
if not self.dry_run:
os.remove(item_path)
elif os.path.isdir(item_path):
size = sum(
os.path.getsize(os.path.join(dp, f))
for dp, dn, fn in os.walk(item_path)
for f in fn
)
total_size += size
file_count += 1
if not self.dry_run:
shutil.rmtree(item_path)
except (PermissionError, OSError):
pass
self.report[name]["count"] = file_count
self.report[name]["size"] = total_size
action = "可清理" if self.dry_run else "已清理"
if file_count > 0:
print(f"  {name}: {action} {file_count} 项 "
f"({total_size/1024/1024:.1f} MB)")
def clean_old_logs(self, days=30, log_dirs=None):
"""清理 N 天前的旧日志"""
print(f"\n=== 清理 {days} 天前的日志 ===")
if log_dirs is None:
log_dirs = [
r"C:\inetpub\logs",
r"C:\Windows\System32\winevt\Logs",
]
cutoff = time.time() - (days * 86400)
total_cleaned = 0
for log_dir in log_dirs:
if not os.path.exists(log_dir):
continue
for root, dirs, files in os.walk(log_dir):
for file in files:
if not file.endswith(".log"):
continue
filepath = os.path.join(root, file)
try:
mtime = os.path.getmtime(filepath)
if mtime < cutoff:
size = os.path.getsize(filepath)
total_cleaned += size
if self.dry_run:
from datetime import datetime as dt
mod_date = dt.fromtimestamp(mtime).strftime("%Y-%m-%d")
print(f"  [模拟] {filepath} ({size/1024/1024:.1f} MB, {mod_date})")
else:
os.remove(filepath)
except (PermissionError, OSError):
pass
action = "可释放" if self.dry_run else "已释放"
print(f"\n日志清理: {action} {total_cleaned/1024/1024:.1f} MB")
def clean_recycle_bin(self):
"""清空回收站"""
import ctypes
print("\n=== 清空回收站 ===")
if self.dry_run:
# 估算回收站大小
print("  [模拟] 清空回收站")
return
try:
# Windows API 清空回收站
result = ctypes.windll.shell32.SHEmptyRecycleBinW(None, None, 7)
if result == 0:
print("  回收站已清空")
else:
print(f"  清空回收站返回: {result}")
except Exception as e:
print(f"  清空回收站失败: {e}")
def run_full_cleanup(self):
"""执行完整清理"""
mode = "模拟" if self.dry_run else "实际"
print(f"{'=' * 60}")
print(f"  磁盘清理 ({mode}模式)")
print(f"{'=' * 60}")
self.clean_temp_directories()
self.clean_old_logs(days=30)
self.clean_recycle_bin()
# 汇总
total_size = sum(item["size"] for item in self.report.values())
print(f"\n{'=' * 60}")
action = "总计可释放" if self.dry_run else "总计已释放"
print(f"  {action}: {total_size / 1024 / 1024:.1f} MB")
print(f"{'=' * 60}")
if self.dry_run:
print("\n确认无误后，使用 DiskCleaner(dry_run=False) 执行实际清理")
return total_size
import time
# 使用示例
# cleaner = DiskCleaner(dry_run=True)   # 先模拟
# cleaner.run_full_cleanup()
# cleaner = DiskCleaner(dry_run=False)   # 确认后执行
# cleaner.run_full_cleanup()

import json
from datetime import datetime, timedelta
from pathlib import Path
class DiskMonitor:
"""磁盘空间持续监控器"""
def __init__(self, data_file="disk_history.json"):
self.data_file = Path(data_file)
self.history = self._load_history()
self.threshold_warning = 80  # 警告阈值 %
self.threshold_critical = 95  # 严重阈值 %
def _load_history(self):
"""加载历史数据"""
if self.data_file.exists():
with open(self.data_file, "r") as f:
return json.load(f)
return []
def _save_history(self):
"""保存历史数据"""
with open(self.data_file, "w") as f:
json.dump(self.history, f, ensure_ascii=False, indent=2)
def record_snapshot(self):
"""记录当前磁盘快照"""
snapshot = {
"timestamp": datetime.now().isoformat(),
"drives": [],
}
for partition in os.popen(
"wmic logicaldisk get caption,size,freespace"
).readlines()[1:]:
parts = partition.strip().split()
if len(parts) < 3:
continue
total = int(parts[1])
free = int(parts[2])
usage = ((total - free) / total * 100) if total > 0 else 0
snapshot["drives"].append({
"drive": parts[0],
"total_gb": round(total / 1024**3, 2),
"free_gb": round(free / 1024**3, 2),
"usage_percent": round(usage, 1),
})
self.history.append(snapshot)
# 只保留最近 90 天的数据
cutoff = (datetime.now() - timedelta(days=90)).isoformat()
self.history = [
h for h in self.history
if h["timestamp"] >= cutoff
]
self._save_history()
return snapshot
def check_alerts(self, snapshot=None):
"""检查是否需要告警"""
if not snapshot:
snapshot = self.record_snapshot()
alerts = []
for drive in snapshot["drives"]:
usage = drive["usage_percent"]
free_gb = drive["free_gb"]
if usage >= self.threshold_critical:
alerts.append({
"severity": "CRITICAL",
"drive": drive["drive"],
"message": (
f"{drive['drive']} 磁盘空间严重不足！"
f"已用 {usage}%，仅剩 {free_gb:.1f} GB"
),
})
elif usage >= self.threshold_warning:
alerts.append({
"severity": "WARNING",
"drive": drive["drive"],
"message": (
f"{drive['drive']} 磁盘空间不足警告："
f"已用 {usage}%，剩余 {free_gb:.1f} GB"
),
})
return alerts
def analyze_trend(self, drive_letter, days=30):
"""分析磁盘使用趋势"""
cutoff = (datetime.now() - timedelta(days=days)).isoformat()
recent = [
h for h in self.history
if h["timestamp"] >= cutoff
]
# 提取目标盘符的数据点
data_points = []
for h in recent:
for d in h["drives"]:
if d["drive"] == drive_letter:
data_points.append({
"time": h["timestamp"],
"usage": d["usage_percent"],
"free_gb": d["free_gb"],
})
break
if len(data_points) < 2:
print(f"数据不足（需要至少 2 个数据点，当前 {len(data_points)} 个）")
return None
# 计算趋势
first = data_points[0]
last = data_points[-1]
usage_change = last["usage"] - first["usage"]
free_change = last["free_gb"] - first["free_gb"]
days_span = len(data_points)  # 近似天数
daily_usage_increase = usage_change / days_span if days_span > 0 else 0
daily_free_decrease = abs(free_change) / days_span if days_span > 0 else 0
# 预计何时达到阈值
if daily_usage_increase > 0:
days_to_warning = (self.threshold_warning - last["usage"]) / daily_usage_increase
days_to_critical = (self.threshold_critical - last["usage"]) / daily_usage_increase
else:
days_to_warning = float('inf')
days_to_critical = float('inf')
report = {
"drive": drive_letter,
"period": f"最近 {days} 天",
"data_points": len(data_points),
"current_usage": f"{last['usage']}%",
"current_free": f"{last['free_gb']} GB",
"usage_change": f"+{usage_change:.1f}%" if usage_change > 0 else f"{usage_change:.1f}%",
"free_change": f"{free_change:.1f} GB",
"daily_increase": f"+{daily_usage_increase:.2f}%" if daily_usage_increase > 0 else "稳定",
"days_to_warning": f"{days_to_warning:.0f} 天" if days_to_warning < 365 else "无风险",
"days_to_critical": f"{days_to_critical:.0f} 天" if days_to_critical < 365 else "无风险",
}
print(f"\n=== {drive_letter} 磁盘使用趋势 ({report['period']}) ===")
print(f"  当前使用率: {report['current_usage']}, 剩余: {report['current_free']}")
print(f"  变化: 使用率 {report['usage_change']}, 空间 {report['free_change']}")
print(f"  日均增长: {report['daily_increase']}")
print(f"  预计触发警告: {report['days_to_warning']}")
print(f"  预计磁盘满: {report['days_to_critical']}")
return report
# 使用示例
# monitor = DiskMonitor()
# monitor.record_snapshot()  # 记录快照
# alerts = monitor.check_alerts()
# monitor.analyze_trend("C:", days=30)