58 lines
1.6 KiB
Python
58 lines
1.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""ファイル名衝突回避コピー・移動、ファイルサイズ安定待ちモジュール。"""
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
|
|
def resolve_unique_path(dest_dir: Path, stem: str, suffix: str) -> Path:
|
|
candidate = dest_dir / f"{stem}{suffix}"
|
|
if not candidate.exists():
|
|
return candidate
|
|
|
|
counter = 2
|
|
while True:
|
|
candidate = dest_dir / f"{stem}({counter}){suffix}"
|
|
if not candidate.exists():
|
|
return candidate
|
|
counter += 1
|
|
|
|
|
|
def copy_with_unique_name(src: Path, dest_dir: Path, stem: str, suffix: str) -> Path:
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_path = resolve_unique_path(dest_dir, stem, suffix)
|
|
shutil.copy2(str(src), str(dest_path))
|
|
return dest_path
|
|
|
|
|
|
def move_with_unique_name(src: Path, dest_dir: Path, stem: str, suffix: str) -> Path:
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_path = resolve_unique_path(dest_dir, stem, suffix)
|
|
shutil.move(str(src), str(dest_path))
|
|
return dest_path
|
|
|
|
|
|
def wait_until_stable(
|
|
path: Path,
|
|
interval_sec: float,
|
|
retries: int,
|
|
sleep_func: Callable[[float], None] = time.sleep,
|
|
) -> bool:
|
|
if not path.exists():
|
|
return False
|
|
|
|
previous_size = path.stat().st_size
|
|
for _ in range(retries):
|
|
sleep_func(interval_sec)
|
|
if not path.exists():
|
|
return False
|
|
current_size = path.stat().st_size
|
|
if current_size == previous_size:
|
|
return True
|
|
previous_size = current_size
|
|
|
|
return False
|