486 lines
17 KiB
Python
486 lines
17 KiB
Python
import os
|
|
import io
|
|
import pandas as pd
|
|
from django.shortcuts import render, redirect, get_object_or_404
|
|
from django.contrib.auth import login, logout, authenticate
|
|
from django.contrib.auth.forms import UserCreationForm, AuthenticationForm
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.contrib import messages
|
|
from django.db import transaction
|
|
from django.http import HttpResponse
|
|
from .models import Company, Profile, JobStatus, RequiredFolder, Job, JobFolderCompletion, JobFile
|
|
from .forms import CompanyForm, JobStatusForm, RequiredFolderForm, JobForm, JobFileForm, ImportJobsForm
|
|
|
|
def home(request):
|
|
if request.user.is_authenticated:
|
|
return redirect('dashboard')
|
|
|
|
context = {
|
|
"project_name": "RepairsHub",
|
|
}
|
|
return render(request, "core/index.html", context)
|
|
|
|
def register_view(request):
|
|
if request.method == 'POST':
|
|
form = UserCreationForm(request.POST)
|
|
if form.is_valid():
|
|
user = form.save()
|
|
login(request, user)
|
|
return redirect('company_setup')
|
|
else:
|
|
form = UserCreationForm()
|
|
return render(request, 'core/register.html', {'form': form})
|
|
|
|
def login_view(request):
|
|
if request.method == 'POST':
|
|
form = AuthenticationForm(data=request.POST)
|
|
if form.is_valid():
|
|
user = form.get_user()
|
|
login(request, user)
|
|
return redirect('dashboard')
|
|
else:
|
|
form = AuthenticationForm()
|
|
return render(request, 'core/login.html', {'form': form})
|
|
|
|
def logout_view(request):
|
|
logout(request)
|
|
return redirect('home')
|
|
|
|
@login_required
|
|
def company_setup(request):
|
|
if request.user.profile.company:
|
|
return redirect('dashboard')
|
|
|
|
if request.method == 'POST':
|
|
company_form = CompanyForm(request.POST)
|
|
status_names = request.POST.getlist('statuses')
|
|
default_status_idx = request.POST.get('default_status')
|
|
folder_names = request.POST.getlist('folders')
|
|
|
|
if company_form.is_valid() and status_names and default_status_idx is not None:
|
|
with transaction.atomic():
|
|
company = company_form.save()
|
|
profile = request.user.profile
|
|
profile.company = company
|
|
profile.role = 'ADMIN'
|
|
profile.save()
|
|
|
|
for i, name in enumerate(status_names):
|
|
if name.strip():
|
|
JobStatus.objects.create(
|
|
company=company,
|
|
name=name.strip(),
|
|
is_starting_status=(str(i) == default_status_idx),
|
|
order=i
|
|
)
|
|
|
|
for name in folder_names:
|
|
if name.strip():
|
|
RequiredFolder.objects.create(
|
|
company=company,
|
|
name=name.strip()
|
|
)
|
|
|
|
messages.success(request, f"Company {company.name} created successfully!")
|
|
return redirect('dashboard')
|
|
else:
|
|
if not status_names:
|
|
messages.error(request, "At least one status is required.")
|
|
if default_status_idx is None:
|
|
messages.error(request, "You must select a starting status.")
|
|
else:
|
|
company_form = CompanyForm()
|
|
|
|
return render(request, 'core/company_setup.html', {
|
|
'company_form': company_form,
|
|
})
|
|
|
|
@login_required
|
|
def dashboard(request):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
jobs = Job.objects.filter(company=company)
|
|
|
|
context = {
|
|
'company': company,
|
|
'total_jobs': jobs.count(),
|
|
'jobs': jobs.order_by('-created_at')[:5],
|
|
}
|
|
return render(request, 'core/dashboard.html', context)
|
|
|
|
@login_required
|
|
def job_list(request):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
jobs = Job.objects.filter(company=company).order_by('-created_at')
|
|
|
|
return render(request, 'core/job_list.html', {
|
|
'jobs': jobs,
|
|
'company': company
|
|
})
|
|
|
|
@login_required
|
|
def job_create(request):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
|
|
if request.method == 'POST':
|
|
form = JobForm(request.POST, company=company)
|
|
if form.is_valid():
|
|
job = form.save(commit=False)
|
|
job.company = company
|
|
job.save()
|
|
|
|
for folder in company.required_folders.all():
|
|
JobFolderCompletion.objects.get_or_create(job=job, folder=folder)
|
|
|
|
messages.success(request, f"Job {job.job_ref} created successfully!")
|
|
return redirect('job_detail', pk=job.pk)
|
|
else:
|
|
form = JobForm(company=company)
|
|
|
|
return render(request, 'core/job_form.html', {
|
|
'form': form,
|
|
'title': 'Create New Job',
|
|
'company': company
|
|
})
|
|
|
|
@login_required
|
|
def job_detail(request, pk):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
|
|
for folder in company.required_folders.all():
|
|
JobFolderCompletion.objects.get_or_create(job=job, folder=folder)
|
|
|
|
folder_completions = job.folder_completions.all().select_related('folder')
|
|
|
|
# Get files for each folder
|
|
for completion in folder_completions:
|
|
completion.files_list = job.files.filter(folder=completion.folder)
|
|
|
|
file_form = JobFileForm()
|
|
|
|
return render(request, 'core/job_detail.html', {
|
|
'job': job,
|
|
'folder_completions': folder_completions,
|
|
'company': company,
|
|
'file_form': file_form
|
|
})
|
|
|
|
@login_required
|
|
def job_update(request, pk):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
|
|
if request.method == 'POST':
|
|
form = JobForm(request.POST, instance=job, company=company)
|
|
if form.is_valid():
|
|
form.save()
|
|
messages.success(request, f"Job {job.job_ref} updated successfully!")
|
|
return redirect('job_detail', pk=job.pk)
|
|
else:
|
|
form = JobForm(instance=job, company=company)
|
|
|
|
return render(request, 'core/job_form.html', {
|
|
'form': form,
|
|
'title': f'Edit Job: {job.job_ref}',
|
|
'company': company
|
|
})
|
|
|
|
@login_required
|
|
def job_delete(request, pk):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
|
|
if request.method == 'POST':
|
|
job_ref = job.job_ref
|
|
job.delete()
|
|
messages.success(request, f"Job {job_ref} deleted.")
|
|
return redirect('job_list')
|
|
|
|
return render(request, 'core/job_confirm_delete.html', {
|
|
'job': job,
|
|
'company': company
|
|
})
|
|
|
|
@login_required
|
|
def toggle_folder_completion(request, pk, folder_id):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
completion = get_object_or_404(JobFolderCompletion, job=job, folder_id=folder_id)
|
|
completion.is_completed = not completion.is_completed
|
|
completion.save()
|
|
|
|
messages.success(request, f"Folder '{completion.folder.name}' status updated.")
|
|
return redirect('job_detail', pk=job.pk)
|
|
|
|
@login_required
|
|
def job_upload_file(request, pk, folder_id):
|
|
profile = request.user.profile
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
folder = get_object_or_404(RequiredFolder, pk=folder_id, company=company)
|
|
|
|
if request.method == 'POST':
|
|
form = JobFileForm(request.POST, request.FILES)
|
|
if form.is_valid():
|
|
job_file = form.save(commit=False)
|
|
job_file.job = job
|
|
job_file.folder = folder
|
|
job_file.uploaded_by = request.user
|
|
job_file.save()
|
|
messages.success(request, f"File uploaded to {folder.name}.")
|
|
else:
|
|
messages.error(request, "Error uploading file.")
|
|
|
|
return redirect('job_detail', pk=job.pk)
|
|
|
|
@login_required
|
|
def job_delete_file(request, pk, file_id):
|
|
profile = request.user.profile
|
|
company = profile.company
|
|
job = get_object_or_404(Job, pk=pk, company=company)
|
|
job_file = get_object_or_404(JobFile, pk=file_id, job=job)
|
|
|
|
job_file.file.delete()
|
|
job_file.delete()
|
|
messages.success(request, "File deleted.")
|
|
return redirect('job_detail', pk=job.pk)
|
|
|
|
@login_required
|
|
def job_export(request):
|
|
profile = request.user.profile
|
|
company = profile.company
|
|
jobs = Job.objects.filter(company=company)
|
|
|
|
data = []
|
|
for job in jobs:
|
|
data.append({
|
|
'Job Ref': job.job_ref,
|
|
'UPRN': job.uprn,
|
|
'Address 1': job.address_line_1,
|
|
'Address 2': job.address_line_2,
|
|
'Address 3': job.address_line_3,
|
|
'Postcode': job.postcode,
|
|
'Status': job.status.name,
|
|
'Description': job.description,
|
|
'Created At': job.created_at.strftime('%Y-%m-%d %H:%M:%S'),
|
|
})
|
|
|
|
df = pd.DataFrame(data)
|
|
|
|
# Export as Excel
|
|
output = io.BytesIO()
|
|
with pd.ExcelWriter(output, engine='openpyxl') as writer:
|
|
df.to_excel(writer, index=False, sheet_name='Jobs')
|
|
|
|
output.seek(0)
|
|
response = HttpResponse(output, content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
|
response['Content-Disposition'] = f'attachment; filename="jobs_export_{company.name}.xlsx"'
|
|
return response
|
|
|
|
@login_required
|
|
def job_import(request):
|
|
profile = request.user.profile
|
|
company = profile.company
|
|
|
|
if request.method == 'POST':
|
|
form = ImportJobsForm(request.POST, request.FILES)
|
|
if form.is_valid():
|
|
file = request.FILES['file']
|
|
try:
|
|
if file.name.endswith('.csv'):
|
|
df = pd.read_csv(file)
|
|
else:
|
|
df = pd.read_excel(file)
|
|
|
|
# Basic mapping: Job Ref, UPRN, Address 1, Postcode, Status
|
|
# Expecting columns to match roughly or using a simple mapping
|
|
|
|
starting_status = company.statuses.filter(is_starting_status=True).first()
|
|
|
|
imported_count = 0
|
|
errors = []
|
|
|
|
with transaction.atomic():
|
|
for index, row in df.iterrows():
|
|
job_ref = str(row.get('Job Ref', '')).strip()
|
|
if not job_ref: continue
|
|
|
|
uprn = str(row.get('UPRN', '')).strip() if pd.notna(row.get('UPRN')) else None
|
|
addr1 = str(row.get('Address 1', '')).strip()
|
|
postcode = str(row.get('Postcode', '')).strip()
|
|
|
|
# Find status or use default
|
|
status_name = str(row.get('Status', '')).strip()
|
|
status = company.statuses.filter(name__iexact=status_name).first() or starting_status
|
|
|
|
try:
|
|
job, created = Job.objects.update_or_create(
|
|
company=company,
|
|
job_ref=job_ref,
|
|
defaults={
|
|
'uprn': uprn,
|
|
'address_line_1': addr1,
|
|
'postcode': postcode,
|
|
'status': status,
|
|
}
|
|
)
|
|
|
|
# Initialize folders
|
|
for folder in company.required_folders.all():
|
|
JobFolderCompletion.objects.get_or_create(job=job, folder=folder)
|
|
|
|
imported_count += 1
|
|
except Exception as e:
|
|
errors.append(f"Row {index+2}: {str(e)}")
|
|
|
|
if errors:
|
|
messages.warning(request, f"Imported {imported_count} jobs with some errors: {', '.join(errors[:5])}")
|
|
else:
|
|
messages.success(request, f"Successfully imported {imported_count} jobs.")
|
|
|
|
return redirect('job_list')
|
|
|
|
except Exception as e:
|
|
messages.error(request, f"Error processing file: {str(e)}")
|
|
else:
|
|
form = ImportJobsForm()
|
|
|
|
return render(request, 'core/job_import.html', {'form': form, 'company': company})
|
|
|
|
@login_required
|
|
def settings_view(request):
|
|
profile = request.user.profile
|
|
if not profile.company:
|
|
return redirect('company_setup')
|
|
|
|
if profile.role != 'ADMIN':
|
|
messages.error(request, "Only admins can access company settings.")
|
|
return redirect('dashboard')
|
|
|
|
company = profile.company
|
|
statuses = company.statuses.all()
|
|
folders = company.required_folders.all()
|
|
|
|
return render(request, 'core/settings.html', {
|
|
'company': company,
|
|
'statuses': statuses,
|
|
'folders': folders
|
|
})
|
|
|
|
@login_required
|
|
def status_create(request):
|
|
profile = request.user.profile
|
|
if profile.role != 'ADMIN': return redirect('dashboard')
|
|
|
|
if request.method == 'POST':
|
|
form = JobStatusForm(request.POST)
|
|
if form.is_valid():
|
|
status = form.save(commit=False)
|
|
status.company = profile.company
|
|
if status.is_starting_status:
|
|
JobStatus.objects.filter(company=profile.company).update(is_starting_status=False)
|
|
status.save()
|
|
messages.success(request, "New status created.")
|
|
return redirect('settings')
|
|
else:
|
|
form = JobStatusForm()
|
|
|
|
return render(request, 'core/status_form.html', {'form': form, 'title': 'Create Status'})
|
|
|
|
@login_required
|
|
def status_update(request, pk):
|
|
profile = request.user.profile
|
|
if profile.role != 'ADMIN': return redirect('dashboard')
|
|
|
|
status = get_object_or_404(JobStatus, pk=pk, company=profile.company)
|
|
if request.method == 'POST':
|
|
form = JobStatusForm(request.POST, instance=status)
|
|
if form.is_valid():
|
|
if form.cleaned_data['is_starting_status']:
|
|
JobStatus.objects.filter(company=profile.company).update(is_starting_status=False)
|
|
form.save()
|
|
messages.success(request, "Status updated.")
|
|
return redirect('settings')
|
|
else:
|
|
form = JobStatusForm(instance=status)
|
|
|
|
return render(request, 'core/status_form.html', {'form': form, 'title': 'Edit Status'})
|
|
|
|
@login_required
|
|
def status_delete(request, pk):
|
|
profile = request.user.profile
|
|
if profile.role != 'ADMIN': return redirect('dashboard')
|
|
|
|
status = get_object_or_404(JobStatus, pk=pk, company=profile.company)
|
|
|
|
if status.is_starting_status:
|
|
messages.error(request, "Cannot delete the starting status. Please set another status as starting first.")
|
|
return redirect('settings')
|
|
|
|
if request.method == 'POST':
|
|
starting_status = profile.company.statuses.filter(is_starting_status=True).first()
|
|
Job.objects.filter(status=status).update(status=starting_status)
|
|
status.delete()
|
|
messages.success(request, f"Status deleted. Jobs reassigned to {starting_status.name}.")
|
|
return redirect('settings')
|
|
|
|
return render(request, 'core/status_confirm_delete.html', {'status': status})
|
|
|
|
@login_required
|
|
def folder_create(request):
|
|
profile = request.user.profile
|
|
if profile.role != 'ADMIN': return redirect('dashboard')
|
|
|
|
if request.method == 'POST':
|
|
form = RequiredFolderForm(request.POST)
|
|
if form.is_valid():
|
|
folder = form.save(commit=False)
|
|
folder.company = profile.company
|
|
folder.save()
|
|
messages.success(request, f"Folder '{folder.name}' added company-wide.")
|
|
return redirect('settings')
|
|
else:
|
|
form = RequiredFolderForm()
|
|
|
|
return render(request, 'core/folder_form.html', {'form': form})
|
|
|
|
@login_required
|
|
def folder_delete(request, pk):
|
|
profile = request.user.profile
|
|
if profile.role != 'ADMIN': return redirect('dashboard')
|
|
|
|
folder = get_object_or_404(RequiredFolder, pk=pk, company=profile.company)
|
|
if request.method == 'POST':
|
|
folder.delete()
|
|
messages.success(request, "Folder removed from company settings.")
|
|
return redirect('settings')
|
|
return render(request, 'core/folder_confirm_delete.html', {'folder': folder})
|