• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

deadc0de6 / catcli / 4601292488

pending completion
4601292488

push

github

deadc0de6
do not depend on fusepy and pyfzf

21 of 21 new or added lines in 3 files covered. (100.0%)

1284 of 1704 relevant lines covered (75.35%)

3.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.12
/catcli/catcli.py
1
#!/usr/bin/env python3
2
# author: deadc0de6
3

4
"""
5✔
5
author: deadc0de6 (https://github.com/deadc0de6)
6
Copyright (c) 2017, deadc0de6
7

8
Catcli command line interface
9
"""
10

11
import sys
5✔
12
import os
5✔
13
import datetime
5✔
14
from typing import Dict, Any, List
5✔
15
from docopt import docopt
5✔
16

17
# local imports
18
from catcli.version import __version__ as VERSION
5✔
19
from catcli.nodes import NodeTop, NodeAny
5✔
20
from catcli.logger import Logger
5✔
21
from catcli.colors import Colors
5✔
22
from catcli.catalog import Catalog
5✔
23
from catcli.walker import Walker
5✔
24
from catcli.noder import Noder
5✔
25
from catcli.utils import ask, edit, path_to_search_all
5✔
26
from catcli.exceptions import BadFormatException, CatcliException
5✔
27

28
NAME = 'catcli'
5✔
29
CUR = os.path.dirname(os.path.abspath(__file__))
5✔
30
CATALOGPATH = f'{NAME}.catalog'
5✔
31
GRAPHPATH = f'/tmp/{NAME}.dot'
5✔
32
FORMATS = ['native', 'csv', 'csv-with-header', 'fzf-native', 'fzf-csv']
5✔
33

34
BANNER = f""" +-+-+-+-+-+-+
5✔
35
 |c|a|t|c|l|i|
36
 +-+-+-+-+-+-+ v{VERSION}"""
37

38
USAGE = f"""
5✔
39
{BANNER}
40

41
Usage:
42
    {NAME} ls     [--catalog=<path>] [--format=<fmt>] [-aBCrVSs] [<path>]
43
    {NAME} find   [--catalog=<path>] [--format=<fmt>]
44
                  [-aBCbdVsP] [--path=<path>] [<term>]
45
    {NAME} index  [--catalog=<path>] [--meta=<meta>...]
46
                  [-aBCcfnV] <name> <path>
47
    {NAME} update [--catalog=<path>] [-aBCcfnV] [--lpath=<path>] <name> <path>
48
    {NAME} mount  [--catalog=<path>] [-V] <mountpoint>
49
    {NAME} rm     [--catalog=<path>] [-BCfV] <storage>
50
    {NAME} rename [--catalog=<path>] [-BCfV] <storage> <name>
51
    {NAME} edit   [--catalog=<path>] [-BCfV] <storage>
52
    {NAME} graph  [--catalog=<path>] [-BCV] [<path>]
53
    {NAME} print_supported_formats
54
    {NAME} help
55
    {NAME} --help
56
    {NAME} --version
57

58
Options:
59
    --catalog=<path>    Path to the catalog [default: {CATALOGPATH}].
60
    --meta=<meta>       Additional attribute to store [default: ].
61
    -a --archive        Handle archive file [default: False].
62
    -B --no-banner      Do not display the banner [default: False].
63
    -b --script         Output script to manage found file(s) [default: False].
64
    -C --no-color       Do not output colors [default: False].
65
    -c --hash           Calculate md5 hash [default: False].
66
    -d --directory      Only directory [default: False].
67
    -F --format=<fmt>   see \"print_supported_formats\" [default: native].
68
    -f --force          Do not ask when updating the catalog [default: False].
69
    -l --lpath=<path>   Path where changes are logged [default: ]
70
    -n --no-subsize     Do not store size of directories [default: False].
71
    -P --parent         Ignore stored relpath [default: True].
72
    -p --path=<path>    Start path.
73
    -r --recursive      Recursive [default: False].
74
    -s --raw-size       Print raw size [default: False].
75
    -S --sortsize       Sort by size, largest first [default: False].
76
    -V --verbose        Be verbose [default: False].
77
    -v --version        Show version.
78
    -h --help           Show this screen.
79
"""  # nopep8
80

