_loop()   C
last analyzed

Complexity

Conditions 7

Size

Total Lines 21

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 47.8546

Importance

Changes 0
Metric Value
cc 7
c 0
b 0
f 0
dl 0
loc 21
rs 6.4705
ccs 1
cts 17
cp 0.0588
crap 47.8546
1
#!/usr/bin/env python
2
3 1
"""Command-line interface for DropTheBeat."""
4
5 1
import os
6 1
import sys
7 1
import time
8 1
import argparse
9 1
import logging
10
11 1
from dtb import CLI
12 1
from dtb import share, user, gui
13 1
from dtb.common import SHARED, WarningFormatter
14 1
from dtb import settings
15
16
17 1
def main(args=None):
18
    """Process command-line arguments and run the program."""
19
20
    # Main parser
21 1
    parser = argparse.ArgumentParser(prog=CLI, description=__doc__, **SHARED)
0 ignored issues
show
Coding Style introduced by
Usage of * or ** arguments should usually be done with care.

Generally, there is nothing wrong with usage of * or ** arguments. For readability of the code base, we suggest to not over-use these language constructs though.

For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect.

Loading history...
22 1
    parser.add_argument('-g', '--gui', action='store_true',
23
                        help="launch the GUI")
24 1
    parser.add_argument('-d', '--daemon', action='store_true',
25
                        help="if terminal mode, run forever")
26 1
    parser.add_argument('-q', '--no-log', action='store_true',
27
                        help="do not create a log for downloads")
28
    # TODO: support sharing multiple songs
29 1
    parser.add_argument('-s', '--share', metavar='PATH',
30
                        help="recommend a song")
31 1
    parser.add_argument('-i', '--incoming', action='store_true',
32
                        help="display the incoming songs")
33 1
    parser.add_argument('-o', '--outgoing', action='store_true',
34
                        help="display the outgoing songs")
35 1
    parser.add_argument('-u', '--users', metavar='n', nargs='*',
36
                        help="filter to the specified usernames")
37 1
    parser.add_argument('-n', '--new', metavar='"First Last"',
38
                        help="create a new user")
39 1
    parser.add_argument('-x', '--delete', action='store_true',
40
                        help="delete the current user")
41
    # Hidden argument to override the root sharing directory path
42 1
    parser.add_argument('--root', metavar="PATH", help=argparse.SUPPRESS)
43
    # Hidden argument to override the home directory
44 1
    parser.add_argument('--home', metavar='PATH', help=argparse.SUPPRESS)
45
    # Hidden argument to run the program as a different user
46 1
    parser.add_argument('--test', metavar='"First Last"',
47
                        help=argparse.SUPPRESS)
48
49
    # Parse arguments
50 1
    args = parser.parse_args(args=args)
51
52
    # Configure logging
53 1
    _configure_logging(args.verbose)
54
55
    # Run the program
56 1
    try:
57 1
        success = _run(args, os.getcwd(), parser.error)
58
    except KeyboardInterrupt:
59
        logging.debug("command canceled")
60
    else:
61 1
        if success:
62 1
            logging.debug("command succeeded")
63
        else:
64
            logging.debug("command failed")
65
            sys.exit(1)
66
67
68 1
def _configure_logging(verbosity=0):
69
    """Configure logging using the provided verbosity level (0+)."""
70
71
    # Configure the logging level and format
72 1
    if verbosity == 0:
73 1
        level = settings.DEFAULT_LOGGING_LEVEL
74 1
        default_format = settings.DEFAULT_LOGGING_FORMAT
75 1
        verbose_format = settings.VERBOSE_LOGGING_FORMAT
76 1
    elif verbosity == 1:
77 1
        level = settings.VERBOSE_LOGGING_LEVEL
78 1
        default_format = verbose_format = settings.VERBOSE_LOGGING_FORMAT
79 1
    elif verbosity == 2:
80 1
        level = settings.VERBOSE2_LOGGING_LEVEL
81 1
        default_format = verbose_format = settings.VERBOSE_LOGGING_FORMAT
82
    else:
83 1
        level = settings.VERBOSE2_LOGGING_LEVEL
84 1
        default_format = verbose_format = settings.VERBOSE2_LOGGING_FORMAT
85
86
    # Set a custom formatter
87 1
    logging.basicConfig(level=level)
88 1
    formatter = WarningFormatter(default_format, verbose_format)
89 1
    logging.root.handlers[0].setFormatter(formatter)
90
91
92 1
def _run(args, cwd, err):  # pylint: disable=W0613
93
    """Process arguments and run the main program.
94
95
    @param args: Namespace of CLI arguments
96
    @param cwd: current working directory
97
    @param err: function to call for CLI errors
98
    """
99
    # Run the GUI
100
    if args.gui:
101
        logging.info("launching the GUI...")
102
        return gui.run(args)
103
104
    # Find the sharing directory
105
    root = args.root or share.find()
106
107
    # Create a new user and exit
108
    if args.new:
109
        return _new(args.new, root)
110
111
    # Get the current user
112
    if args.test:
113
        this = user.User(os.path.join(root, args.test))
114
    else:
115
        this = user.get_current(root)
116
    this.cleanup()
117
118
    # Delete user and exit
119
    if args.delete:
120
        this.delete()
121
        print("deleted: {}".format(this))
122
        return True
123
124
    # Display incoming, share a song, and/or display outgoing and exit
125
    if any((args.incoming, args.share, args.outgoing)):
126
127
        if args.incoming:
128
            logging.info("displaying incoming songs...")
129
            for song in this.incoming:
130
                print("incoming: {}".format(song))
131
132
        if args.share:
133
            path = os.path.abspath(args.share)
134
            song = this.recommend(path, args.users)
135
            print("shared: {}".format(path))
136
137
        if args.outgoing:
138
            logging.info("displaying outgoing songs...")
139
            for song in this.outgoing:
140
                print("outgoing: {}".format(song))
141
142
        return True
143
144
    # Run the command-line interface loop
145
    logging.info("starting the main loop...")
146
    return _loop(this, args.daemon, not args.no_log)
147
148
149 1
def _new(name, root):
150
    """Create a new user."""
151
    try:
152
        this = user.User.new(root, name)
153
        print("created: {}".format(this))
154
    except EnvironmentError as error:
155
        logging.error(error)
156
        return False
157
158
    return True
159
160
161 1
def _loop(this, daemon, log):
162
    """Run the main CLI loop."""
163
    while True:
164
        for song in this.incoming:
165
            path = song.download()
166
            if path:
167
                print("downloaded: {}".format(path))
168
                # Append download message to the log
169
                if log:
170
                    dirpath, filename = os.path.split(path)
171
                    logpath = os.path.join(dirpath, CLI + '.log')
172
                    with open(logpath, 'a') as log:
173
                        msg = "{} from {}".format(filename, song.friendname)
174
                        log.write(msg + '\n')
175
        if daemon:
176
            logging.debug("daemon sleeping for 5 seconds...")
177
            time.sleep(5)
178
        else:
179
            break
180
181
    return True
182
183
184
if __name__ == '__main__':  # pragma: no cover (manual test)
185
    main()
186