Refactor custom commands with modern syntax

This commit is contained in:
Benbb96 2024-10-19 00:31:00 +02:00
parent 2686c46dbf
commit 0a1e73918d
No known key found for this signature in database
5 changed files with 152 additions and 296 deletions

View File

@ -38,7 +38,7 @@ Before django-helpdesk will be much use, you need to do some basic configuration
5. If you wish to exclude some days (eg, weekends) from escalation calculations, enter the dates manually via the Admin, or setup a cronjob to run a management command on a regular basis:: 5. If you wish to exclude some days (eg, weekends) from escalation calculations, enter the dates manually via the Admin, or setup a cronjob to run a management command on a regular basis::
0 0 * * 0 /path/to/helpdesksite/manage.py create_escalation_exclusions --days saturday,sunday --escalate-verbosely 0 0 * * 0 /path/to/helpdesksite/manage.py create_escalation_exclusions --days saturday sunday
This will, on a weekly basis, create exclusions for the coming weekend. This will, on a weekly basis, create exclusions for the coming weekend.

View File

@ -11,63 +11,7 @@ scripts/create_escalation_exclusion.py - Easy way to routinely add particular
from datetime import date, timedelta from datetime import date, timedelta
from django.core.management.base import BaseCommand, CommandError from django.core.management.base import BaseCommand, CommandError
import getopt
from helpdesk.models import EscalationExclusion, Queue from helpdesk.models import EscalationExclusion, Queue
from optparse import make_option
import sys
class Command(BaseCommand):
def __init__(self):
BaseCommand.__init__(self)
self.option_list += (
make_option(
'--days', '-d',
help='Days of week (monday, tuesday, etc)'),
make_option(
'--occurrences', '-o',
type='int',
default=1,
help='Occurrences: How many weeks ahead to exclude this day'),
make_option(
'--queues', '-q',
help='Queues to include (default: all). Use queue slugs'),
make_option(
'--escalate-verbosely', '-x',
action='store_true',
default=False,
dest='escalate-verbosely',
help='Display a list of dates excluded'),
)
def handle(self, *args, **options):
days = options['days']
# optparse should already handle the `or 1`
occurrences = options['occurrences'] or 1
verbose = False
queue_slugs = options['queues']
queues = []
if options['escalate-verbosely']:
verbose = True
if not (days and occurrences):
raise CommandError('One or more occurrences must be specified.')
if queue_slugs is not None:
queue_set = queue_slugs.split(',')
for queue in queue_set:
try:
q = Queue.objects.get(slug__exact=queue)
except Queue.DoesNotExist:
raise CommandError("Queue %s does not exist." % queue)
queues.append(q)
create_exclusions(days=days, occurrences=occurrences,
verbose=verbose, queues=queues)
day_names = { day_names = {
'monday': 0, 'monday': 0,
@ -80,79 +24,70 @@ day_names = {
} }
def create_exclusions(days, occurrences, verbose, queues): class Command(BaseCommand):
days = days.split(',') def add_arguments(self, parser):
for day in days: parser.add_argument(
day_name = day '-d',
day = day_names[day] '--days',
workdate = date.today() nargs='*',
i = 0 choices=list(day_names.keys()),
while i < occurrences: required=True,
if day == workdate.weekday(): help='Days of week (monday, tuesday, etc)'
if EscalationExclusion.objects.filter(date=workdate).count() == 0: )
esc = EscalationExclusion( parser.add_argument(
name='Auto Exclusion for %s' % day_name, date=workdate) '-o',
esc.save() '--occurrences',
default=1,
type=int,
help='Occurrences: How many weeks ahead to exclude this day'
)
parser.add_argument(
'-q',
'--queues',
nargs='*',
choices=list(Queue.objects.values_list('slug', flat=True)),
help='Queues to include (default: all). Use queue slugs'
)
parser.add_argument(
'-x',
'--exclude-verbosely',
action='store_true',
default=False,
help='Display a list of dates excluded'
)
if verbose: def handle(self, *args, **options):
print("Created exclusion for %s %s" % days = options['days']
(day_name, workdate)) occurrences = options['occurrences']
verbose = options['exclude_verbosely']
queue_slugs = options['queues']
if not (days and occurrences):
raise CommandError('One or more occurrences must be specified.')
queues = []
if queue_slugs is not None:
queues = Queue.objects.filter(slug__in=queue_slugs)
for day_name in days:
day = day_names[day_name]
workdate = date.today()
i = 0
while i < occurrences:
if day == workdate.weekday():
if EscalationExclusion.objects.filter(date=workdate).count() == 0:
esc = EscalationExclusion.objects.create(
name=f'Auto Exclusion for {day_name}',
date=workdate
)
for q in queues:
esc.queues.add(q)
if verbose: if verbose:
print(" - for queue %s" % q) self.stdout.write(f"Created exclusion for {day_name} {workdate}")
i += 1 for q in queues:
workdate += timedelta(days=1) esc.queues.add(q)
if verbose:
self.stdout.write(f" - for queue {q}")
i += 1
def usage(): workdate += timedelta(days=1)
print("Options:")
print(" --days, -d: Days of week (monday, tuesday, etc)")
print(" --occurrences, -o: Occurrences: How many weeks ahead to exclude this day")
print(" --queues, -q: Queues to include (default: all). Use queue slugs")
print(" --verbose, -v: Display a list of dates excluded")
if __name__ == '__main__':
# This script can be run from the command-line or via Django's manage.py.
try:
opts, args = getopt.getopt(sys.argv[1:], 'd:o:q:v', [
'days=', 'occurrences=', 'verbose', 'queues='])
except getopt.GetoptError:
usage()
sys.exit(2)
days = None
occurrences = 1
verbose = False
queue_slugs = None
queues = []
for o, a in opts:
if o in ('-x', '--escalate-verbosely'):
verbose = True
if o in ('-d', '--days'):
days = a
if o in ('-q', '--queues'):
queue_slugs = a
if o in ('-o', '--occurrences'):
occurrences = int(a) or 1
if not (days and occurrences):
usage()
sys.exit(2)
if queue_slugs is not None:
queue_set = queue_slugs.split(',')
for queue in queue_set:
try:
q = Queue.objects.get(slug__exact=queue)
except Queue.DoesNotExist:
print("Queue %s does not exist." % queue)
sys.exit(2)
queues.append(q)
create_exclusions(days=days, occurrences=occurrences,
verbose=verbose, queues=queues)

View File

@ -15,55 +15,54 @@ scripts/create_queue_permissions.py -
from django.contrib.auth.models import Permission from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django.core.management.base import BaseCommand, CommandError from django.core.management.base import BaseCommand
from django.db.utils import IntegrityError from django.db.utils import IntegrityError
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from helpdesk.models import Queue from helpdesk.models import Queue
from optparse import make_option
class Command(BaseCommand): class Command(BaseCommand):
def __init__(self): def add_arguments(self, parser):
BaseCommand.__init__(self) parser.add_argument(
'-q',
self.option_list += ( '--queues',
make_option( nargs='*',
'--queues', '-q', choices=list(Queue.objects.values_list('slug', flat=True)),
help='Queues to include (default: all). Use queue slugs'), help='Queues to include (default: all). Use queue slugs'
)
parser.add_argument(
'-x',
'--escalate-verbosely',
action='store_true',
default=False,
help='Display a list of dates excluded'
) )
def handle(self, *args, **options): def handle(self, *args, **options):
queue_slugs = options['queues'] queue_slugs = options['queues']
queues = []
if queue_slugs is not None: if queue_slugs is not None:
queue_set = queue_slugs.split(',') queues = Queue.objects.filter(slug__in=queue_slugs)
for queue in queue_set:
try:
q = Queue.objects.get(slug__exact=queue)
except Queue.DoesNotExist:
raise CommandError("Queue %s does not exist." % queue)
queues.append(q)
else: else:
queues = list(Queue.objects.all()) queues = Queue.objects.all()
# Create permissions for the queues, which may be all or not # Create permissions for the queues, which may be all or not
for q in queues: for q in queues:
self.stdout.write("Preparing Queue %s [%s]" % (q.title, q.slug)) self.stdout.write(f"Preparing Queue {q} [{q.slug}]")
if q.permission_name: if q.permission_name:
self.stdout.write( self.stdout.write(
" .. already has `permission_name=%s`" % q.permission_name) f" .. already has `permission_name={q.permission_name}`")
basename = q.permission_name[9:] basename = q.permission_name[9:]
else: else:
basename = q.generate_permission_name() basename = q.generate_permission_name()
self.stdout.write( self.stdout.write(
" .. generated `permission_name=%s`" % q.permission_name) f" .. generated `permission_name={q.permission_name}`")
q.save() q.save()
self.stdout.write( self.stdout.write(
" .. checking permission codename `%s`" % basename) f" .. checking permission codename `{basename}`")
try: try:
Permission.objects.create( Permission.objects.create(

View File

@ -8,168 +8,93 @@ scripts/escalate_tickets.py - Easy way to escalate tickets based on their age,
designed to be run from Cron or similar. designed to be run from Cron or similar.
""" """
from datetime import date, timedelta from datetime import date, timedelta
from django.core.management.base import BaseCommand, CommandError from django.core.management.base import BaseCommand
from django.db.models import Q from django.db.models import Q
from django.utils import timezone from django.utils import timezone
from django.utils.translation import gettext as _ from django.utils.translation import gettext as _
import getopt
from helpdesk.lib import safe_template_context from helpdesk.lib import safe_template_context
from helpdesk.models import EscalationExclusion, FollowUp, Queue, Ticket, TicketChange from helpdesk.models import EscalationExclusion, Queue, Ticket
from optparse import make_option
import sys
class Command(BaseCommand): class Command(BaseCommand):
def add_arguments(self, parser):
def __init__(self): parser.add_argument(
BaseCommand.__init__(self) '-q',
'--queues',
self.option_list = ( nargs='*',
make_option( choices=list(Queue.objects.values_list('slug', flat=True)),
'--queues', help='Queues to include (default: all). Use queue slugs'
help='Queues to include (default: all). Use queue slugs'), )
make_option( parser.add_argument(
'--verboseescalation', '-x',
action='store_true', '--escalate-verbosely',
default=False, action='store_true',
help='Display a list of dates excluded'), default=False,
help='Display escalated tickets'
) )
def handle(self, *args, **options): def handle(self, *args, **options):
verbose = False verbose = options['escalate_verbosely']
queue_slugs = None
queues = []
if 'verboseescalation' in options:
verbose = True
if 'queues' in options:
queue_slugs = options['queues']
queue_slugs = options['queues']
# Only include queues with escalation configured
queues = Queue.objects.filter(escalate_days__isnull=False).exclude(escalate_days=0)
if queue_slugs is not None: if queue_slugs is not None:
queue_set = queue_slugs.split(',') queues = queues.filter(slug__in=queue_slugs)
for queue in queue_set:
try:
Queue.objects.get(slug__exact=queue)
except Queue.DoesNotExist:
raise CommandError("Queue %s does not exist." % queue)
queues.append(queue)
escalate_tickets(queues=queues, verbose=verbose)
def escalate_tickets(queues, verbose):
""" Only include queues with escalation configured """
queryset = Queue.objects.filter(
escalate_days__isnull=False).exclude(escalate_days=0)
if queues:
queryset = queryset.filter(slug__in=queues)
for q in queryset:
last = date.today() - timedelta(days=q.escalate_days)
today = date.today()
workdate = last
days = 0
while workdate < today:
if EscalationExclusion.objects.filter(date=workdate).count() == 0:
days += 1
workdate = workdate + timedelta(days=1)
req_last_escl_date = date.today() - timedelta(days=days)
if verbose: if verbose:
print("Processing: %s" % q) self.stdout.write(f"Processing: {queues}")
Q_OPEN_STATUSES = Q() for queue in queues:
for open_status in Ticket.OPEN_STATUSES: last = date.today() - timedelta(days=queue.escalate_days)
Q_OPEN_STATUSES |= Q(status=open_status) today = date.today()
workdate = last
for t in q.ticket_set.filter(
Q_OPEN_STATUSES days = 0
).exclude(
priority=1 while workdate < today:
).filter( if not EscalationExclusion.objects.filter(date=workdate).exists():
Q(on_hold__isnull=True) | days += 1
Q(on_hold=False) workdate = workdate + timedelta(days=1)
).filter(
Q(last_escalation__lte=req_last_escl_date) | req_last_escl_date = timezone.now() - timedelta(days=days)
for ticket in queue.ticket_set.filter(
status__in=Ticket.OPEN_STATUSES
).exclude(
priority=1
).filter(
Q(on_hold__isnull=True) | Q(on_hold=False)
).filter(
Q(last_escalation__lte=req_last_escl_date) |
Q(last_escalation__isnull=True, created__lte=req_last_escl_date) Q(last_escalation__isnull=True, created__lte=req_last_escl_date)
): ):
t.last_escalation = timezone.now() ticket.last_escalation = timezone.now()
t.priority -= 1 ticket.priority -= 1
t.save() ticket.save()
context = safe_template_context(t) context = safe_template_context(ticket)
t.send( ticket.send(
{'submitter': ('escalated_submitter', context), {'submitter': ('escalated_submitter', context),
'ticket_cc': ('escalated_cc', context), 'ticket_cc': ('escalated_cc', context),
'assigned_to': ('escalated_owner', context)}, 'assigned_to': ('escalated_owner', context)},
fail_silently=True, fail_silently=True,
)
if verbose:
print(" - Esclating %s from %s>%s" % (
t.ticket,
t.priority + 1,
t.priority
)
) )
f = FollowUp( if verbose:
ticket=t, self.stdout.write(f" - Esclating {ticket.ticket} from {ticket.priority + 1}>{ticket.priority}")
title='Ticket Escalated',
date=timezone.now(),
public=True,
comment=_('Ticket escalated after %s days' % q.escalate_days),
)
f.save()
tc = TicketChange( followup = ticket.followup_set.create(
followup=f, title=_('Ticket Escalated'),
field=_('Priority'), public=True,
old_value=t.priority + 1, comment=_('Ticket escalated after %(nb)s days') % {'nb': queue.escalate_days},
new_value=t.priority, )
)
tc.save()
followup.ticketchange_set.create(
def usage(): field=_('Priority'),
print("Options:") old_value=ticket.priority + 1,
print(" --queues: Queues to include (default: all). Use queue slugs") new_value=ticket.priority,
print(" --verboseescalation: Display a list of dates excluded") )
if __name__ == '__main__':
try:
opts, args = getopt.getopt(
sys.argv[1:], ['queues=', 'verboseescalation'])
except getopt.GetoptError:
usage()
sys.exit(2)
verbose = False
queue_slugs = None
queues = []
for o, a in opts:
if o == '--verboseescalation':
verbose = True
if o == '--queues':
queue_slugs = a
if queue_slugs is not None:
queue_set = queue_slugs.split(',')
for queue in queue_set:
try:
q = Queue.objects.get(slug__exact=queue)
except Queue.DoesNotExist:
print("Queue %s does not exist." % queue)
sys.exit(2)
queues.append(queue)
escalate_tickets(queues=queues, verbose=verbose)

View File

@ -16,9 +16,6 @@ from helpdesk.email import process_email
class Command(BaseCommand): class Command(BaseCommand):
def __init__(self):
BaseCommand.__init__(self)
help = 'Process django-helpdesk queues and process e-mails via POP3/IMAP or ' \ help = 'Process django-helpdesk queues and process e-mails via POP3/IMAP or ' \
'from a local mailbox directory as required, feeding them into the helpdesk.' 'from a local mailbox directory as required, feeding them into the helpdesk.'
@ -39,8 +36,8 @@ class Command(BaseCommand):
) )
def handle(self, *args, **options): def handle(self, *args, **options):
quiet = options.get('quiet', False) quiet = options.get('quiet')
debug_to_stdout = options.get('debug_to_stdout', False) debug_to_stdout = options.get('debug_to_stdout')
process_email(quiet=quiet, debug_to_stdout=debug_to_stdout) process_email(quiet=quiet, debug_to_stdout=debug_to_stdout)