pyceo/ceo/cli/members.py

226 lines
8.2 KiB
Python

import sys
from typing import Dict
import click
from zope import component
from ..term_utils import get_terms_for_renewal_for_user
from ..utils import http_post, http_get, http_patch, http_delete, \
get_failed_operations, user_dict_lines, get_adduser_operations
from .utils import handle_stream_response, handle_sync_response, print_lines, \
check_if_in_development
from ceo_common.interfaces import IConfig
from ceo_common.model.Term import get_terms_for_new_user
from ceod.transactions.members import DeleteMemberTransaction
@click.group(short_help='Perform operations on CSC members and club reps')
def members():
pass
@members.command(short_help='Add a new member or club rep')
@click.argument('username')
@click.option('--cn', help='Full name', required=False)
@click.option('--given-name', help='First name', required=False)
@click.option('--sn', help='Last name', required=False)
@click.option('--program', required=False, help='Academic program')
@click.option('--terms', 'num_terms', type=click.IntRange(1, 100),
help='Number of terms to add', default=1)
@click.option('--clubrep', is_flag=True, default=False,
help='Add non-member terms instead of member terms')
@click.option('--forwarding-address', required=False,
help=('Forwarding address to set in ~/.forward. '
'Default is UW address. '
'Set to the empty string to disable forwarding.'))
def add(username, cn, given_name, sn, program, num_terms, clubrep, forwarding_address):
cfg = component.getUtility(IConfig)
uw_domain = cfg.get('uw_domain')
# Try to get info from UWLDAP
resp = http_get('/api/uwldap/' + username)
if resp.ok:
result = handle_sync_response(resp)
if cn is None and result.get('cn'):
cn = result['cn']
if given_name is None and result.get('given_name'):
given_name = result['given_name']
if sn is None and result.get('sn'):
sn = result['sn']
if program is None and result.get('program'):
program = result['program']
if forwarding_address is None and result.get('mail_local_addresses'):
forwarding_address = result['mail_local_addresses'][0]
if cn is None:
cn = click.prompt('Full name')
if given_name is None:
given_name = click.prompt('First name')
if sn is None:
sn = click.prompt('Last name')
if forwarding_address is None:
forwarding_address = username + '@' + uw_domain
terms = get_terms_for_new_user(num_terms)
body = {
'uid': username,
'cn': cn,
'given_name': given_name,
'sn': sn,
}
if program is not None:
body['program'] = program
if clubrep:
body['non_member_terms'] = terms
else:
body['terms'] = terms
if forwarding_address != '':
body['forwarding_addresses'] = [forwarding_address]
else:
body['forwarding_addresses'] = []
click.echo("The following user will be created:")
print_user_lines(body)
click.confirm('Do you want to continue?', abort=True)
operations = get_adduser_operations(body)
resp = http_post('/api/members', json=body)
data = handle_stream_response(resp, operations)
result = data[-1]['result']
print_user_lines(result)
failed_operations = get_failed_operations(data)
if 'send_welcome_message' in failed_operations:
click.echo(click.style(
'Warning: welcome message was not sent. You now need to manually '
'send the user their password.', fg='yellow'))
def print_user_lines(d: Dict):
"""Pretty-print a serialized User."""
print_lines(user_dict_lines(d))
@members.command(short_help='Get info about a user')
@click.argument('username')
def get(username):
resp = http_get('/api/members/' + username)
result = handle_sync_response(resp)
print_user_lines(result)
@members.command(short_help="Replace a user's login shell or forwarding addresses")
@click.argument('username')
@click.option('--login-shell', required=False, help='Login shell')
@click.option('--forwarding-addresses', required=False,
help=(
'Comma-separated list of forwarding addresses. '
'Set to the empty string to disable forwarding.'
))
def modify(username, login_shell, forwarding_addresses):
if login_shell is None and forwarding_addresses is None:
click.echo('Nothing to do.')
sys.exit()
operations = []
body = {}
if login_shell is not None:
body['login_shell'] = login_shell
operations.append('replace_login_shell')
click.echo('Login shell will be set to: ' + login_shell)
if forwarding_addresses is not None:
if forwarding_addresses == '':
forwarding_addresses = []
else:
forwarding_addresses = forwarding_addresses.split(',')
body['forwarding_addresses'] = forwarding_addresses
operations.append('replace_forwarding_addresses')
prefix = '~/.forward will be set to: '
if len(forwarding_addresses) > 0:
click.echo(prefix + forwarding_addresses[0])
for address in forwarding_addresses[1:]:
click.echo((' ' * len(prefix)) + address)
else:
click.echo(prefix)
click.confirm('Do you want to continue?', abort=True)
resp = http_patch('/api/members/' + username, json=body)
handle_stream_response(resp, operations)
@members.command(short_help="Renew a member or club rep's membership")
@click.argument('username')
@click.option('--terms', 'num_terms', type=click.IntRange(1, 100),
help='Number of terms to add', prompt='Number of terms')
@click.option('--clubrep', is_flag=True, default=False,
help='Add non-member terms instead of member terms')
def renew(username, num_terms, clubrep):
terms = get_terms_for_renewal_for_user(username, num_terms, clubrep)
if clubrep:
body = {'non_member_terms': terms}
click.echo('The following non-member terms will be added: ' + ','.join(terms))
else:
body = {'terms': terms}
click.echo('The following member terms will be added: ' + ','.join(terms))
click.confirm('Do you want to continue?', abort=True)
resp = http_post(f'/api/members/{username}/renew', json=body)
handle_sync_response(resp)
click.echo('Done.')
@members.command(short_help="Reset a user's password")
@click.argument('username')
def pwreset(username):
click.confirm(f"Are you sure you want to reset {username}'s password?", abort=True)
resp = http_post(f'/api/members/{username}/pwreset')
result = handle_sync_response(resp)
click.echo('New password: ' + result['password'])
@members.command(short_help="Delete a user")
@click.argument('username')
def delete(username):
check_if_in_development()
click.confirm(f"Are you sure you want to delete {username}?", abort=True)
resp = http_delete(f'/api/members/{username}')
handle_stream_response(resp, DeleteMemberTransaction.operations)
@members.command(short_help="Check for and mark expired members")
@click.option('--dry-run', is_flag=True, default=False)
def expire(dry_run):
resp = http_post(f'/api/members/expire?dry_run={dry_run and "yes" or "no"}')
result = handle_sync_response(resp)
if len(result) > 0:
if dry_run:
click.echo("The following members will be marked as expired:")
else:
click.echo("The following members has been marked as expired:")
for username in result:
click.echo(username)
@members.command(short_help="Send renewal reminder emails to expiring members")
@click.option('--dry-run', is_flag=True, default=False)
def remindexpire(dry_run):
url = '/api/members/remindexpire'
if dry_run:
url += '?dry_run=true'
resp = http_post(url)
result = handle_sync_response(resp)
if len(result) > 0:
if dry_run:
click.echo("The following members will be sent membership renewal reminders:")
else:
click.echo("The following members were sent membership renewal reminders:")
for username in result:
click.echo(username)
else:
click.echo("No members are pending expiration.")