1
|
|
|
"""File operations packages. Provide functions to interact with the filesystem. |
2
|
|
|
|
3
|
|
|
.. Authors: |
4
|
|
|
Philippe Dessauw |
5
|
|
|
[email protected] |
6
|
|
|
|
7
|
|
|
.. Sponsor: |
8
|
|
|
Alden Dima |
9
|
|
|
[email protected] |
10
|
|
|
Information Systems Group |
11
|
|
|
Software and Systems Division |
12
|
|
|
Information Technology Laboratory |
13
|
|
|
National Institute of Standards and Technology |
14
|
|
|
http://www.nist.gov/itl/ssd/is |
15
|
|
|
""" |
16
|
|
|
from zipfile import ZipFile |
17
|
|
|
from os import walk, remove, makedirs |
18
|
|
|
from os.path import join, splitext, exists |
19
|
|
|
from shutil import rmtree |
20
|
|
|
from hashlib import sha256 |
21
|
|
|
|
22
|
|
|
module_conf = { |
23
|
|
|
"zip_ext": ".zip" |
24
|
|
|
} |
25
|
|
|
|
26
|
|
|
|
27
|
|
|
def zip_directory(directory): |
28
|
|
|
"""Zip a directory |
29
|
|
|
|
30
|
|
|
Parameters: |
31
|
|
|
directory (str): Path to the directory to zip. |
32
|
|
|
|
33
|
|
|
Returns: |
34
|
|
|
str: Path to the archive. |
35
|
|
|
""" |
36
|
|
|
archive_name = directory + module_conf["zip_ext"] |
37
|
|
|
zip_dir = ZipFile(archive_name, "w") |
38
|
|
|
|
39
|
|
|
for root, folders, files in walk(directory): |
40
|
|
|
for item in folders+files: |
41
|
|
|
orig_path = join(root, item) |
42
|
|
|
dest_path = orig_path[len(directory):] |
43
|
|
|
|
44
|
|
|
zip_dir.write(orig_path, dest_path) |
45
|
|
|
|
46
|
|
|
rmtree(directory) # Clean directory |
47
|
|
|
return archive_name |
48
|
|
|
|
49
|
|
|
|
50
|
|
|
def unzip_directory(archive): |
51
|
|
|
"""Unzip an archive. |
52
|
|
|
|
53
|
|
|
Parameters: |
54
|
|
|
archive (str): Path to the archive to unzip. |
55
|
|
|
|
56
|
|
|
Returns: |
57
|
|
|
str: Path to the directory. |
58
|
|
|
""" |
59
|
|
|
zip_dir = ZipFile(archive, "r") |
60
|
|
|
directory = splitext(archive)[0] |
61
|
|
|
|
62
|
|
|
zip_dir.extractall(directory) |
63
|
|
|
remove(archive) |
64
|
|
|
|
65
|
|
|
return directory |
66
|
|
|
|
67
|
|
|
|
68
|
|
|
def create_directories(dir_conf, prefix=None): |
69
|
|
|
"""Create application directories and subdirectories given a configuration dictionary. |
70
|
|
|
|
71
|
|
|
Parameters: |
72
|
|
|
dir_conf (str): List of directories to create. |
73
|
|
|
prefix (str): Root directory for the directories to create. Default to `None` (directories will be built in |
74
|
|
|
the current directory). |
75
|
|
|
|
76
|
|
|
Raises: |
77
|
|
|
ValueError: If there is a subdirectory with no root or if the subdirectory key is not a dictionary. |
78
|
|
|
""" |
79
|
|
|
dirnames = [d for d in dir_conf.values() if isinstance(d, str)] |
80
|
|
|
|
81
|
|
|
for dirname in dirnames: |
82
|
|
|
dirpath = join(prefix, dirname) if prefix is not None else dirname |
83
|
|
|
|
84
|
|
|
if not exists(dirpath): |
85
|
|
|
makedirs(dirpath) |
86
|
|
|
|
87
|
|
|
dir_keys = dir_conf.keys() |
88
|
|
|
roots = [d for d in dir_keys if d.endswith("_root") and d.split("_root")[0] in dir_keys] |
89
|
|
|
dir_dicts = [d for d in dir_conf.values() if not isinstance(d, str)] |
90
|
|
|
|
91
|
|
|
# More dictionaries than roots |
92
|
|
|
if len(roots) < len(dir_dicts): |
93
|
|
|
raise TypeError("All subdirectory must have a _root key") |
94
|
|
|
|
95
|
|
|
for r in roots: |
96
|
|
|
key = r.split("_root")[0] |
97
|
|
|
subfolders = dir_conf[key] |
98
|
|
|
|
99
|
|
|
if not isinstance(subfolders, dict): |
100
|
|
|
raise TypeError("Expecting dict, got "+str(type(subfolders))) |
101
|
|
|
|
102
|
|
|
if prefix is not None: |
103
|
|
|
prefix = join(prefix, dir_conf[r]) |
104
|
|
|
else: |
105
|
|
|
prefix = dir_conf[r] |
106
|
|
|
|
107
|
|
|
create_directories(subfolders, prefix) |
108
|
|
|
|
109
|
|
|
|
110
|
|
|
def file_checksum(filename): |
111
|
|
|
"""Return the sha256 digest of a file. |
112
|
|
|
|
113
|
|
|
Parameters: |
114
|
|
|
filename (str): The file to hash. |
115
|
|
|
|
116
|
|
|
Returns: |
117
|
|
|
str: Hash of the file. |
118
|
|
|
""" |
119
|
|
|
return sha256(open(filename).read()).hexdigest() |
120
|
|
|
|
121
|
|
|
|