81

82
def cmd_mount(args: Dict[str, Any],
5✔
83
              top: NodeTop,
84
              noder: Noder) -> bool:
85
    """mount action"""
86
    mountpoint = args['<mountpoint>']
×
87
    debug = args['--verbose']
×
88
    try:
×
89
        from catcli.fuser import Fuser  # pylint: disable=C0415
×
90
        Fuser(mountpoint, top, noder,
×
91
              debug=debug)
92
    except ModuleNotFoundError:
×
93
        Logger.err('install fusepy to use mount')
×
94
        return False
×
95
    return True
×
96

97

98
def cmd_index(args: Dict[str, Any],
5✔
99
              noder: Noder,
100
              catalog: Catalog,
101
              top: NodeTop) -> None:
102
    """index action"""
103
    path = args['<path>']
5✔
104
    name = args['<name>']
5✔
105
    usehash = args['--hash']
5✔
106
    debug = args['--verbose']
5✔
107
    subsize = not args['--no-subsize']
5✔
108
    if not os.path.exists(path):
5✔
109
        Logger.err(f'\"{path}\" does not exist')
×
110
        return
×
111
    if name in noder.get_storage_names(top):
5✔
112
        try:
×
113
            if not ask(f'Overwrite storage \"{name}\"'):
×
114
                Logger.err(f'storage named \"{name}\" already exist')
×
115
                return
×
116
        except KeyboardInterrupt:
×
117
            Logger.err('aborted')
×
118
            return
×
119
        node = noder.get_storage_node(top, name)
×
120
        node.parent = None
×
121

122
    start = datetime.datetime.now()
5✔
123
    walker = Walker(noder, usehash=usehash, debug=debug)
5✔
124
    attr = args['--meta']
5✔
125
    root = noder.new_storage_node(name, path, top, attr)
5✔
126
    _, cnt = walker.index(path, root, name)
5✔
127
    if subsize:
5✔
128
        noder.rec_size(root, store=True)
5✔
129
    stop = datetime.datetime.now()
5✔
130
    diff = stop - start
5✔
131
    Logger.info(f'Indexed {cnt} file(s) in {diff}')
5✔
132
    if cnt > 0:
5✔
133
        catalog.save(top)
5✔
134

135

136
def cmd_update(args: Dict[str, Any],
5✔
137
               noder: Noder,
138
               catalog: Catalog,
139
               top: NodeTop) -> None:
140
    """update action"""
141
    path = args['<path>']
5✔
142
    name = args['<name>']
5✔
143
    usehash = args['--hash']
5✔
144
    logpath = args['--lpath']
5✔
145
    debug = args['--verbose']
5✔
146
    subsize = not args['--no-subsize']
5✔
147
    if not os.path.exists(path):
5✔
148
        Logger.err(f'\"{path}\" does not exist')
×
149
        return
×
150
    root = noder.get_storage_node(top, name, newpath=path)
5✔
151
    if not root:
5✔
152
        Logger.err(f'storage named \"{name}\" does not exist')
×
153
        return
×
154
    start = datetime.datetime.now()
5✔
155
    walker = Walker(noder, usehash=usehash, debug=debug,
5✔
156
                    logpath=logpath)
157
    cnt = walker.reindex(path, root, top)
5✔
158
    if subsize:
5✔
159
        noder.rec_size(root, store=True)
5✔
160
    stop = datetime.datetime.now()
5✔
161
    diff = stop - start
5✔
162
    Logger.info(f'updated {cnt} file(s) in {diff}')
5✔
163
    if cnt > 0:
5✔
164
        catalog.save(top)
5✔
165

166

167
def cmd_ls(args: Dict[str, Any],
5✔
168
           noder: Noder,
169
           top: NodeTop) -> List[NodeAny]:
170
    """ls action"""
171
    path = path_to_search_all(args['<path>'])
5✔
172
    fmt = args['--format']
5✔
173
    if fmt.startswith('fzf'):
