Completed
Push — master ( 2f4fc0...f5d9f6 )
by John
03:06
created

compress_config_loader()   A

Complexity

Conditions 3

Size

Total Lines 12

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 6
CRAP Score 3

Importance

Changes 2
Bugs 0 Features 0
Metric Value
cc 3
c 2
b 0
f 0
dl 0
loc 12
ccs 6
cts 6
cp 1
crap 3
rs 9.4285
1
#!/usr/bin/env python3
2 4
"""This module is used to operate with archives."""
3
4 4
import os  # filesystem read
5 4
import subprocess  # invocation of 7z, cap
6 4
import zipfile  # zip compresssion
7 4
import tarfile  # txz/tbz/tgz/tar compression
8 4
from bbarchivist import barutils  # zip tester
9 4
from bbarchivist import utilities  # platform determination
10 4
from bbarchivist import bbconstants  # premade stuff
11 4
from bbarchivist import decorators  # timer
12 4
from bbarchivist import iniconfig  # config parsing
13
14 4
__author__ = "Thurask"
15 4
__license__ = "WTFPL v2"
16 4
__copyright__ = "Copyright 2015-2016 Thurask"
17
18
19 4
def smart_is_tarfile(filepath):
20
    """
21
    :func:`tarfile.is_tarfile` plus error handling.
22
23
    :param filepath: Filename.
24
    :type filepath: str
25
    """
26 4
    try:
27 4
        istar = tarfile.is_tarfile(filepath)
28 4
    except (OSError, IOError):
29 4
        return False
30
    else:
31 4
        return istar
32
33
34 4
@decorators.timer
35 4
def sz_compress(filepath, filename, szexe=None, strength=5, errors=False):
36
    """
37
    Pack a file into a LZMA2 7z file.
38
39
    :param filepath: Basename of file, no extension.
40
    :type filepath: str
41
42
    :param filename: Name of file to pack.
43
    :type filename: str
44
45
    :param szexe: Path to 7z executable.
46
    :type szexe: str
47
48
    :param strength: Compression strength. 5 is normal, 9 is ultra.
49
    :type strength: int
50
51
    :param errors: Print completion status message. Default is false.
52
    :type errors: bool
53
    """
54 4
    szcodes = {
55
        0: "NO ERRORS",
56
        1: "COMPLETED WITH WARNINGS",
57
        2: "FATAL ERROR",
58
        7: "COMMAND LINE ERROR",
59
        8: "OUT OF MEMORY ERROR",
60
        255: "USER STOPPED PROCESS"
61
    }
62 4
    strength = str(strength)
63 4
    rawname = os.path.dirname(filepath)
64 4
    thr = str(utilities.get_core_count())
65 4
    fold = os.path.join(rawname, filename)
66 4
    cmd = '{0} a -mx{1} -m0=lzma2 -mmt{2} "{3}.7z" "{4}"'.format(szexe, strength,
67
                                                                 thr, filepath, fold)
68 4
    excode = sz_subprocess(cmd)
69 4
    if errors:
70 4
        print(szcodes[excode])
71
72
73 4
def sz_subprocess(cmd):
74
    """
75
    Subprocess wrapper for 7-Zip commands.
76
77
    :param cmd: Command to pass to subprocess.
78
    :type cmd: str
79
    """
80 4
    with open(os.devnull, 'wb') as dnull:
81 4
        output = subprocess.call(cmd, stdout=dnull, stderr=subprocess.STDOUT, shell=True)
82 4
    return output
83
84
85 4
def sz_verify(filepath, szexe=None):
86
    """
87
    Verify that a .7z file is valid and working.
88
89
    :param filepath: Filename.
90
    :type filepath: str
91
92
    :param szexe: Path to 7z executable.
93
    :type szexe: str
94
    """
95 4
    filepath = os.path.abspath(filepath)
96 4
    cmd = '{0} t "{1}"'.format(szexe, filepath)
97 4
    excode = sz_subprocess(cmd)
98 4
    return excode == 0
99
100
101 4
def generic_tarfile_verify(filepath, method):
102
    """
103
    Verify that a tar/tgz/tbz/txz file is valid and working.
104
105
    :param filepath: Filename.
106
    :type filepath: str
107
108
    :param method: Tarfile read method.
109
    :type method: str
110
    """
111 4
    if smart_is_tarfile(filepath):
112 4
        with tarfile.open(filepath, method) as thefile:
113 4
            mems = thefile.getmembers()
114 4
        return False if not mems else True
115
    else:
116 4
        return False
