Python 自动化监控大文件

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
large_files_scanner.py
Usage: python large_files_scanner.py
"""

import os
from pathlib import Path
from datetime import datetime, timedelta
import humanize

def scan_large_files(
    root_dir: str,
    size_threshold: int = 100 * 1024 * 1024,  # 100 MB
    days_old: int | None = None,
    exclude_dirs: set[str] | None = None,
    output_file: str | None = None,
) -> list[dict]:
    """
    扫描 root_dir 下满足条件的大文件。

    参数
    ----
    root_dir : 扫描根目录，支持 ~
    size_threshold : 文件大小阈值（单位：字节）
    days_old : 只扫描最后修改时间在 N 天前的文件
    exclude_dirs : 需要跳过的子目录名集合，如 {'.git', 'node_modules'}
    output_file : 若提供，则将结果写入该路径

    返回
    ----
    字典列表，每项包含 path/size/modified
    """
    root_path = Path(root_dir).expanduser().resolve()
    if not root_path.is_dir():
        raise NotADirectoryError(root_path)

    exclude_dirs = exclude_dirs or set()
    cutoff_ts = (datetime.now() - timedelta(days=days_old)).timestamp() if days_old else None
    results: list[dict] = []

    for item in root_path.rglob("*"):
        if item.is_file():
            # 跳过排除目录
            if any(part in exclude_dirs for part in item.relative_to(root_path).parts):
                continue

            stat = item.stat()
            size, mtime = stat.st_size, stat.st_mtime
            if size >= size_threshold and (cutoff_ts is None or mtime < cutoff_ts):
                results.append(
                    {
                        "path": str(item),
                        "size": humanize.naturalsize(size),
                        "modified": datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S"),
                    }
                )

    # 控制台输出
    print(f"\n扫描完成，共发现 {len(results)} 个大文件\n")
    for idx, f in enumerate(results, 1):
        print(f"{idx:>3}. {f['size']:<10}  {f['modified']}  {f['path']}")

    # 写入报告
    if output_file:
        output_path = Path(output_file).expanduser()
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with output_path.open("w", encoding="utf-8") as fp:
            fp.write("大文件扫描报告\n")
            fp.write(f"生成时间 : {datetime.now():%F %T}\n")
            fp.write(f"扫描目录 : {root_path}\n")
            fp.write(f"大小阈值 : {humanize.naturalsize(size_threshold)}\n")
            if days_old:
                fp.write(f"时间阈值 : 修改时间早于 {days_old} 天前\n")
            fp.write("-" * 80 + "\n")
            for f in results:
                fp.write(f"{f['size']:<12}  {f['modified']}  {f['path']}\n")
        print(f"\n报告已保存至 → {output_path}")

    return results


if __name__ == "__main__":
    import sys

    root = input("扫描目录 [默认为当前目录]: ").strip() or "."
    try:
        size_mb = int(input("大小阈值 MB [100]: ") or "100")
    except ValueError:
        print("输入无效，使用默认 100 MB")
        size_mb = 100

    try:
        days = input("仅扫描修改时间早于 N 天的文件 [无限制]: ").strip()
        days = int(days) if days else None
    except ValueError:
        print("输入无效，将不限制修改时间")
        days = None

    out = input("报告保存路径 [直接回车仅控制台输出]: ").strip() or None

    scan_large_files(
        root_dir=root,
        size_threshold=size_mb * 1024 * 1024,
        days_old=days,
        exclude_dirs={".git", "node_modules", "__pycache__"},
        output_file=out,
    )