scanner.py

"""Recursive folder scanner for FolderVisualizer."""

from __future__ import annotations

import os
from dataclasses import dataclass, field

from models import FileNode, ScanResult


@dataclass
class _ScanState:
    """Mutable counters shared by recursive scan calls."""

    file_count: int = 0
    directory_count: int = 0
    errors: list[str] = field(default_factory=list)


def _error_message(action: str, path: str, error: OSError) -> str:
    detail = error.strerror or str(error)
    return f'{action} "{path}": {detail}'


def _scan_directory(path: str, state: _ScanState) -> FileNode:
    node = FileNode(
        name=os.path.basename(path) or path,
        path=path,
        is_directory=True,
    )
    state.directory_count += 1

    try:
        with os.scandir(path) as iterator:
            entries = sorted(iterator, key=lambda entry: entry.name.casefold())
    except PermissionError as error:
        message = _error_message("无权限访问目录", path, error)
        node.error = message
        state.errors.append(message)
        return node
    except FileNotFoundError as error:
        message = _error_message("扫描时目录已消失", path, error)
        node.error = message
        state.errors.append(message)
        return node
    except OSError as error:
        message = _error_message("无法扫描目录", path, error)
        node.error = message
        state.errors.append(message)
        return node

    for entry in entries:
        entry_path = os.path.abspath(entry.path)

        try:
            is_directory = entry.is_dir(follow_symlinks=False)
        except FileNotFoundError as error:
            state.errors.append(_error_message("扫描时项目已消失", entry_path, error))
            continue
        except OSError as error:
            state.errors.append(_error_message("无法读取项目", entry_path, error))
            continue

        if is_directory:
            child = _scan_directory(entry_path, state)
        else:
            try:
                size = entry.stat(follow_symlinks=False).st_size
                child = FileNode(
                    name=entry.name,
                    path=entry_path,
                    is_directory=False,
                    size=size,
                )
            except FileNotFoundError as error:
                state.errors.append(
                    _error_message("扫描时文件已消失", entry_path, error)
                )
                continue
            except OSError as error:
                message = _error_message("无法读取文件", entry_path, error)
                child = FileNode(
                    name=entry.name,
                    path=entry_path,
                    is_directory=False,
                    error=message,
                )
                state.errors.append(message)

            state.file_count += 1

        node.children.append(child)
        node.size += child.size

    return node


def scan_folder(path: str) -> ScanResult:
    """Recursively scan *path* and return its tree and aggregate statistics.

    Directory counts include the root directory. Symbolic links and directory
    junctions are not followed, preventing recursive cycles.
    """

    normalized_path = os.path.abspath(os.path.expanduser(path.strip()))
    if not os.path.exists(normalized_path):
        raise FileNotFoundError(f'文件夹不存在: "{normalized_path}"')
    if not os.path.isdir(normalized_path):
        raise NotADirectoryError(f'路径不是文件夹: "{normalized_path}"')

    state = _ScanState()
    root = _scan_directory(normalized_path, state)
    return ScanResult(
        root=root,
        total_size=root.size,
        file_count=state.file_count,
        directory_count=state.directory_count,
        errors=state.errors,
    )