117
118
119 4
def generic_tarfile_compress(archivename, filename, method, strength=5):
120
    """
121
    Pack a file into an uncompressed/gzip/bzip2/LZMA tarfile.
122
123
    :param archivename: Archive name.
124
    :type archivename: str
125
126
    :param filename: Name of file to pack into archive.
127
    :type filename: str
128
129
    :param method: Tarfile compress method.
130
    :type method: str
131
132
    :param strength: Compression strength. 5 is normal, 9 is ultra.
133
    :type strength: int
134
    """
135 4
    nocomp = ["w:", "w:xz"]  # methods w/o compression: tar, tar.xz
136 4
    if method in nocomp:
137 4
        generic_nocompresslevel(archivename, filename, method)
138
    else:
139 4
        generic_compresslevel(archivename, filename, method, strength)
140
141
142 4
def generic_compresslevel(archivename, filename, method, strength=5):
143
    """
144
    Pack a file into a gzip/bzip2 tarfile.
145
146
    :param archivename: Archive name.
147
    :type archivename: str
148
149
    :param filename: Name of file to pack into archive.
150
    :type filename: str
151
152
    :param method: Tarfile compress method.
153
    :type method: str
154
155
    :param strength: Compression strength. 5 is normal, 9 is ultra.
156
    :type strength: int
157
    """
158 4
    with tarfile.open(archivename, method, compresslevel=strength) as afile:
159 4
        afile.add(filename, filter=None, arcname=os.path.basename(filename))
160
161
162 4
def generic_nocompresslevel(archivename, filename, method):
163
    """
164
    Pack a file into an uncompressed/LZMA tarfile.
165
166
    :param archivename: Archive name.
167
    :type archivename: str
168
169
    :param filename: Name of file to pack into archive.
170
    :type filename: str
171
172
    :param method: Tarfile compress method.
173
    :type method: str
174
    """
175 4
    with tarfile.open(archivename, method) as afile:
176 4
        afile.add(filename, filter=None, arcname=os.path.basename(filename))
177
178
179 4
@decorators.timer
180
def tar_compress(filepath, filename):
181
    """
182
    Pack a file into an uncompressed tarfile.
183
184
    :param filepath: Basename of file, no extension.
185
    :type filepath: str
186
187
    :param filename: Name of file to pack.
188
    :type filename: str
189
    """
190 4
    generic_tarfile_compress("{0}.tar".format(filepath), filename, "w:")
191
192
193 4
def tar_verify(filepath):
194
    """
195
    Verify that a tar file is valid and working.
196
197
    :param filepath: Filename.
198
    :type filepath: str
199
    """
200 4
    return generic_tarfile_verify(filepath, "r:")
201
202
203 4
@decorators.timer
204 4
def tgz_compress(filepath, filename, strength=5):
205
    """
206
    Pack a file into a gzip tarfile.
207
208
    :param filepath: Basename of file, no extension.
209
    :type filepath: str
210
211
    :param filename: Name of file to pack.
212
    :type filename: str
213
214
    :param strength: Compression strength. 5 is normal, 9 is ultra.
215
    :type strength: int
216
    """
217 4
    generic_tarfile_compress("{0}.tar.gz".format(filepath), filename, "w:gz", strength)
218
219
220 4
def tgz_verify(filepath):
221
    """
222
    Verify that a tar.gz file is valid and working.
223
224
    :param filepath: Filename.
225
    :type filepath: str
226
    """
227 4
    return generic_tarfile_verify(filepath, "r:gz")
228
229
230 4
@decorators.timer
231 4
def tbz_compress(filepath, filename, strength=5):
232
    """
233
    Pack a file into a bzip2 tarfile.
234
235
    :param filepath: Basename of file, no extension.
236
    :type filepath: str
237
238
    :param filename: Name of file to pack.
239
    :type filename: str
240
241
    :param strength: Compression strength. 5 is normal, 9 is ultra.
242
    :type strength: int
243
    """
244 4
    generic_tarfile_compress("{0}.tar.bz2".format(filepath), filename, "w:bz2", strength)
245
246
247 4
def tbz_verify(filepath):
248
    """
249
    Verify that a tar.bz2 file is valid and working.
250
251
    :param filepath: Filename.
252
    :type filepath: str
253
    """
254 4
    return generic_tarfile_verify(filepath, "r:bz2")
255
256
257 4
@decorators.timer
258
def txz_compress(filepath, filename):
259
    """
260
    Pack a file into a LZMA tarfile.
261
262
    :param filepath: Basename of file, no extension.
263
    :type filepath: str
264
265
    :param filename: Name of file to pack.
266
    :type filename: str
267
    """
268 3
    if not utilities.new_enough(3):
269
        pass
270
    else:
271 3
        generic_tarfile_compress("{0}.tar.xz".format(filepath), filename, "w:xz")
