utils.py

from __future__ import annotations

import re
import time
from pathlib import Path


_slug_re = re.compile(r"[^a-zA-Z0-9]+")


def slugify(value: str) -> str:
    cleaned = _slug_re.sub("_", value.strip().lower()).strip("_")
    return cleaned or "file"


def ensure_dir(path: str | Path) -> Path:
    directory = Path(path)
    directory.mkdir(parents=True, exist_ok=True)
    return directory


def unique_path(path: Path) -> Path:
    if not path.exists():
        return path

    stem = path.stem
    suffix = path.suffix
    parent = path.parent
    counter = 2
    while True:
        candidate = parent / f"{stem}_{counter}{suffix}"
        if not candidate.exists():
            return candidate
        counter += 1


def _can_open_exclusive(path: Path) -> bool:
    """Windowsでコピー中のファイルは排他オープンできない。開けるかどうかで書き込み完了を判定する。"""
    try:
        with path.open("rb"):
            return True
    except OSError:
        return False


def wait_until_file_ready(path: Path, checks: int, interval_sec: float, timeout_sec: float) -> bool:
    start = time.time()
    stable_count = 0
    last_size = -1

    while time.time() - start < timeout_sec:
        if not path.exists() or not path.is_file():
            time.sleep(interval_sec)
            continue

        # Windowsではコピー中ファイルを開けないので、開けるまで待つ
        if not _can_open_exclusive(path):
            stable_count = 0
            last_size = -1
            time.sleep(interval_sec)
            continue

        current_size = path.stat().st_size
        if current_size > 0 and current_size == last_size:
            stable_count += 1
            if stable_count >= checks:
                return True
        else:
            stable_count = 0
            last_size = current_size
        time.sleep(interval_sec)
    return False