-
Notifications
You must be signed in to change notification settings - Fork 0
/
copy_files.py
94 lines (74 loc) · 2.77 KB
/
copy_files.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
#!/usr/bin/env python
'''
io_utils - contains scripts for read/write of main files
'''
import os
import shutil
import glob
#*********************************************
def copy_tree(source: str, destination: str, diagnostics: bool = False) -> None:
"""
Perform local copy from networked storage to working area
(e.g. GWS to /work/scratch )
Automatically wipes and clobbers
:param str source: source directory
:param str destination: destination directory
:param bool diagnostics: verbose output
"""
# remove entire directory
if os.path.exists(destination):
try:
shutil.rmtree(destination)
except OSError:
# already removed
pass
# don't need to make as copytree will do
if diagnostics:
print(f"{destination} removed")
# copy entire tree
shutil.copytree(source, destination)
# ensure update of timestamps
for root, diry, files in os.walk(destination):
for fname in files:
os.utime(os.path.join(root, fname), None)
if diagnostics:
print(f"copied {source} to {destination}")
return # copy_tree
#*********************************************
def copy_files(source: str, destination: str, extension:str = "",
clobber:bool = True, wipe: bool = True, diagnostics: bool = False) -> None:
"""
Perform local copy from networked storage to working area
(e.g. GWS/file.txt to /work/scratch/file.txt )
:param str source: source directory
:param str destination: destination directory
:param bool clobber: overwrite
:param bool wipe: clean out destination in advance of copy
:param str extension: optional filename extension
:param bool diagnostics: verbose output
"""
# remove entire directory and recreate as blank
if wipe:
shutil.rmtree(destination)
if not os.path.exists(destination):
os.mkdir(destination)
# for each file at a time
for filename in glob.glob(fr'{os.path.expanduser(source)}*{extension}'):
if not os.path.exists(os.path.join(destination, filename.split("/")[-1])):
# file doesn't exist, so copy
shutil.copy(filename, destination)
if diagnostics:
print(filename)
else:
# file exists
if clobber:
# overwrite
shutil.copy(filename, destination)
if diagnostics:
print(filename)
else:
if diagnostics:
print(f"{filename} exists")
# force update of timestamps
os.utime(os.path.join(destination, filename.split("/")[-1]), None)
return # copy_files