_sync.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  1. from __future__ import annotations
  2. import logging
  3. import os
  4. import shutil
  5. import sys
  6. from stat import S_IWUSR
  7. from typing import TYPE_CHECKING
  8. if TYPE_CHECKING:
  9. from pathlib import Path
  10. LOGGER = logging.getLogger(__name__)
  11. def ensure_dir(path: Path) -> None:
  12. if not path.exists():
  13. LOGGER.debug("create folder %s", path)
  14. os.makedirs(str(path))
  15. def ensure_safe_to_do(src: Path, dest: Path) -> None:
  16. if src == dest:
  17. msg = f"source and destination is the same {src}"
  18. raise ValueError(msg)
  19. if not dest.exists():
  20. return
  21. if dest.is_dir() and not dest.is_symlink():
  22. LOGGER.debug("remove directory %s", dest)
  23. safe_delete(dest)
  24. else:
  25. LOGGER.debug("remove file %s", dest)
  26. dest.unlink()
  27. def symlink(src: Path, dest: Path) -> None:
  28. ensure_safe_to_do(src, dest)
  29. LOGGER.debug("symlink %s", _Debug(src, dest))
  30. dest.symlink_to(src, target_is_directory=src.is_dir())
  31. def copy(src: Path, dest: Path) -> None:
  32. ensure_safe_to_do(src, dest)
  33. is_dir = src.is_dir()
  34. method = copytree if is_dir else shutil.copy
  35. LOGGER.debug("copy %s", _Debug(src, dest))
  36. method(str(src), str(dest))
  37. def copytree(src: str, dest: str) -> None:
  38. for root, _, files in os.walk(src):
  39. dest_dir = os.path.join(dest, os.path.relpath(root, src))
  40. if not os.path.isdir(dest_dir):
  41. os.makedirs(dest_dir)
  42. for name in files:
  43. src_f = os.path.join(root, name)
  44. dest_f = os.path.join(dest_dir, name)
  45. shutil.copy(src_f, dest_f)
  46. def safe_delete(dest: Path) -> None:
  47. def onerror(func: object, path: str, exc_info: object) -> None: # noqa: ARG001
  48. if not os.access(path, os.W_OK):
  49. os.chmod(path, S_IWUSR)
  50. func(path) # ty: ignore[call-non-callable]
  51. else:
  52. raise # noqa: PLE0704
  53. if sys.version_info >= (3, 12):
  54. shutil.rmtree(str(dest), ignore_errors=True, onexc=onerror)
  55. else:
  56. shutil.rmtree(str(dest), ignore_errors=True, onerror=onerror)
  57. class _Debug:
  58. def __init__(self, src: Path, dest: Path) -> None:
  59. self.src = src
  60. self.dest = dest
  61. def __str__(self) -> str:
  62. return f"{'directory ' if self.src.is_dir() else ''}{self.src!s} to {self.dest!s}"
  63. __all__ = [
  64. "copy",
  65. "copytree",
  66. "ensure_dir",
  67. "safe_delete",
  68. "symlink",
  69. ]