Disk Manager

#!/usr/bin/env python3
"""
Archive a project folder while excluding recreatable heavy directories.
"""

import argparse
import json
import tarfile
import time
from pathlib import Path

EXCLUDES = {"node_modules", ".next", "dist", "build", "__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache"}


def should_exclude(path: Path) -> bool:
    return any(part in EXCLUDES for part in path.parts)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--workspace', default='/data/workspace')
    parser.add_argument('--project', required=True, help='project folder relative to workspace')
    parser.add_argument('--output-dir', default='/data/workspace/output/archives')
    args = parser.parse_args()

    workspace = Path(args.workspace)
    project = workspace / args.project
    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    if not project.exists() or not project.is_dir():
        raise SystemExit(f'Project not found: {project}')

    ts = time.strftime('%Y%m%d-%H%M%S')
    archive_path = output_dir / f'{project.name}-{ts}.tar.gz'

    with tarfile.open(archive_path, 'w:gz') as tf:
        for item in project.rglob('*'):
            rel = item.relative_to(workspace)
            if should_exclude(rel):
                continue
            tf.add(item, arcname=str(rel), recursive=False)

    print(json.dumps({
        "status": "ok",
        "project": str(project),
        "archive": str(archive_path),
        "excluded_names": sorted(list(EXCLUDES)),
    }, indent=2))


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Safe cleanup runner for disk-manager skill.

Removes cache/recreatable artifacts from workspace with dry-run support.
Human mode includes ASCII summary bars.
"""

import argparse
import json
import os
import shutil
import time
from pathlib import Path

SAFE_REMOVE_NAMES = {"__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache"}
SAFE_FILE_SUFFIXES = {".pyc", ".pyo"}


def dir_size(path: Path) -> int:
    total = 0
    for root, _, files in os.walk(path, onerror=lambda e: None):
        for f in files:
            fp = Path(root) / f
            try:
                total += fp.stat().st_size
            except Exception:
                pass
    return total


def collect_targets(workspace: Path):
    targets = []
    for root, dirs, files in os.walk(workspace, onerror=lambda e: None):
        root_path = Path(root)

        for d in list(dirs):
            if d in SAFE_REMOVE_NAMES:
                p = root_path / d
                targets.append({
                    "path": str(p),
                    "kind": "dir",
                    "reason": f"safe-cache:{d}",
                    "size_bytes": dir_size(p),
                })

        for f in files:
            p = root_path / f
            if p.suffix in SAFE_FILE_SUFFIXES:
                try:
                    size = p.stat().st_size
                except Exception:
                    size = 0
                targets.append({
                    "path": str(p),
                    "kind": "file",
                    "reason": f"bytecode:{p.suffix}",
                    "size_bytes": size,
                })

    targets.sort(key=lambda x: x["size_bytes"], reverse=True)
    return targets


def apply_cleanup(targets, dry_run: bool):
    cleaned = []
    errors = []
    for t in targets:
        p = Path(t["path"])
        if not p.exists():
            continue

        if dry_run:
            cleaned.append({**t, "status": "would_remove"})
            continue

        try:
            if t["kind"] == "dir":
                shutil.rmtree(p)
            else:
                p.unlink(missing_ok=True)
            cleaned.append({**t, "status": "removed"})
        except Exception as e:
            errors.append({**t, "status": "error", "error": str(e)})

    return cleaned, errors


def fmt_bytes(n: int) -> str:
    units = ["B", "KB", "MB", "GB", "TB"]
    v = float(max(0, n))
    for u in units:
        if v < 1024 or u == units[-1]:
            return f"{v:.1f} {u}" if u != "B" else f"{int(v)} B"
        v /= 1024
    return f"{n} B"


def bar(part: int, whole: int, width: int = 24) -> str:
    if whole <= 0:
        filled = 0
    else:
        filled = int(round((part / whole) * width))
    filled = max(0, min(width, filled))
    return "█" * filled + "░" * (width - filled)


def print_human(report: dict):
    before = report["before_bytes"]
    freed = report["freed_bytes"]
    mode = "DRY RUN" if report["dry_run"] else "EXECUTED"

    print(f"SAFE CLEANUP ({mode})")
    print(f"Workspace: {report['workspace']}")
    print(f"Before: {fmt_bytes(before)}")
    if report["dry_run"]:
        print(f"Estimated freed: {fmt_bytes(freed)}")
    else:
        print(f"After:  {fmt_bytes(report['after_bytes'])}")
        print(f"Freed:  {fmt_bytes(freed)}")

    pct = (freed / before * 100) if before else 0
    print(f"Impact: [{bar(freed, before)}] {pct:5.2f}%")
    print(f"Targets: {report['targets_count']} | Cleaned: {report['cleaned_count']} | Errors: {report['errors_count']}")

    top = report.get("cleaned", [])[:10]
    if top:
        print("Top cleaned targets")
        for t in top:
            print(f"- {fmt_bytes(t.get('size_bytes', 0)):>9}  {t.get('path')}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--workspace', default='/data/workspace')
    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('--json', action='store_true', help='print JSON report')
    args = parser.parse_args()

    workspace = Path(args.workspace)
    before = dir_size(workspace)

    targets = collect_targets(workspace)
    cleaned, errors = apply_cleanup(targets, args.dry_run)

    after = dir_size(workspace)
    freed = max(0, before - after) if not args.dry_run else sum(t["size_bytes"] for t in targets)

    report = {
        "workspace": str(workspace),
        "generated_at": int(time.time()),
        "dry_run": args.dry_run,
        "before_bytes": before,
        "after_bytes": after,
        "freed_bytes": freed,
        "targets_count": len(targets),
        "cleaned_count": len(cleaned),
        "errors_count": len(errors),
        "cleaned": cleaned,
        "errors": errors,
    }

    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print_human(report)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Workspace disk scanner for disk-manager skill.

Outputs JSON with categorized cleanup opportunities.
Human mode includes ASCII visualization bars.
"""

