aboutsummaryrefslogtreecommitdiff
path: root/cbc/cli/cbc_remote_purge.py
blob: 397a161f9aefc53ebb63d8d664868a7cec26ba06 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
#!/usr/bin/env python
'''
DANGER. THIS WILL WIPE AN ENTIRE ANACONDA.ORG REPOSITORY CHANNEL.

YOU HAVE BEEN WARNED.
'''
import argparse
from subprocess import check_output, STDOUT


def choose(answer):
    answer = answer.upper()
    if answer == 'Y' or answer == 'YES':
        return True

    return False


def prompt_user(channel):
    message = 'You about to REMOVE every package inside of: {0}'.format(channel)
    message_length = len(message)
    print('!' * message_length)
    print(message)
    print('!' * message_length)
    print('')
    print('Continue? (y/N) ', end='')
    answer = input()
    print('')

    return choose(answer)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('channel', help='Name of channel to be purged of its contents')
    args = parser.parse_args()

    channel = args.channel

    show_command = 'conda-server channel --show {0}'.format(channel).split()
    show_output = check_output(show_command, stderr=STDOUT)
    show_output = show_output.decode()

    found = []
    for line in show_output.splitlines():
        line = line.lstrip()
        if not line:
            continue
        if not line.startswith('+'):
            continue
        line = line.replace('+', '').lstrip()
        package = '/'.join(line.split('/')[:2])
        found.append(package)

    if found:
        print("Packages to remove:")
        for pkg in found:
            print(pkg)

        if not prompt_user(channel):
            print('Operation aborted by user...')
            exit(0)

        print('')
        for pkg in found:
            purge_command = 'conda-server remove -f {0}'.format(pkg).split()
            print("Removing [{0:>10s}] :: {1:>10s}".format(channel, pkg))
            purge_output = check_output(purge_command, stderr=STDOUT)
    else:
        print("No packages in channel: {0}".format(channel))

if __name__ == '__main__':
    main()