from django.shortcuts import render from .forms import LoginForm, BranchCreationForm, BranchChangeForm, UploadForm, ResetPasswordForm from django.shortcuts import render, redirect, reverse from django.http import HttpRequest, JsonResponse from django.contrib.auth.decorators import login_required, user_passes_test from django.contrib import messages from django.contrib.auth import get_user_model, login, logout from .models import Accounts, Media, Campaign import os from django.core.mail import send_mail, BadHeaderError from django.db.models.query_utils import Q from django.utils.http import urlsafe_base64_encode from django.utils.encoding import force_bytes from django.template.loader import render_to_string from django.http import HttpResponse from django.contrib.auth.tokens import default_token_generator from django.contrib.auth.forms import PasswordResetForm from datetime import datetime import logging import requests from requests.auth import HTTPBasicAuth from decouple import config from .tasks import upload_task from celery.result import AsyncResult logger = logging.getLogger(__name__) # Create your views here. def login_view(request): '''View for the login page''' msg = None data = {} first_login = request.GET.get('new_next') form = LoginForm(request.POST or None) data['form'] = form if request.method == 'POST': if form.is_valid(): user = form.login(request) if user is not None: if first_login: login(request, user) return redirect('first-login') elif user.is_admin: login(request, user) return redirect('main-dashboard') else: login(request, user) return redirect('branch-dashboard') else: messages.add_message(request, messages.WARNING, "Invalid credentials") logger.info('Invalid credentials') return render(request, 'register/login.html', context=data) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def main_dashboard(request): return render(request, 'main-dashboard.html') @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def media_upload(request): if request.method == 'POST': form = UploadForm(request.POST, request.FILES) files = request.FILES.getlist('file') file_ids = [] media_name = request.POST.get('media_name') + '_' + datetime.utcnow().strftime('%Y%m%d%H%M%S') if form.is_valid(): for f in files: media_type = f.content_type[:f.content_type.find('/')] file_name, extension = os.path.splitext(f.name) file_name = file_name.replace(' ', '_') file_instance = Media.objects.create(file=f, media_name=media_name, media_type=media_type, file_name=file_name, extension=extension) file_ids.append(file_instance.id) task = upload_task.delay(file_ids) task = upload_task.apply_async(queue='content_share_queue', kwargs={'file_ids':file_ids}) return JsonResponse({'task_id': task.id}) else: logger.info(f'Form invalid: {form.errors}') else: form = UploadForm() return render(request, 'media-upload.html') @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def additional_upload(request): if request.method == 'POST': form = UploadForm(request.POST, request.FILES) files = request.FILES.getlist('file') file_ids = [] media_name = request.POST.get('media_name') if form.is_valid(): for f in files: media_type = f.content_type[:f.content_type.find('/')] file_name, extension = os.path.splitext(f.name) file_name = file_name.replace(' ', '_') file_instance = Media.objects.create(file=f, media_name=media_name, media_type=media_type, file_name=file_name, extension=extension) file_ids.append(file_instance.id) task = upload_task.delay(file_ids) task = upload_task.apply_async(queue='content_share_queue', kwargs={'file_ids':file_ids}) return JsonResponse({'task_id': task.id}) else: logger.info(f'Form invalid: {form.errors}') else: context = {} form = UploadForm() context['campaign_name'] = request.GET.get('campaign_name') return render(request, 'add-media.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def delete_media(request): if request.method == 'POST': context = {} campaign_name = request.POST.get('campaign_name') file_name = request.POST.get('file_name') req = requests.post(url=config('API_DELETE_MEDIA'), json={'campaign_name':campaign_name, 'file_name':file_name}, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) try: req = requests.post(url=config('API_CAMPAIGN_CONTENT'), json={'campaign_name':campaign_name}, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: context['campaign'] = response['results'] else: logger.info(f'An error occured while trying to retrieve campaign content ') except Exception as e: logger.info(f'failed with error{e}') context['campaign'] = [] return render(request, 'delete-campaign-content.html', context) else: context = {} campaign_name = request.GET.get('campaign_name') context['campaign_name'] = campaign_name try: req = requests.post(url=config('API_CAMPAIGN_CONTENT'), json={'campaign_name':campaign_name}, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: messages.add_message(request, messages.SUCCESS, "File deleted!") context['campaign'] = response['results'] else: logger.info(f'An error occured while trying to retrieve campaign content ') except Exception as e: logger.info(f'failed with error{e}') context['campaign'] = [] return render(request, 'delete-campaign-content.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def delete_campaign(request): if request.method == 'POST': msg = None campaign_name = request.POST.get('campaign_name') try: req = requests.post(url=config('API_DELETE_CAMPAIGN'), json={'campaign_name':campaign_name}, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: messages.add_message(request, messages.SUCCESS, "Campaign deleted!") redirect('campaign-gallery') else: logger.info(f'An error occured while trying to retrieve campaign content ') except Exception as e: logger.info(f'failed with error{e}') messages.add_message(request, messages.WARNING, "Failed to delete campaign") return redirect(reverse('campaign-content') + f'?campaign_name={campaign_name}') def media_upload_status(request, task_id): if request.method == 'GET': task = AsyncResult(task_id) if task.state == 'PENDING': # job did not start yet response = { 'state': task.state, 'progress': 0, 'status': 'Pending...' } elif task.state != 'FAILURE': response = { 'state': task.state, 'progress': task.info.get('progress', 0), 'status': task.info.get('status', '') } if 'result' in task.info: response['result'] = task.info['result'] else: # something went wrong in the background job response = { 'state': task.state, 'progress': 1, 'status': str(task.info), # this is the exception raised } return JsonResponse(response) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def delete_upload(request, pk): if request.method == 'POST': media = Media.objects.get(pk=pk) media.delete() return redirect('display-media') @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def password_reset_request(request): '''Password reset route ''' if request.method == "POST": password_reset_form = PasswordResetForm(request.POST) if password_reset_form.is_valid(): data = password_reset_form.cleaned_data['email'] User = get_user_model() associated_users = User.objects.filter(Q(email=data)) if associated_users.exists(): for user in associated_users: subject = "Password Reset Requested" email_template_name = "password/password-email.html" c = { "email": user.email, 'domain': 'contentshare.npontu.com', 'site_name': 'Website', "uid": urlsafe_base64_encode(force_bytes(user.pk)), "user": user, 'token': default_token_generator.make_token(user), 'protocol': 'https' } email = render_to_string(email_template_name, c) try: send_mail(subject, email, "", [ user.email], fail_silently=False) return redirect('login') except BadHeaderError: return HttpResponse('Invalid header found.') password_reset_form = PasswordResetForm() return render(request=request, template_name="password/forgot-password.html", context={"password_reset_form": password_reset_form}) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def branch_register(request): msg = None form = BranchCreationForm() context = {} context['title'] = 'Users Add' context['settings'] = 'active' context['settings_trigger'] = 'pcoded-trigger' context['add'] = 'active' context['form'] = form if request.method == 'POST': form = BranchCreationForm(request.POST) if form.is_valid(): user = form.save(commit=False) raw_password = Accounts.objects.make_random_password() user.set_password(raw_password) #user email email = user.email username = user.username subject = "Account Created" try: send_mail(subject, f'''Account has been created for you on the internal tool username: {username} password: {raw_password} click on this link to login and change password, {config('APP_URL')}login?new_next={True}''', "", [email], fail_silently=False) except Exception as e: logger.info(f'failed with error {e}') logger.error('Could not send email with password') # logger.error("Could not send email with password ") user.save() messages.add_message(request, messages.SUCCESS, f"{user.username} successfully created") return redirect('branches-view') else: messages.add_message(request, messages.WARNING, form.errors) logger.info(f'Form invalid: {form.errors}') return render(request, 'branch-register.html', context) @login_required def branch_first_time_login(request): form = ResetPasswordForm() if request.method == "POST": form = ResetPasswordForm(request.POST) if form.is_valid(): try: new_branch = get_user_model().objects.get(pk = request.user.pk) new_branch.set_password(form.cleaned_data.get("password1")) new_branch.save() return redirect('login') except: logger.error("branch does not exist to change new password") return render(request, 'register/first-login.html', context={'form':form}) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def branches_view(request): '''View for the admin to manage users''' branches = get_user_model().objects.all() context = {} context['branches'] = branches if request.method == "POST": search_query = request.POST.get('search_query', '') branches = get_user_model().objects.filter(username__icontains=search_query) context['branches'] = branches return render(request, 'branches-view.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def branch_profile_view(request): '''View for branch profile''' context = {} branches = get_user_model().objects.filter(pk=request.user.branch_id) context['accounts'] = branches return render(request, 'branch-profile.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def branch_edit(request, pk): msg = None account_user_queried = get_user_model().objects.get(pk=pk) if request.method == 'POST': form = BranchChangeForm(request.POST, instance=account_user_queried) if form.is_valid(): user = form.save(commit=False) messages.add_message(request, messages.SUCCESS, "branch edited successfully") user.save() return redirect('branches-view') else: messages.add_message(request, messages.ERROR, "Form is not valid") logger.info(f'Form invalid: {form.errors}') return redirect('branch-edit', account_user_queried.branch_id) else: current_user = account_user_queried initial_dict = { "username": current_user.username, "branch_name": current_user.branch_name, "email": current_user.email, "is_admin": current_user.is_admin, "location" : current_user.location, "branch_id" : current_user.branch_id } form = BranchChangeForm(initial=initial_dict) context = {} context['account_user'] = account_user_queried return render(request, 'branch-edit.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def branch_delete(request, pk): if request.method == 'POST': msg = None try: u = Accounts.objects.get(pk=pk) u.delete() messages.success(request, "branch deleted") except u.DoesNotExist: messages.error(request, "branch does not exist") logger.info(f'branch does not exist') return redirect('branch-edit', pk) except Exception as e: logger.info(f'failed with error {e}') return redirect('branches-view') @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def campaign_gallery(request): if request.method == 'GET': context = {} try: req = requests.get(url=config('API_CAMPAIGN_LIST'), auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: context['campaigns'] = response['results'] else: logger.info(f'An error occured while trying to retrieve campaign list') except Exception as e: logger.info(f'failed with error{e}') context['campaigns'] = [] return render(request, 'campaign-gallery.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def campaign_content(request): if request.method == 'GET': context = {} campaign_name = request.GET.get('campaign_name') context['campaign_name'] = campaign_name try: req = requests.post(url=config('API_CAMPAIGN_CONTENT'), json={'campaign_name':campaign_name}, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: context['campaign'] = response['results'] else: logger.info(f'An error occured while trying to retrieve campaign content ') except Exception as e: logger.info(f'failed with error{e}') context['campaign'] = [] return render(request, 'campaign-content.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def schedule_campaign(request): if request.method == 'GET': context = {} campaign_name = request.GET.get('campaign_name') context['campaign_name'] = campaign_name branches = get_user_model().objects.all() context['branches'] = branches return render(request, 'schedule-campaign.html', context) else: context = {} campaign_name = request.POST.get('campaign_name') context['campaign_name'] = campaign_name if request.POST.get('fixed_rotation_type') == 'on': start_date = request.POST.get('start_date') end_date = request.POST.get('end_date') payload = { "pay_load" : { "campaign_name" : request.POST.get('campaign_name'), "branch" : request.POST.getlist('branch'), "rotation_type" : "fixed", "start_date" : start_date, "end_date" : end_date } } elif request.POST.get('daily_rotation_type') == 'on': start_time = request.POST.get('day_start') end_time = request.POST.get('day_end') payload = { "pay_load" : { "campaign_name" : request.POST.get('campaign_name'), "branch" : request.POST.getlist('branch'), "rotation_type" : "daily", "start_time" : start_time, "end_time" : end_time} } elif request.POST.get('daily_rotation_type') == 'on': start_time = request.POST.get('week_start') end_time = request.POST.get('week_end') days_week = request.POST.getlist('week_days') payload = { "pay_load" : { "campaign_name" : request.POST.get('campaign_name'), "branch" : request.POST.getlist('branch'), "rotation_type" : "week", "start_time" : start_time, "end_time" : end_time, "days_week" : days_week} } else: payload = {} try: req = requests.post(url=config('API_SCHEDULE_CAMPAIGN'), json=payload, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: messages.success(request, "campaign scheduled") return redirect('campaign-schedule') except Exception as e: logger.info(f'failed with error {e} while scheduling campaign') messages.error(request, "Error scheduling campaign") return render(request, 'schedule-campaign.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def campaign_schedule(request): if request.method == 'GET': context = {} context['campaigns'] = Campaign.objects.all() return render(request, 'campaign-schedule.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def campaign_deactivate(request): if request.method == 'POST': kill_switch = request.POST.get('kill_switch') campaign_name = request.POST.get('campaign_name') payload = { 'kill_switch' : kill_switch, 'campaign_name' : campaign_name } try: req = requests.post(url=config('API_CAMPAIGN_DEACTIVATE'), json=payload, auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: messages.success(request, "success") return redirect('campaign-schedule') except Exception as e: logger.error(f'failed with error {e} while activating campaign') messages.error(request, "failed") return render(request, 'campaign-schedule.html') @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def unassigned_branches(request): if request.method == 'GET': context = {} try: req = requests.get(url=config('API_UNASSIGNED_BRANCHES'), auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: context['branches'] = response['results'] else: logger.info(f'An error occured while trying to retrieve unassigned branches') except Exception as e: logger.info(f'failed with error{e}') context['branches'] = [] return render(request, 'unassigned-branches.html', context) @login_required @user_passes_test(lambda user: user.is_admin, login_url='forbidden') def unassigned_campaigns(request): if request.method == 'GET': context = {} try: req = requests.get(url=config('API_UNASSIGNED_CAMPAIGNS'), auth=HTTPBasicAuth(username=config('API_USERNAME'), password=config('API_PASSWORD'))) response = req.json() if response['code'] == 200: context['campaigns'] = response['results'] else: logger.info(f'An error occured while trying to retrieve unassigned campaigns') except Exception as e: logger.info(f'failed with error{e}') context['campaigns'] = [] return render(request, 'unassigned-campaigns.html', context) @login_required @user_passes_test(lambda user: not user.is_admin) def branch_dashboard(request): context = {} context['branch'] = request.user.branch_name return render(request, 'branch-dashboard.html', context) def forbidden_view(request): if request.method == 'GET': return render(request, 'register/forbidden.html')