import argparse
import json
import os
import time
from pathlib import Path

WORKSPACE = Path('/data/workspace')

SAFE_DIR_NAMES = {"__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache", ".cache"}
RECREATABLE_DIR_NAMES = {"node_modules", "dist", "build", ".next", "target"}


def dir_size_bytes(path: Path) -> int:
    total = 0
    try:
        for root, _, files in os.walk(path, onerror=lambda e: None):
            for f in files:
                fp = Path(root) / f
                try:
                    total += fp.stat().st_size
                except Exception:
                    pass
    except Exception:
        pass
    return total


def top_level_usage(workspace: Path):
    items = []
    for p in workspace.iterdir():
        if p.name.startswith('.'):
            continue
        try:
            size = dir_size_bytes(p) if p.is_dir() else p.stat().st_size
            items.append({"path": str(p), "size_bytes": size})
        except Exception:
            continue
    items.sort(key=lambda x: x["size_bytes"], reverse=True)
    return items


def find_dirs_by_name(workspace: Path, names):
    found = []
    for root, dirs, _ in os.walk(workspace, onerror=lambda e: None):
        root_path = Path(root)
        for d in list(dirs):
            if d in names:
                dp = root_path / d
                size = dir_size_bytes(dp)
                try:
                    mtime = int(dp.stat().st_mtime)
                except Exception:
                    mtime = 0
                found.append({"path": str(dp), "size_bytes": size, "mtime": mtime})
    found.sort(key=lambda x: x["size_bytes"], reverse=True)
    return found


def find_large_files(workspace: Path, min_mb: int):
    threshold = min_mb * 1024 * 1024
    found = []
    for root, _, files in os.walk(workspace, onerror=lambda e: None):
        for f in files:
            fp = Path(root) / f
            try:
                st = fp.stat()
            except Exception:
                continue
            if st.st_size >= threshold:
                found.append({
                    "path": str(fp),
                    "size_bytes": st.st_size,
                    "mtime": int(st.st_mtime),
                })
    found.sort(key=lambda x: x["size_bytes"], reverse=True)
    return found


def find_old_files(root: Path, older_than_days: int):
    cutoff = time.time() - older_than_days * 86400
    if not root.exists():
        return []
    found = []
    for r, _, files in os.walk(root, onerror=lambda e: None):
        for f in files:
            fp = Path(r) / f
            try:
                st = fp.stat()
            except Exception:
                continue
            if st.st_mtime < cutoff:
                found.append({
                    "path": str(fp),
                    "size_bytes": st.st_size,
                    "mtime": int(st.st_mtime),
                })
    found.sort(key=lambda x: x["size_bytes"], reverse=True)
    return found


def summarize_size(items):
    return sum(i.get("size_bytes", 0) for i in items)


def fmt_bytes(n: int) -> str:
    units = ["B", "KB", "MB", "GB", "TB"]
    v = float(max(0, n))
    for u in units:
        if v < 1024 or u == units[-1]:
            return f"{v:.1f} {u}" if u != "B" else f"{int(v)} B"
        v /= 1024
    return f"{n} B"


def bar(part: int, whole: int, width: int = 24) -> str:
    if whole <= 0:
        filled = 0
    else:
        filled = int(round((part / whole) * width))
    filled = max(0, min(width, filled))
    return "█" * filled + "░" * (width - filled)


def print_human(report: dict, top_n: int = 8):
    totals = report["totals"]
    total = totals["top_level_total_bytes"]
    safe = totals["safe_clean_bytes"]
    rec = totals["recreatable_bytes"]
    large = totals["large_files_bytes"]
    old = totals["old_output_bytes"]

    print("DISK MANAGER SCAN")
    print(f"Workspace: {report['workspace']}")
    print(f"Total scanned: {fmt_bytes(total)} ({total} bytes)")
    print()

    rows = [
        ("Safe to clean", safe),
        ("Recreatable", rec),
        ("Large files", large),
        ("Old output", old),
    ]

    print("Category view")
    for label, value in rows:
        pct = (value / total * 100) if total else 0
        print(f"- {label:<13} [{bar(value, total)}] {pct:5.1f}%  {fmt_bytes(value)}")
    print()

    print(f"Top {top_n} heavy paths")
    for item in report["top_level"][:top_n]:
        value = item["size_bytes"]
        pct = (value / total * 100) if total else 0
        print(f"- [{bar(value, total, width=18)}] {pct:5.1f}%  {fmt_bytes(value):>9}  {item['path']}")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--workspace', default=str(WORKSPACE))
    parser.add_argument('--large-file-mb', type=int, default=100)
    parser.add_argument('--old-days', type=int, default=30)
    parser.add_argument('--json', action='store_true', help='print JSON only')
    args = parser.parse_args()

    workspace = Path(args.workspace)

    top = top_level_usage(workspace)
    safe_dirs = find_dirs_by_name(workspace, SAFE_DIR_NAMES)
    recreatable_dirs = find_dirs_by_name(workspace, RECREATABLE_DIR_NAMES)
    large_files = find_large_files(workspace, args.large_file_mb)
    old_output = find_old_files(workspace / 'output', args.old_days)

    report = {
        "workspace": str(workspace),
        "generated_at": int(time.time()),
        "totals": {
            "top_level_total_bytes": summarize_size(top),
            "safe_clean_bytes": summarize_size(safe_dirs),
            "recreatable_bytes": summarize_size(recreatable_dirs),
            "large_files_bytes": summarize_size(large_files),
            "old_output_bytes": summarize_size(old_output),
        },
        "top_level": top[:25],
        "safe_to_clean": safe_dirs,
        "recreatable": recreatable_dirs,
        "large_files": large_files,
        "old_output": old_output,
    }

    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print_human(report)


if __name__ == '__main__':
    main()