5✔
174
        raise BadFormatException('fzf is not supported in ls, use find')
×
175
    found = noder.list(top,
5✔
176
                       path,
177
                       rec=args['--recursive'],
178
                       fmt=fmt,
179
                       raw=args['--raw-size'])
180
    if not found:
5✔
181
        path = args['<path>']
5✔
182
        Logger.err(f'\"{path}\": nothing found')
5✔
183
    return found
5✔
184

185

186
def cmd_rm(args: Dict[str, Any],
5✔
187
           noder: Noder,
188
           catalog: Catalog,
189
           top: NodeTop) -> NodeTop:
190
    """rm action"""
191
    name = args['<storage>']
5✔
192
    node = noder.get_storage_node(top, name)
5✔
193
    if node:
5✔
194
        node.parent = None
5✔
195
        if catalog.save(top):
5✔
196
            Logger.info(f'Storage \"{name}\" removed')
5✔
197
    else:
198
        Logger.err(f'Storage named \"{name}\" does not exist')
5✔
199
    return top
5✔
200

201

202
def cmd_find(args: Dict[str, Any],
5✔
203
             noder: Noder,
204
             top: NodeTop) -> List[NodeAny]:
205
    """find action"""
206
    fromtree = args['--parent']
5✔
207
    directory = args['--directory']
5✔
208
    startpath = args['--path']
5✔
209
    fmt = args['--format']
5✔
210
    raw = args['--raw-size']
5✔
211
    script = args['--script']
5✔
212
    search_for = args['<term>']
5✔
213
    found = noder.find_name(top, search_for,
5✔
214
                            script=script,
215
                            startnode=startpath,
216
                            only_dir=directory,
217
                            parentfromtree=fromtree,
218
                            fmt=fmt, raw=raw)
219
    return found
5✔
220

221

222
def cmd_graph(args: Dict[str, Any],
5✔
223
              noder: Noder,
224
              top: NodeTop) -> None:
225
    """graph action"""
226
    path = args['<path>']
5✔
227
    if not path:
5✔
228
        path = GRAPHPATH
×
229
    cmd = noder.to_dot(top, path)
5✔
230
    Logger.info(f'create graph with \"{cmd}\" (you need graphviz)')
5✔
231

232

233
def cmd_rename(args: Dict[str, Any],
5✔
234
               catalog: Catalog,
235
               top: NodeTop) -> None:
236
    """rename action"""
237
    storage = args['<storage>']
×
238
    new = args['<name>']
×
239
    storages = list(x.name for x in top.children)
×
240
    if storage in storages:
×
241
        node = next(filter(lambda x: x.name == storage, top.children))
×
242
        node.name = new
×
243
        if catalog.save(top):
×
244
            msg = f'Storage \"{storage}\" renamed to \"{new}\"'
×
245
            Logger.info(msg)
×
246
    else:
247
        Logger.err(f'Storage named \"{storage}\" does not exist')
×
248

249

250
def cmd_edit(args: Dict[str, Any],
5✔
251
             noder: Noder,
252
             catalog: Catalog,
253
             top: NodeTop) -> None:
254
    """edit action"""
255
    storage = args['<storage>']
×
256
    storages = list(x.name for x in top.children)
×
257
    if storage in storages:
×
258
        node = next(filter(lambda x: x.name == storage, top.children))
×
259
        attr = node.attr
×
260
        if not attr:
×
261
            attr = ''
×
262
        new = edit(attr)
×
263
        node.attr = noder.attrs_to_string(new)
×
264
        if catalog.save(top):
×
265
            Logger.info(f'Storage \"{storage}\" edited')
×
266
    else:
267
        Logger.err(f'Storage named \"{storage}\" does not exist')
×
268

269

270
def banner() -> None:
5✔
271
    """print banner"""
272
    Logger.stderr_nocolor(BANNER)
×
273
    Logger.stderr_nocolor("")
×
274

275

276
def print_supported_formats() -> None:
5✔
277
    """print all supported formats to stdout"""
278
    print('"native"     : native format')