272
273
274 4
def txz_verify(filepath):
275
    """
276
    Verify that a tar.xz file is valid and working.
277
278
    :param filepath: Filename.
279
    :type filepath: str
280
    """
281 3
    if not utilities.new_enough(3):
282
        return None
283
    else:
284 3
        return generic_tarfile_verify(filepath, "r:xz")
285
286
287 4
@decorators.timer
288
def zip_compress(filepath, filename):
289
    """
290
    Pack a file into a DEFLATE zipfile.
291
292
    :param filepath: Basename of file, no extension.
293
    :type filepath: str
294
295
    :param filename: Name of file to pack.
296
    :type filename: str
297
    """
298 4
    with zipfile.ZipFile("{0}.zip".format(filepath), 'w', zipfile.ZIP_DEFLATED, allowZip64=True) as zfile:
299 4
        zfile.write(filename, arcname=os.path.basename(filename))
300
301
302 4
def zip_verify(filepath):
303
    """
304
    Verify that a .zip file is valid and working.
305
306
    :param filepath: Filename.
307
    :type filepath: str
308
    """
309 4
    if zipfile.is_zipfile(filepath):
310 4
        brokens = barutils.bar_tester(filepath)
311 4
        return brokens != filepath
312
    else:
313 4
        return False
314
315
316 4
def filter_method(method, szexe=None):
317
    """
318
    Make sure methods are OK.
319
320
    :param method: Compression method to use.
321
    :type method: str
322
323
    :param szexe: Path to 7z executable, if needed.
324
    :type szexe: str
325
    """
326 4
    if not utilities.new_enough(3) and method == "txz":
327 1
        method = "zip"  # fallback
328 4
    if method == "7z" and szexe is None:
329 4
        ifexists = utilities.prep_seven_zip()  # see if 7z exists
330 4
        if not ifexists:
331 4
            method = "zip"  # fallback
332
        else:
333 4
            szexe = utilities.get_seven_zip(False)
334 4
    return method
335
336
337 4
def calculate_strength():
338
    """
339
    Determine zip/gzip/bzip2 strength by OS bit setting.
340
    """
341 4
    strength = 9 if utilities.is_amd64() else 5
342 4
    return strength
343
344
345 4
def filtercomp(files, criterion, critargs, boolfilt=True):
346
    """
347
    :param files: Files to work on.
348
    :type files: list(str)
349
350
    :param criterion: Function to use for evaluation.
351
    :type criterion: func
352
353
    :param critargs: Arguments for function, other than file.
354
    :type critargs: list
355
356
    :param boolfilt: True if comparing criterion, False if comparing not criterion.
357
    :type boolfilt: bool
358
    """
359 4
    if boolfilt:
360 4
        fx2 = [file for file in files if criterion(file, *critargs)]
361
    else:
362 4
        fx2 = [file for file in files if not criterion(file, *critargs)]
363 4
    return fx2
364
365
366 4
def compressfilter(filepath, selective=False):
367
    """
368
    Filter directory listing of working directory.
369
370
    :param filepath: Working directory. Required.
371
    :type filepath: str
372
373
    :param selective: Only compress autoloaders. Default is false.
374
    :type selective: bool/str
375
    """
376 4
    arx = bbconstants.ARCS
377 4
    pfx = bbconstants.PREFIXES
378 4
    files = [file for file in os.listdir(filepath) if not os.path.isdir(file)]
379 4
    if selective is None:
380 4
        filt2 = os.listdir(filepath)
381 4
    elif selective == "arcsonly":
382 4
        filt2 = filtercomp(files, utilities.prepends, ("", arx))
383 4
    elif selective:
384 4
        filt0 = filtercomp(files, utilities.prepends, (pfx, ""))
385 4
        filt1 = filtercomp(filt0, utilities.prepends, ("", arx), False)  # pop archives
386 4
        filt2 = filtercomp(filt1, utilities.prepends, ("", ".exe"))  # include exes
387
    else:
388 4
        filt2 = filtercomp(files, utilities.prepends, ("", arx), False)  # pop archives
389 4
    filt3 = [os.path.join(filepath, file) for file in filt2]
390 4
    return filt3
391
392
393 4
def prep_compress_function(method="7z", szexe=None, errors=False):
394
    """
395
    Prepare compression function and partial arguments.
396
397
    :param method: Compression type. Default is "7z".
398
    :type method: str
399
400
    :param szexe: Path to 7z executable, if needed.
401
    :type szexe: str
402
403
    :param errors: Print completion status message. Default is false.
404
    :type errors: bool
405
    """
406 4
    methods = {"7z": sz_compress, "tgz": tgz_compress, "txz": txz_compress, "tbz": tbz_compress,
407
               "tar": tar_compress, "zip": zip_compress}
