Statistiques
| Branche: | Tag: | Révision :

dockonsurf / modules / utilities.py @ 082685ad

Historique | Voir | Annoter | Télécharger (3 ko)

1
import logging
2

    
3
logger = logging.getLogger('DockOnSurf')
4

    
5

    
6
def tail(f, lines=20):
7
    total_lines_wanted = lines
8

    
9
    block_size = 1024
10
    f.seek(0, 2)
11
    block_end_byte = f.tell()
12
    lines_to_go = total_lines_wanted
13
    block_number = -1
14
    blocks = []
15
    while lines_to_go > 0 and block_end_byte > 0:
16
        if block_end_byte - block_size > 0:
17
            f.seek(block_number * block_size, 2)
18
            blocks.append(f.read(block_size))
19
        else:
20
            f.seek(0, 0)
21
            blocks.append(f.read(block_end_byte))
22
        lines_found = blocks[-1].count(b'\n')
23
        lines_to_go -= lines_found
24
        block_end_byte -= block_size
25
        block_number -= 1
26
    all_read_text = b''.join(reversed(blocks))
27
    return b'\n'.join(all_read_text.splitlines()[-total_lines_wanted:]).decode()
28

    
29

    
30
def check_bak(file_name):
31
    """Checks if a file already exists and backs it up if so.
32
    @param file_name: file to be checked if exists
33
    """
34
    import os
35
    new_name = file_name
36
    bak_num = 0
37
    while os.path.isdir(new_name) or os.path.isfile(new_name):
38
        bak_num += 1
39
        new_name = new_name.split(".bak")[0] + f".bak{bak_num}"
40
    if bak_num > 0:
41
        os.rename(file_name, new_name)
42
        logger.warning(f"'{file_name}' already present. Backed it up to "
43
                       f"{new_name}.")
44

    
45

    
46
def try_command(command, expct_error_types: list, *args, **kwargs):
47
    """Try to run a command and record exceptions (expected and not) on a log.
48

49
    @param command: method or function, the command to be executed.
50
    @param expct_error_types: list of tuples, every inner tuple is supposed to
51
    contain an exception type (eg. ValueError, TypeError, etc.) to be caught and
52
    a message to print in the log and on the screen explaining the exception.
53
    Error types that are not allow to be called with a custom message as only
54
    error argument are not supported.
55
    The outer tuple encloses all couples of error types and their relative
56
    messages.
57
    *args and **kwargs: arguments and keyword-arguments of the command to be
58
    executed.
59
    When trying to run 'command' with its args and kwargs, if an exception
60
    present on the 'error_types' occurs, its relative error message is recorded
61
    on the log and a same type exception is raised with the custom message.
62
    """
63
    unexp_error = "An unexpected error occurred"
64

    
65
    err = False
66
    try:
67
        return_val = command(*args, **kwargs)
68
    except Exception as e:
69
        for expct_err in expct_error_types:
70
            if isinstance(e, expct_err[0]):
71
                logger.error(expct_err[1])
72
                err = expct_err[0](expct_err[1])
73
                break
74
        else:
75
            logger.exception(unexp_error)
76
            err = e
77
    else:
78
        err = False
79
        return return_val
80
    finally:
81
        if isinstance(err, BaseException):
82
            raise err
83

    
84

    
85
def _human_key(key):
86
    import re
87
    parts = re.split('(\d*\.\d+|\d+)', key)
88
    return tuple((e.swapcase() if i % 2 == 0 else float(e))
89
                 for i, e in enumerate(parts))