ixgadmin
cdefbd010c
Media was unreachable from either portal, which is why the certificate button on the buyer compliance page returned page-not-found and no document row had anything to click. The cause was structural: the portal and ops hosts proxy location / to the app with a path prefix, so a stored /media/... URL arrives as /portal/media/... and matches no route. Add a download view per portal behind the same role gate as the page that links to it — buyer-visible documents and published certificates for buyers, anything the ops list shows for operations — resolving the stored path safely under MEDIA_ROOT so a traversal or an absolute path is refused. Point the seven templates that list files at the new routes. Also add a portal-wide render sweep: nothing in the suite was fetching the ops or portal pages, so a view could lose its template_name and 500 in front of a user with every test still green.
410 lines
15 KiB
Python
410 lines
15 KiB
Python
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
|
|
from django.db.models import Count, Q
|
|
from django.shortcuts import redirect
|
|
from django.utils import timezone
|
|
from django.views.generic import TemplateView, ListView, DetailView
|
|
from django.views import View
|
|
from django.shortcuts import get_object_or_404
|
|
|
|
from apps.batches.models import ProcessingBatch
|
|
from apps.qc.models import QCResult
|
|
from apps.shipments.models import Shipment
|
|
from apps.capa.models import CAPARecord
|
|
from apps.documents.downloads import file_response
|
|
from apps.documents.models import Document, CertificateOfAnalysis
|
|
from apps.esg.models import ESGFieldReport
|
|
from apps.notifications.models import Notification
|
|
from apps.compliance.models import DueDiligenceCase
|
|
from apps.guides.views import GuideDetailView as BaseGuideDetailView
|
|
from apps.guides.views import GuideIndexView as BaseGuideIndexView
|
|
|
|
|
|
class OpsRequiredMixin(UserPassesTestMixin):
|
|
def test_func(self):
|
|
user = self.request.user
|
|
return user.is_authenticated and (user.role in ('ops', 'admin') or user.is_superuser)
|
|
|
|
|
|
class DashboardView(OpsRequiredMixin, TemplateView):
|
|
template_name = 'ops/dashboard.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['total_batches'] = ProcessingBatch.objects.count()
|
|
ctx['batches_by_status'] = {
|
|
s: ProcessingBatch.objects.filter(status=s).count()
|
|
for s, _ in ProcessingBatch.Status.choices
|
|
}
|
|
ctx['recent_qc'] = QCResult.objects.select_related('batch').order_by('-tested_at')[:10]
|
|
ctx['qc_summary'] = QCResult.objects.aggregate(
|
|
passed=Count('id', filter=Q(overall_pass=True)),
|
|
failed=Count('id', filter=Q(overall_pass=False)),
|
|
)
|
|
ctx['active_shipments'] = Shipment.objects.exclude(status='delivered').count()
|
|
ctx['open_capas'] = CAPARecord.objects.filter(status__in=('open', 'in_progress')).count()
|
|
ctx['recent_notifications'] = Notification.objects.filter(
|
|
user=self.request.user, is_read=False
|
|
)[:5]
|
|
ctx['recent_batches'] = ProcessingBatch.objects.order_by('-created_at')[:5]
|
|
return ctx
|
|
|
|
|
|
class BatchListView(OpsRequiredMixin, ListView):
|
|
model = ProcessingBatch
|
|
template_name = 'ops/batch_list.html'
|
|
context_object_name = 'batches'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = ProcessingBatch.objects.all()
|
|
status = self.request.GET.get('status')
|
|
grade = self.request.GET.get('grade')
|
|
search = self.request.GET.get('q')
|
|
if status:
|
|
qs = qs.filter(status=status)
|
|
if grade:
|
|
qs = qs.filter(grade=grade)
|
|
if search:
|
|
qs = qs.filter(batch_reference__icontains=search)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['status_choices'] = ProcessingBatch.Status.choices
|
|
ctx['grade_choices'] = ProcessingBatch.Grade.choices
|
|
ctx['current_status'] = self.request.GET.get('status', '')
|
|
ctx['current_grade'] = self.request.GET.get('grade', '')
|
|
ctx['current_q'] = self.request.GET.get('q', '')
|
|
return ctx
|
|
|
|
|
|
class BatchDetailView(OpsRequiredMixin, DetailView):
|
|
model = ProcessingBatch
|
|
template_name = 'ops/batch_detail.html'
|
|
context_object_name = 'batch'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
batch = self.object
|
|
ctx['qc_results'] = QCResult.objects.filter(batch=batch)
|
|
ctx['shipments'] = Shipment.objects.filter(batch=batch)
|
|
ctx['capa_records'] = CAPARecord.objects.filter(batch=batch)
|
|
ctx['documents'] = Document.objects.filter(batch=batch)
|
|
ctx['esg_reports'] = ESGFieldReport.objects.filter(batch=batch)
|
|
return ctx
|
|
|
|
|
|
class QCListView(OpsRequiredMixin, ListView):
|
|
model = QCResult
|
|
template_name = 'ops/qc_list.html'
|
|
context_object_name = 'results'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = QCResult.objects.select_related('batch').all()
|
|
passed = self.request.GET.get('passed')
|
|
if passed == '1':
|
|
qs = qs.filter(overall_pass=True)
|
|
elif passed == '0':
|
|
qs = qs.filter(overall_pass=False)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['current_passed'] = self.request.GET.get('passed', '')
|
|
return ctx
|
|
|
|
|
|
class QCDetailView(OpsRequiredMixin, DetailView):
|
|
model = QCResult
|
|
template_name = 'ops/qc_detail.html'
|
|
context_object_name = 'result'
|
|
|
|
|
|
class ShipmentListView(OpsRequiredMixin, ListView):
|
|
model = Shipment
|
|
template_name = 'ops/shipment_list.html'
|
|
context_object_name = 'shipments'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = Shipment.objects.select_related('batch').all()
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
qs = qs.filter(status=status)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['status_choices'] = Shipment.Status.choices
|
|
ctx['current_status'] = self.request.GET.get('status', '')
|
|
return ctx
|
|
|
|
|
|
class ShipmentDetailView(OpsRequiredMixin, DetailView):
|
|
model = Shipment
|
|
template_name = 'ops/shipment_detail.html'
|
|
context_object_name = 'shipment'
|
|
|
|
|
|
class CAPAListView(OpsRequiredMixin, ListView):
|
|
model = CAPARecord
|
|
template_name = 'ops/capa_list.html'
|
|
context_object_name = 'records'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = CAPARecord.objects.select_related('batch').all()
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
qs = qs.filter(status=status)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['status_choices'] = CAPARecord.Status.choices
|
|
ctx['current_status'] = self.request.GET.get('status', '')
|
|
return ctx
|
|
|
|
|
|
class CAPADetailView(OpsRequiredMixin, DetailView):
|
|
model = CAPARecord
|
|
template_name = 'ops/capa_detail.html'
|
|
context_object_name = 'record'
|
|
|
|
|
|
class DocumentListView(OpsRequiredMixin, ListView):
|
|
model = Document
|
|
template_name = 'ops/document_list.html'
|
|
context_object_name = 'documents'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = Document.objects.select_related('batch', 'uploaded_by').all()
|
|
doc_type = self.request.GET.get('doc_type')
|
|
if doc_type:
|
|
qs = qs.filter(doc_type=doc_type)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['doc_type_choices'] = Document.DocType.choices
|
|
ctx['current_doc_type'] = self.request.GET.get('doc_type', '')
|
|
return ctx
|
|
|
|
|
|
class DocumentDownloadView(OpsRequiredMixin, View):
|
|
"""Serve the stored file behind a document row to operations staff.
|
|
|
|
Operations may fetch any document the ops document list shows it. Note the
|
|
list itself has always shown every row regardless of visibility, so this
|
|
matches the list rather than being stricter than it; the buyer-side view is
|
|
where visibility does the gatekeeping.
|
|
"""
|
|
|
|
def get(self, request, pk):
|
|
doc = get_object_or_404(Document, pk=pk)
|
|
return file_response(doc.file_url, doc.file_name)
|
|
|
|
|
|
class CoADownloadView(OpsRequiredMixin, View):
|
|
"""Serve a certificate of analysis to operations staff, at any status.
|
|
|
|
Unlike the buyer's view, a certificate still awaiting release is exactly what
|
|
a reviewer needs to open, so the status is not part of the gate here.
|
|
"""
|
|
|
|
def get(self, request, pk):
|
|
coa = get_object_or_404(CertificateOfAnalysis, pk=pk)
|
|
return file_response(coa.file_url, coa.file_name)
|
|
|
|
|
|
class ESGListView(OpsRequiredMixin, ListView):
|
|
model = ESGFieldReport
|
|
template_name = 'ops/esg_list.html'
|
|
context_object_name = 'reports'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
return ESGFieldReport.objects.select_related('batch').all()
|
|
|
|
|
|
class QCVerifyListView(OpsRequiredMixin, ListView):
|
|
model = QCResult
|
|
template_name = 'ops/qc_verify_list.html'
|
|
context_object_name = 'results'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = QCResult.objects.select_related('batch', 'entered_by').filter(
|
|
verification_status=QCResult.VerificationStatus.DRAFT
|
|
)
|
|
passed = self.request.GET.get('passed')
|
|
if passed == '1':
|
|
qs = qs.filter(overall_pass=True)
|
|
elif passed == '0':
|
|
qs = qs.filter(overall_pass=False)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['current_passed'] = self.request.GET.get('passed', '')
|
|
return ctx
|
|
|
|
|
|
class QCVerifyDetailView(OpsRequiredMixin, DetailView):
|
|
model = QCResult
|
|
template_name = 'ops/qc_verify_detail.html'
|
|
context_object_name = 'result'
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
from django.contrib import messages
|
|
from apps.compliance.services import create_coa_for_qc
|
|
|
|
qc = self.get_object()
|
|
user = request.user
|
|
if qc.entered_by is not None and qc.entered_by_id == user.id:
|
|
messages.error(request, 'The submitter cannot approve their own QC result (four-eyes control).')
|
|
return redirect(self.request.path)
|
|
|
|
approved = request.POST.get('approved') == '1'
|
|
if approved:
|
|
if not qc.overall_pass:
|
|
messages.error(request, 'A failed QC result cannot be approved; it remains quarantined with an open CAPA.')
|
|
return redirect(self.request.path)
|
|
qc.verification_status = QCResult.VerificationStatus.APPROVED
|
|
qc.verified_by = user
|
|
qc.approved_at = timezone.now()
|
|
qc.save()
|
|
coa = create_coa_for_qc(qc, approved_by=user)
|
|
messages.success(request, f'QC result approved. CoA created (status: {coa.status}).')
|
|
return redirect('ops:coa_review_detail', pk=coa.pk)
|
|
else:
|
|
qc.verification_status = QCResult.VerificationStatus.REJECTED
|
|
qc.verified_by = user
|
|
qc.approved_at = timezone.now()
|
|
qc.save()
|
|
messages.warning(request, 'QC result rejected.')
|
|
return redirect('ops:qc_verify_list')
|
|
|
|
|
|
class COAReviewListView(OpsRequiredMixin, ListView):
|
|
model = CertificateOfAnalysis
|
|
template_name = 'ops/coa_review_list.html'
|
|
context_object_name = 'coas'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = CertificateOfAnalysis.objects.select_related('batch', 'qc_result', 'quality_approved_by').all()
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
qs = qs.filter(status=status)
|
|
else:
|
|
qs = qs.filter(status=CertificateOfAnalysis.Status.PENDING_APPROVAL)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['status_choices'] = CertificateOfAnalysis.Status.choices
|
|
ctx['current_status'] = self.request.GET.get('status', '')
|
|
return ctx
|
|
|
|
|
|
class COAReviewDetailView(OpsRequiredMixin, DetailView):
|
|
model = CertificateOfAnalysis
|
|
template_name = 'ops/coa_review_detail.html'
|
|
context_object_name = 'coa'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['lab_documents'] = Document.objects.filter(batch=self.object.batch, doc_type=Document.DocType.LAB_REPORT)
|
|
return ctx
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
from django.contrib import messages
|
|
from apps.compliance.services import publish_coa
|
|
|
|
coa = self.get_object()
|
|
if request.POST.get('action') == 'approve':
|
|
try:
|
|
publish_coa(coa, released_by=request.user)
|
|
messages.success(request, 'CoA published to buyers.')
|
|
except ValueError as exc:
|
|
messages.error(request, str(exc))
|
|
return redirect(self.request.path)
|
|
|
|
|
|
class EUDRCaseListView(OpsRequiredMixin, ListView):
|
|
model = DueDiligenceCase
|
|
template_name = 'ops/eudr_list.html'
|
|
context_object_name = 'cases'
|
|
paginate_by = 25
|
|
|
|
def get_queryset(self):
|
|
qs = DueDiligenceCase.objects.select_related('batch', 'approved_by').all()
|
|
status = self.request.GET.get('status')
|
|
if status:
|
|
qs = qs.filter(status=status)
|
|
return qs
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
ctx['status_choices'] = DueDiligenceCase.Status.choices
|
|
ctx['current_status'] = self.request.GET.get('status', '')
|
|
return ctx
|
|
|
|
|
|
class EUDRCaseDetailView(OpsRequiredMixin, DetailView):
|
|
model = DueDiligenceCase
|
|
template_name = 'ops/eudr_detail.html'
|
|
context_object_name = 'case'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
ctx = super().get_context_data(**kwargs)
|
|
case = self.object
|
|
ctx['lineage'] = list(case.batch.intake_lineage.select_related(
|
|
'harvest_intake', 'harvest_intake__origin_zone', 'harvest_intake__collector_group'
|
|
))
|
|
ctx['evidence'] = case.evidence.all()
|
|
ctx['risk_assessments'] = case.risk_assessments.all()
|
|
ctx['mitigation_actions'] = case.mitigation_actions.all()
|
|
ctx['statement_references'] = case.statement_references.all()
|
|
return ctx
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
from django.contrib import messages
|
|
from apps.compliance import services
|
|
|
|
case = self.get_object()
|
|
action = request.POST.get('action')
|
|
try:
|
|
if action == 'check':
|
|
services.run_completeness_check(case)
|
|
messages.info(request, 'Completeness check completed.')
|
|
elif action == 'assess_risk':
|
|
services.evaluate_risk(case, user=request.user)
|
|
messages.info(request, 'Risk assessment recorded.')
|
|
elif action == 'approve':
|
|
services.approve_case(case, request.user)
|
|
messages.success(request, 'Case approved.')
|
|
elif action == 'reject':
|
|
services.reject_case(case, request.user, request.POST.get('reason', ''))
|
|
messages.warning(request, 'Case rejected.')
|
|
elif action == 'record_submission':
|
|
ref = request.POST.get('dds_reference')
|
|
if not ref:
|
|
messages.error(request, 'DDS reference required.')
|
|
else:
|
|
services.record_submission(case, request.user, ref, request.POST.get('submission_date') or None)
|
|
messages.success(request, 'External submission recorded.')
|
|
except ValueError as exc:
|
|
messages.error(request, str(exc))
|
|
return redirect(self.request.path)
|
|
|
|
|
|
class GuideIndexView(OpsRequiredMixin, BaseGuideIndexView):
|
|
template_name = 'guides/index_ops.html'
|
|
|
|
|
|
class GuideDetailView(OpsRequiredMixin, BaseGuideDetailView):
|
|
template_name = 'guides/detail_ops.html'
|