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
|
# coding=utf-8
import errno
import os
import shutil
import contextlib
import library.python.fs as lpf
def replace_in_file(path, old, new):
"""
Replace text occurrences in a file
:param path: path to the file
:param old: text to replace
:param new: replacement
"""
with open(path) as fp:
content = fp.read()
lpf.ensure_removed(path)
with open(path, 'w') as fp:
fp.write(content.replace(old, new))
@contextlib.contextmanager
def change_dir(path):
old = os.getcwd()
try:
os.chdir(path)
yield path
finally:
os.chdir(old)
def copytree(src, dst, symlinks=False, ignore=None, postprocessing=None):
'''
Copy an entire directory of files into an existing directory
instead of raising Exception what shtuil.copytree does
'''
if not os.path.exists(dst) and os.path.isdir(src):
os.makedirs(dst)
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
if postprocessing:
postprocessing(dst, False)
for root, dirs, files in os.walk(dst):
for path in dirs:
postprocessing(os.path.join(root, path), False)
for path in files:
postprocessing(os.path.join(root, path), True)
def get_unique_file_path(dir_path, file_pattern, create_file=True, max_suffix=10000):
def atomic_file_create(path):
try:
fd = os.open(path, os.O_CREAT | os.O_EXCL, 0o644)
os.close(fd)
return True
except OSError as e:
if e.errno in [errno.EEXIST, errno.EISDIR, errno.ETXTBSY]:
return False
# Access issue with file itself, not parent directory.
if e.errno == errno.EACCES and os.path.exists(path):
return False
raise e
def atomic_dir_create(path):
try:
os.mkdir(path)
return True
except OSError as e:
if e.errno == errno.EEXIST:
return False
raise e
file_path = os.path.join(dir_path, file_pattern)
lpf.ensure_dir(os.path.dirname(file_path))
file_counter = 0
handler = atomic_file_create if create_file else atomic_dir_create
while os.path.exists(file_path) or not handler(file_path):
file_path = os.path.join(dir_path, file_pattern + ".{}".format(file_counter))
file_counter += 1
assert file_counter < max_suffix
return file_path
|