408 4
    args = [szexe] if method == "7z" else []
409 4
    if method in ("7z", "tbz", "tgz"):
410 4
        args.append(calculate_strength())
411 4
    if method == "7z":
412 4
        args.append(errors)
413 4
    return methods[method], args
414
415
416 4
def compress(filepath, method="7z", szexe=None, selective=False, errors=False):
417
    """
418
    Compress all autoloader files in a given folder, with a given method.
419
420
    :param filepath: Working directory. Required.
421
    :type filepath: str
422
423
    :param method: Compression type. Default is "7z".
424
    :type method: str
425
426
    :param szexe: Path to 7z executable, if needed.
427
    :type szexe: str
428
429
    :param selective: Only compress autoloaders. Default is false.
430
    :type selective: bool
431
432
    :param errors: Print completion status message. Default is false.
433
    :type errors: bool
434
    """
435 4
    method = filter_method(method, szexe)
436 4
    files = compressfilter(filepath, selective)
437 4
    for file in files:
438 4
        fname = os.path.basename(file)
439 4
        filename = os.path.splitext(fname)[0]
440 4
        fileloc = os.path.join(filepath, filename)
441 4
        print("COMPRESSING: {0}".format(fname))
442 4
        compfunc, extargs = prep_compress_function(method, szexe, errors)
443 4
        compfunc(fileloc, file, *extargs)
444 4
    return True
445
446
447 4
def tarzip_verifier(file):
448
    """
449
    Assign .tar.xxx, .tar and .zip verifiers.
450
451
    :param file: Filename.
452
    :type file: str
453
    """
454 4
    maps = {".tar.gz": tgz_verify, ".tar.xz": txz_verify,
455
            ".tar.bz2": tbz_verify, ".tar": tar_verify,
456
            ".zip": zip_verify, ".bar": zip_verify}
457 4
    for key, value in maps.items():
458 4
        if file.endswith(key):
459
            return value(file)
460
461
462 4
def decide_verifier(file, szexe=None):
463
    """
464
    Decide which verifier function to use.
465
466
    :param file: Filename.
467
    :type file: str
468
469
    :param szexe: Path to 7z executable, if needed.
470
    :type szexe: str
471
    """
472 4
    print("VERIFYING: {0}".format(file))
473 4
    if file.endswith(".7z") and szexe is not None:
474
        verif = sz_verify(os.path.abspath(file), szexe)
475
    else:
476 4
        verif = tarzip_verifier(file)
477 4
    decide_verifier_printer(file, verif)
478 4
    return verif
479
480
481 4
def decide_verifier_printer(file, verif):
482
    """
483
    Print status of verifier function.
484
485
    :param file: Filename.
486
    :type file: str
487
488
    :param verif: If the file is OK or not.
489
    :type verif: bool
490
    """
491 4
    if not verif:
492 4
        print("{0} IS BROKEN!".format(os.path.basename(file)))
493
    else:
494
        print("{0} OK".format(os.path.basename(file)))
495
496
497 4
def verify(filepath, method="7z", szexe=None, selective=False):
498
    """
499
    Verify specific archive files in a given folder.
500
501
    :param filepath: Working directory. Required.
502
    :type filepath: str
503
504
    :param method: Compression type. Default is "7z". Defined in source.
505
    :type method: str
506
507
    :param szexe: Path to 7z executable, if needed.
508
    :type szexe: str
509
510
    :param selective: Only verify autoloaders. Default is false.
511
    :type selective: bool
512
    """
513 4
    method = filter_method(method, szexe)
514 4
    files = compressfilter(filepath, selective)
515 4
    for file in files:
516 4
        decide_verifier(file, szexe)
517
518
519 4
def compress_config_loader(homepath=None):
520
    """
521
    Read a ConfigParser file to get compression preferences.
522
523
    :param homepath: Folder containing ini file. Default is user directory.
524
    :type homepath: str
525
    """
526 4
    compini = iniconfig.generic_loader('compression', homepath)
527 4
    method = compini.get('method', fallback="7z")
528 4
    if not utilities.new_enough(3) and method == "txz":
529 1
        method = "zip"  # for 3.2 compatibility
530 4
    return method
531
532
533 4
def compress_config_writer(method=None, homepath=None):
534
    """
535
    Write a ConfigParser file to store compression preferences.
536
537
    :param method: Method to use.
538
    :type method: str
539
540
    :param homepath: Folder containing ini file. Default is user directory.
541
    :type homepath: str
542
    """
543 4
    method = compress_config_loader() if method is None else method
544 4
    results = {"method": method}
545
    iniconfig.generic_writer("compression", results, homepath)
546