×
279
    print('"csv"        : CSV format')
×
280
    print(f'               {Noder.CSV_HEADER}')
×
281
    print('"fzf-native" : fzf to native (only valid for find)')
×
282
    print('"fzf-csv"    : fzf to csv (only valid for find)')
×
283

284

285
def main() -> bool:
5✔
286
    """entry point"""
287
    args = docopt(USAGE, version=VERSION)
5✔
288

289
    if args['help'] or args['--help']:
5✔
290
        print(USAGE)
×
291
        return True
×
292

293
    if args['print_supported_formats']:
5✔
294
        print_supported_formats()
×
295
        return True
×
296

297
    # check format
298
    fmt = args['--format']
5✔
299
    if fmt not in FORMATS:
5✔
300
        Logger.err(f'bad format: {fmt}')
×
301
        print_supported_formats()
×
302
        return False
×
303

304
    if args['--verbose']:
5✔
305
        print(args)
×
306

307
    # print banner
308
    if not args['--no-banner']:
5✔
309
        banner()
×
310

311
    # set colors
312
    if args['--no-color']:
5✔
313
        Colors.no_color()
×
314

315
    # init noder
316
    noder = Noder(debug=args['--verbose'], sortsize=args['--sortsize'],
5✔
317
                  arc=args['--archive'])
318
    # init catalog
319
    catalog_path = args['--catalog']
5✔
320
    catalog = Catalog(catalog_path, debug=args['--verbose'],
5✔
321
                      force=args['--force'])
322
    # init top node
323
    top = catalog.restore()
5✔
324
    if not top:
5✔
325
        top = noder.new_top_node()
5✔
326

327
    # handle the meta node
328
    meta = noder.update_metanode(top)
5✔
329
    catalog.set_metanode(meta)
5✔
330

331
    # parse command
332
    try:
5✔
333
        if args['index']:
5✔
334
            cmd_index(args, noder, catalog, top)
5✔
335
        if args['update']:
5✔
336
            if not catalog.exists():
5✔
337
                Logger.err(f'no such catalog: {catalog_path}')
×
338
                return False
×
339
            cmd_update(args, noder, catalog, top)
5✔
340
        elif args['find']:
5✔
341
            if not catalog.exists():
×
342
                Logger.err(f'no such catalog: {catalog_path}')
×
343
                return False
×
344
            cmd_find(args, noder, top)
×
345
        elif args['ls']:
5✔
346
            if not catalog.exists():
5✔
347
                Logger.err(f'no such catalog: {catalog_path}')
×
348
                return False
×
349
            cmd_ls(args, noder, top)
5✔
350
        elif args['mount']:
5✔
351
            if not catalog.exists():
×
352
                Logger.err(f'no such catalog: {catalog_path}')
×
353
                return False
×
354
            if not cmd_mount(args, top, noder):
×
355
                return False
×
356
        elif args['rm']:
5✔
357
            if not catalog.exists():
×
358
                Logger.err(f'no such catalog: {catalog_path}')
×
359
                return False
×
360
            cmd_rm(args, noder, catalog, top)
×
361
        elif args['graph']:
5✔
362
            if not catalog.exists():
×
363
                Logger.err(f'no such catalog: {catalog_path}')
×
364
                return False
×
365
            cmd_graph(args, noder, top)
×
366
        elif args['rename']:
5✔
367
            if not catalog.exists():
×
368
                Logger.err(f'no such catalog: {catalog_path}')
×
369
                return False
×
370
            cmd_rename(args, catalog, top)
×
371
        elif args['edit']:
5✔
372
            if not catalog.exists():
×
373
                Logger.err(f'no such catalog: {catalog_path}')
×
374
                return False
×
375
            cmd_edit(args, noder, catalog, top)
×
376
    except CatcliException as exc:
×
377
        Logger.stderr_nocolor('ERROR ' + str(exc))
×
378
        return False
×
379

380
    return True
5✔
381

382

383
if __name__ == '__main__':
5✔
384
    if main():
5✔
385
        sys.exit(0)
5✔
386
    sys.exit(1)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc