import os
root = "/data/upload"
for name in os.listdir(root):
    path = os.path.join(root, name)
    print(path)

from pathlib import Path
import os

def scan_one_level(folder: str):
    base = Path(folder)
    if not base.exists():
        print(f"[bad_path] not exists: {base}")
        return
    if not base.is_dir():
        print(f"[bad_path] not dir: {base}")
        return
    # 上下文管理器自動(dòng)釋放資源
    with os.scandir(base) as items:
        for item in items:
            # 禁止跟隨軟鏈接，防止死循環(huán)
            if item.is_dir(follow_symlinks=False):
                print(f"[dir ] {item.path}")
            elif item.is_file(follow_symlinks=False):
                print(f"[file] {item.path}")
            else:
                print(f"[skip] {item.path}")

scan_one_level("/data/upload")

from pathlib import Path
import os

def walk_files(root: str, max_depth: int = 20):
    start = Path(root).resolve()
    stack = [(start, 0)]
    while stack:
        folder, depth = stack.pop()
        # 超出設(shè)定深度直接跳過
        if depth > max_depth:
            print(f"[skip_depth] {folder}")
            continue
        try:
            with os.scandir(folder) as entries:
                for entry in entries:
                    path = Path(entry.path)
                    try:
                        # 目錄入棧實(shí)現(xiàn)遞歸，不跟隨軟鏈接
                        if entry.is_dir(follow_symlinks=False):
                            stack.append((path, depth + 1))
                            continue
                        # 文件通過yield逐個(gè)返回，惰性加載不占內(nèi)存
                        if entry.is_file(follow_symlinks=False):
                            yield path
                    except OSError as e:
                        print(f"[entry_error] path={path} err={e}")
        except PermissionError as e:
            # 無權(quán)限目錄記錄日志，繼續(xù)遍歷其他目錄
            print(f"[no_permission] path={folder} err={e}")
        except FileNotFoundError:
            # 遍歷瞬間目錄被業(yè)務(wù)刪除，常見于動(dòng)態(tài)上傳目錄
            print(f"[gone] path={folder}")
        except OSError as e:
            print(f"[scan_error] path={folder} err={e}")

for file_path in walk_files("/data/upload"):
    print(file_path)

def find_big_files(root: str, limit_mb: int = 100):
    limit = limit_mb * 1024 * 1024
    for path in walk_files(root):
        try:
            size = path.stat().st_size
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        if size >= limit:
            print(f"[big_file] size_mb={size / 1024 / 1024:.1f} path={path}")

find_big_files("/data/upload", 100)

def iter_by_suffix(root: str, suffixes: set[str]):
    suffixes = {s.lower() for s in suffixes}
    for path in walk_files(root):
        if path.suffix.lower() in suffixes:
            yield path

# 只遍歷.log .txt文件
for log_file in iter_by_suffix("/data/app", {".log", ".txt"}):
    print(f"[match] {log_file}")

from collections import defaultdict

def summary_folder(root: str):
    counter = defaultdict(int)
    size_map = defaultdict(int)
    for path in walk_files(root):
        suffix = path.suffix.lower() or "<no_ext>"
        try:
            size = path.stat().st_size
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        counter[suffix] += 1
        size_map[suffix] += size
    # 按占用空間倒序輸出
    rows = sorted(size_map.items(), key=lambda x: x[1], reverse=True)
    for suffix, total_size in rows:
        print(f"{suffix:10s} count={counter[suffix]:6d} size_mb={total_size / 1024 / 1024:10.2f}")

summary_folder("/data/upload")

import time
def delete_old_tmp(root: str, days: int = 30, dry_run: bool = True):
    now = time.time()
    expire_seconds = days * 24 * 3600
    for path in iter_by_suffix(root, {".tmp", ".bak"}):
        try:
            mtime = path.stat().st_mtime
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        if now - mtime <= expire_seconds:
            continue
        if dry_run:
            print(f"[dry_run] delete {path}")
            continue
        try:
            path.unlink()
            print(f"[deleted] {path}")
        except OSError as e:
            print(f"[delete_error] path={path} err={e}")
# 先試運(yùn)行，確認(rèn)輸出無誤再改為False
delete_old_tmp("/data/upload/tmp", 30, dry_run=True)