"""BPM REST API views — function-based views with @api_view decorators.""" from __future__ import annotations from decimal import Decimal, InvalidOperation from django.core.exceptions import ObjectDoesNotExist from django.utils.translation import gettext_lazy as _ from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.exceptions import PermissionDenied, ValidationError from rest_framework.request import Request from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated from simorgh.apps.bpm import selectors, services from simorgh.apps.bpm.models import ( ControlPointExecutionOutcome, InstanceStatus, ProcessDefinition, ProcessInstance, StepStatus, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _tenant(request: Request): tenant = getattr(request, "tenant", None) if tenant is None: raise PermissionDenied("No active tenant.") return tenant def _require_perm(request: Request, codename: str) -> None: from simorgh.apps.iam.engine import is_allowed from simorgh.core.context import current_request_context ctx = current_request_context() if not is_allowed(ctx, codename): raise PermissionDenied(codename) # --------------------------------------------------------------------------- # Serialization helpers # --------------------------------------------------------------------------- def _serialize_framework(fw) -> dict: return { "id": fw.pk, "code": fw.code, "name": fw.name, "industry": fw.industry, "version": fw.version, "language": fw.language, "description": fw.description, } def _serialize_element(el, *, children: list | None = None) -> dict: d = { "id": el.id, "pcf_id": el.pcf_id, "hierarchy_id": el.hierarchy_id, "level": el.level, "name_en": el.name_en, "parent_id": el.parent_id, } if children is not None: d["children"] = children return d def _build_element_tree(elements) -> list: """Convert a flat list of elements into a nested tree.""" by_id: dict = {} roots: list = [] for el in elements: by_id[el.id] = _serialize_element(el, children=[]) for el in elements: node = by_id[el.id] if el.parent_id and el.parent_id in by_id: by_id[el.parent_id]["children"].append(node) else: roots.append(node) return roots def _serialize_process(proc, *, detail: bool = False) -> dict: d = { "id": str(proc.public_id), "name": proc.name, "name_fa": proc.name_fa, "hierarchy_id": proc.hierarchy_id, "level": proc.level, "status": proc.status, "version": proc.version, "trigger": proc.trigger, "effective_date": proc.effective_date.isoformat() if proc.effective_date else None, "review_date": proc.review_date.isoformat() if proc.review_date else None, "framework_code": proc.framework.code if proc.framework else None, "pcf_element_id": proc.pcf_element_id, "parent_id": str(proc.parent.public_id) if proc.parent else None, "created_at": proc.created_at.isoformat(), "updated_at": proc.updated_at.isoformat(), } if detail: d["description"] = proc.description return d def _serialize_role(role) -> dict: return { "id": role.pk, "code": role.code, "name": role.name, "name_fa": role.name_fa, "iam_role_code": role.iam_role_code, "order": role.order, } def _serialize_raci_entry(entry) -> dict: return { "id": entry.pk, "step_id": entry.step_id, "step_title": entry.step.title if entry.step else None, "role_id": entry.role_id, "role_code": entry.role.code if entry.role else None, "role_name": entry.role.name if entry.role else None, "responsibility": entry.responsibility, "activity_label": entry.activity_label, "notes": entry.notes, } def _serialize_kpi(kpi) -> dict: return { "id": str(kpi.public_id), "code": kpi.code, "name": kpi.name, "name_fa": kpi.name_fa, "category": kpi.category, "formula": kpi.formula, "target_value": str(kpi.target_value), "target_operator": kpi.target_operator, "unit": kpi.unit, "frequency": kpi.frequency, "owner_role": _serialize_role(kpi.owner_role) if kpi.owner_role else None, } def _serialize_rule(rule) -> dict: return { "id": rule.pk, "code": rule.code, "description": rule.description, "source": rule.source, "is_configurable": rule.is_configurable, "order": rule.order, } def _serialize_control_point(cp) -> dict: return { "id": cp.pk, "code": cp.code, "stage": cp.stage, "description": cp.description, "output_description": cp.output_description, "responsible_role": _serialize_role(cp.responsible_role) if cp.responsible_role else None, } def _serialize_io(io_link) -> dict: item = io_link.data_item return { "id": io_link.pk, "direction": io_link.direction, "is_required": io_link.is_required, "notes": io_link.notes, "data_item": { "id": str(item.public_id), "code": item.code, "name": item.name, "description": item.description, "is_external": item.is_external, "format": item.format, }, } def _serialize_instance(inst, *, detail: bool = False) -> dict: d = { "id": str(inst.public_id), "instance_number": inst.instance_number, "title": inst.title, "status": inst.status, "process_id": str(inst.process.public_id), "process_name": inst.process.name, "triggered_by": inst.triggered_by, "start_date": inst.start_date.isoformat() if inst.start_date else None, "end_date": inst.end_date.isoformat() if inst.end_date else None, "actual_end_date": inst.actual_end_date.isoformat() if inst.actual_end_date else None, "created_by": inst.created_by_id, "notes": inst.notes, "created_at": inst.created_at.isoformat(), } if detail: steps = selectors.list_instance_steps(inst) rule_checks = selectors.list_instance_rule_checks(inst) cp_executions = selectors.list_instance_cp_executions(inst) d["steps"] = [_serialize_instance_step(s) for s in steps] d["rule_checks"] = [_serialize_rule_check(rc) for rc in rule_checks] d["cp_executions"] = [_serialize_cp_execution(cp) for cp in cp_executions] return d def _serialize_instance_step(inst_step) -> dict: return { "id": inst_step.pk, "step_id": inst_step.step_id, "step_number": inst_step.step.step_number, "step_title": inst_step.step.title, "status": inst_step.status, "started_at": inst_step.started_at.isoformat() if inst_step.started_at else None, "completed_at": inst_step.completed_at.isoformat() if inst_step.completed_at else None, "assigned_to_id": inst_step.assigned_to_id, "notes": inst_step.notes, "output_document_ids": inst_step.output_document_ids, } def _serialize_rule_check(rc) -> dict: return { "id": rc.pk, "rule_id": rc.rule_id, "rule_code": rc.rule.code if rc.rule else None, "rule_description": rc.rule.description if rc.rule else None, "is_compliant": rc.is_compliant, "checked_at": rc.checked_at.isoformat() if rc.checked_at else None, "checked_by_id": rc.checked_by_id, "notes": rc.notes, } def _serialize_cp_execution(execution) -> dict: return { "id": execution.pk, "control_point_id": execution.control_point_id, "control_point_code": execution.control_point.code if execution.control_point else None, "outcome": execution.outcome, "executed_at": execution.executed_at.isoformat() if execution.executed_at else None, "executed_by_id": execution.executed_by_id, "notes": execution.notes, } # --------------------------------------------------------------------------- # Framework views # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def framework_list(request: Request) -> Response: frameworks = selectors.list_frameworks() return Response( {"results": [_serialize_framework(f) for f in frameworks], "count": frameworks.count()} ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def framework_elements(request: Request, code: str) -> Response: try: framework = selectors.get_framework_by_code(code) except ObjectDoesNotExist: return Response({"detail": "Framework not found."}, status=status.HTTP_404_NOT_FOUND) elements = list(selectors.list_framework_elements(framework)) tree = _build_element_tree(elements) return Response({"framework": _serialize_framework(framework), "elements": tree}) # --------------------------------------------------------------------------- # Process views # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def process_list_create(request: Request) -> Response: tenant = _tenant(request) if request.method == "GET": _require_perm(request, "bpm.processdefinition.view") qs = selectors.list_processes(tenant) return Response( {"results": [_serialize_process(p) for p in qs], "count": qs.count()} ) # POST _require_perm(request, "bpm.processdefinition.add") allowed_fields = { "name", "name_fa", "description", "hierarchy_id", "level", "status", "version", "trigger", "effective_date", "review_date", "framework_id", "pcf_element_id", "parent_id", } data = {k: v for k, v in request.data.items() if k in allowed_fields} if not data.get("name"): raise ValidationError({"name": "This field is required."}) process = services.create_process(tenant, data) return Response(_serialize_process(process, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET", "PATCH"]) @permission_classes([IsAuthenticated]) def process_detail(request: Request, public_id) -> Response: tenant = _tenant(request) try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, "bpm.processdefinition.view") return Response(_serialize_process(process, detail=True)) # PATCH _require_perm(request, "bpm.processdefinition.change") allowed_fields = { "name", "name_fa", "description", "status", "version", "trigger", "effective_date", "review_date", } data = {k: v for k, v in request.data.items() if k in allowed_fields} process = services.update_process(process, data) return Response(_serialize_process(process, detail=True)) @api_view(["GET", "PUT"]) @permission_classes([IsAuthenticated]) def process_raci(request: Request, public_id) -> Response: tenant = _tenant(request) try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, "bpm.racientry.view") entries = selectors.list_process_raci(process) return Response( {"results": [_serialize_raci_entry(e) for e in entries], "count": entries.count()} ) # PUT — bulk upsert _require_perm(request, "bpm.racientry.add") entries_data = request.data.get("entries", []) if not isinstance(entries_data, list): raise ValidationError({"entries": "Must be a list."}) created = services.bulk_upsert_raci(process, entries_data) return Response( {"results": [_serialize_raci_entry(e) for e in created], "count": len(created)}, status=status.HTTP_200_OK, ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_kpis(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processkpi.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_kpis(process) return Response({"results": [_serialize_kpi(k) for k in qs], "count": qs.count()}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_rules(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processrule.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_rules(process) return Response({"results": [_serialize_rule(r) for r in qs], "count": qs.count()}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_control_points(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processcontrolpoint.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_control_points(process) return Response({"results": [_serialize_control_point(cp) for cp in qs], "count": qs.count()}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_io(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processio.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) direction = request.query_params.get("direction") qs = selectors.list_process_io(process, direction=direction) return Response({"results": [_serialize_io(io) for io in qs], "count": qs.count()}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_roles(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.racientry.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_roles(process) return Response({"results": [_serialize_role(r) for r in qs], "count": qs.count()}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_steps(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processdefinition.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_steps(process) return Response({ "results": [ { "id": s.pk, "step_number": s.step_number, "title": s.title, "title_fa": s.title_fa, "description": s.description, "output_description": s.output_description, } for s in qs ], "count": qs.count(), }) # --------------------------------------------------------------------------- # Instance views # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def instance_list_create(request: Request) -> Response: tenant = _tenant(request) if request.method == "GET": _require_perm(request, "bpm.processinstance.view") qs = selectors.list_instances(tenant) return Response( {"results": [_serialize_instance(i) for i in qs], "count": qs.count()} ) # POST _require_perm(request, "bpm.processinstance.add") process_id = request.data.get("process_id") if not process_id: raise ValidationError({"process_id": "This field is required."}) try: process = selectors.get_process(process_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) allowed_fields = {"title", "triggered_by", "status", "start_date", "end_date", "notes", "workspace_id"} data = {k: v for k, v in request.data.items() if k in allowed_fields} instance = services.create_instance(tenant, process, data, request.user) return Response(_serialize_instance(instance, detail=True), status=status.HTTP_201_CREATED) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def instance_detail(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processinstance.view") try: instance = selectors.get_instance(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Instance not found."}, status=status.HTTP_404_NOT_FOUND) return Response(_serialize_instance(instance, detail=True)) @api_view(["PATCH"]) @permission_classes([IsAuthenticated]) def instance_step_update(request: Request, public_id, step_id: int) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processinstancestep.change") try: instance = selectors.get_instance(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Instance not found."}, status=status.HTTP_404_NOT_FOUND) allowed_fields = {"status", "assigned_to_id", "notes", "started_at", "completed_at", "output_document_ids"} data = {k: v for k, v in request.data.items() if k in allowed_fields} try: inst_step = services.update_instance_step(instance, step_id, data) except ObjectDoesNotExist: return Response({"detail": "Step not found."}, status=status.HTTP_404_NOT_FOUND) return Response(_serialize_instance_step(inst_step)) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def instance_cp_execute(request: Request, public_id, cp_code: str) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.controlpointexecution.add") try: instance = selectors.get_instance(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Instance not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.bpm.models import ProcessControlPoint # noqa: PLC0415 try: cp = ProcessControlPoint.objects.get(process=instance.process, code=cp_code) except ObjectDoesNotExist: return Response({"detail": "Control point not found."}, status=status.HTTP_404_NOT_FOUND) outcome = request.data.get("outcome") valid_outcomes = [o.value for o in ControlPointExecutionOutcome] if outcome not in valid_outcomes: raise ValidationError({"outcome": f"Must be one of: {valid_outcomes}"}) notes = request.data.get("notes", "") execution = services.execute_control_point(instance, cp, outcome, request.user, notes) return Response(_serialize_cp_execution(execution), status=status.HTTP_201_CREATED) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def instance_kpi_measure(request: Request, public_id, kpi_code: str) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processkpimeasurement.add") try: instance = selectors.get_instance(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Instance not found."}, status=status.HTTP_404_NOT_FOUND) try: kpi = selectors.list_process_kpis(instance.process).get(code=kpi_code) except ObjectDoesNotExist: return Response({"detail": "KPI not found."}, status=status.HTTP_404_NOT_FOUND) raw_value = request.data.get("value") if raw_value is None: raise ValidationError({"value": "This field is required."}) try: value = Decimal(str(raw_value)) except InvalidOperation: raise ValidationError({"value": "Must be a numeric value."}) notes = request.data.get("notes", "") measurement = services.record_kpi_measurement( kpi, value, request.user, notes, process_instance_id=instance.public_id, ) return Response( { "id": measurement.pk, "kpi_code": kpi.code, "value": str(measurement.value), "measured_at": measurement.measured_at.isoformat(), "is_target_met": measurement.is_target_met, }, status=status.HTTP_201_CREATED, ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def instance_kpi_gauges(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processinstance.view") try: instance = selectors.get_instance(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Instance not found."}, status=status.HTTP_404_NOT_FOUND) gauges = selectors.list_instance_kpi_gauges(instance) return Response({"results": gauges, "count": len(gauges)}) # --------------------------------------------------------------------------- # KPI dashboard # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def kpi_dashboard(request: Request) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processkpi.view") data = selectors.get_kpi_dashboard(tenant) return Response(data) # --------------------------------------------------------------------------- # Benchmarking (Phase 13) # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def benchmark_list(request: Request) -> Response: _require_perm(request, "bpm.apqcbenchmark.view") qs = selectors.list_benchmarks( pcf_metric=request.query_params.get("pcf_metric"), industry=request.query_params.get("industry"), ) data = [ { "id": b.pk, "pcf_metric_id": b.pcf_metric_id, "pcf_metric_name": b.pcf_metric.name, "industry": b.industry, "year": b.year, "source": b.source, "unit": b.unit, "percentile_25": str(b.percentile_25), "percentile_50": str(b.percentile_50), "percentile_75": str(b.percentile_75), "percentile_90": str(b.percentile_90), } for b in qs ] return Response({"results": data, "count": len(data)}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_benchmark_comparisons(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.orgbenchmarkcomparison.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_benchmark_comparisons(process) data = [ { "id": str(c.public_id), "process_id": str(process.public_id), "kpi_code": c.kpi.code, "kpi_name": c.kpi.name, "benchmark_id": c.benchmark_id, "org_value": str(c.org_value), "percentile_position": float(c.percentile_position) if c.percentile_position is not None else None, "gap_to_median": str(c.gap_to_median) if c.gap_to_median is not None else None, "assessment_date": c.assessment_date.isoformat(), "notes": c.notes, } for c in qs ] return Response({"results": data, "count": len(data)}) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def process_benchmark_comparisons_create(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.orgbenchmarkcomparison.add") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.bpm.models import APQCBenchmark, ProcessKPI # noqa: PLC0415 kpi_code = request.data.get("kpi_code") benchmark_id = request.data.get("benchmark_id") org_value = request.data.get("org_value") notes = request.data.get("notes", "") if not all([kpi_code, benchmark_id, org_value]): return Response({"detail": "kpi_code, benchmark_id and org_value are required."}, status=status.HTTP_400_BAD_REQUEST) try: kpi = ProcessKPI.objects.get(process=process, code=kpi_code) benchmark = APQCBenchmark.objects.get(pk=benchmark_id) except (ProcessKPI.DoesNotExist, APQCBenchmark.DoesNotExist): return Response({"detail": "KPI or benchmark not found."}, status=status.HTTP_404_NOT_FOUND) comparison = services.create_benchmark_comparison( process=process, kpi=kpi, benchmark=benchmark, org_value=org_value, user=request.user, notes=notes, ) return Response( { "id": str(comparison.public_id), "kpi_code": kpi.code, "org_value": str(comparison.org_value), "percentile_position": float(comparison.percentile_position) if comparison.percentile_position is not None else None, "gap_to_median": str(comparison.gap_to_median) if comparison.gap_to_median is not None else None, "assessment_date": comparison.assessment_date.isoformat(), }, status=status.HTTP_201_CREATED, ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_benchmark_summary(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.orgbenchmarkcomparison.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) summary = selectors.get_benchmark_summary(process) return Response({"results": summary, "count": len(summary)}) # --------------------------------------------------------------------------- # Maturity Assessment (Phase 14) # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def maturity_levels(request: Request) -> Response: _require_perm(request, "bpm.maturitylevel.view") levels = selectors.list_maturity_levels() data = [ { "level": ml.level, "name": ml.name, "name_fa": ml.name_fa, "description": ml.description, "description_fa": ml.description_fa, "criteria": ml.criteria, } for ml in levels ] return Response({"results": data, "count": len(data)}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_maturity_assessments(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processmaturityassessment.view") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) qs = selectors.list_process_maturity_assessments(process) data = [ { "id": str(a.public_id), "process_id": str(process.public_id), "assessed_by": a.assessed_by.get_full_name() if a.assessed_by else None, "assessment_date": a.assessment_date.isoformat(), "current_level": a.current_level, "target_level": a.target_level, "gap_to_target": a.gap_to_target, "dimension_scores": a.dimension_scores, "improvement_notes": a.improvement_notes, "next_review_date": a.next_review_date.isoformat() if a.next_review_date else None, "average_score": a.average_dimension_score(), } for a in qs ] return Response({"results": data, "count": len(data)}) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def process_maturity_assessments_create(request: Request, public_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processmaturityassessment.add") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) current_level = request.data.get("current_level") target_level = request.data.get("target_level") dimension_scores = request.data.get("dimension_scores", {}) improvement_notes = request.data.get("improvement_notes", "") next_review_date = request.data.get("next_review_date") if not current_level or not target_level: return Response({"detail": "current_level and target_level are required."}, status=status.HTTP_400_BAD_REQUEST) assessment = services.create_maturity_assessment( process=process, assessed_by=request.user, current_level=int(current_level), target_level=int(target_level), dimension_scores=dimension_scores, improvement_notes=improvement_notes, next_review_date=next_review_date, ) return Response( { "id": str(assessment.public_id), "current_level": assessment.current_level, "target_level": assessment.target_level, "gap_to_target": assessment.gap_to_target, "assessment_date": assessment.assessment_date.isoformat(), }, status=status.HTTP_201_CREATED, ) @api_view(["GET", "PATCH"]) @permission_classes([IsAuthenticated]) def process_maturity_assessment_detail(request: Request, public_id, assessment_id) -> Response: tenant = _tenant(request) try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.bpm.models import ProcessMaturityAssessment # noqa: PLC0415 try: assessment = ProcessMaturityAssessment.objects.get(public_id=assessment_id, process=process) except ProcessMaturityAssessment.DoesNotExist: return Response({"detail": "Assessment not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": _require_perm(request, "bpm.processmaturityassessment.view") actions_qs = selectors.list_maturity_improvement_actions(assessment) return Response({ "id": str(assessment.public_id), "process_id": str(process.public_id), "assessed_by": assessment.assessed_by.get_full_name() if assessment.assessed_by else None, "assessment_date": assessment.assessment_date.isoformat(), "current_level": assessment.current_level, "target_level": assessment.target_level, "gap_to_target": assessment.gap_to_target, "dimension_scores": assessment.dimension_scores, "improvement_notes": assessment.improvement_notes, "next_review_date": assessment.next_review_date.isoformat() if assessment.next_review_date else None, "average_score": assessment.average_dimension_score(), "actions": [ { "id": str(a.public_id), "title": a.title, "description": a.description, "target_level": a.target_level, "due_date": a.due_date.isoformat() if a.due_date else None, "owner": a.owner.get_full_name() if a.owner else None, "status": a.status, } for a in actions_qs ], }) # PATCH _require_perm(request, "bpm.processmaturityassessment.change") updated = services.update_maturity_assessment( assessment, current_level=request.data.get("current_level"), target_level=request.data.get("target_level"), dimension_scores=request.data.get("dimension_scores"), improvement_notes=request.data.get("improvement_notes"), next_review_date=request.data.get("next_review_date"), ) return Response({ "id": str(updated.public_id), "current_level": updated.current_level, "target_level": updated.target_level, "gap_to_target": updated.gap_to_target, "dimension_scores": updated.dimension_scores, }) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def process_maturity_action_create(request: Request, public_id, assessment_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.maturityimprovementaction.add") try: process = selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.bpm.models import ProcessMaturityAssessment # noqa: PLC0415 try: assessment = ProcessMaturityAssessment.objects.get(public_id=assessment_id, process=process) except ProcessMaturityAssessment.DoesNotExist: return Response({"detail": "Assessment not found."}, status=status.HTTP_404_NOT_FOUND) title = request.data.get("title") target_level = request.data.get("target_level") if not title or not target_level: return Response({"detail": "title and target_level are required."}, status=status.HTTP_400_BAD_REQUEST) action = services.create_improvement_action( assessment=assessment, title=title, description=request.data.get("description", ""), target_level=int(target_level), due_date=request.data.get("due_date"), owner=request.user, ) return Response( { "id": str(action.public_id), "title": action.title, "target_level": action.target_level, "status": action.status, }, status=status.HTTP_201_CREATED, ) @api_view(["PATCH"]) @permission_classes([IsAuthenticated]) def process_maturity_action_status(request: Request, public_id, assessment_id, action_id) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.maturityimprovementaction.change") try: selectors.get_process(public_id, tenant) except ObjectDoesNotExist: return Response({"detail": "Process not found."}, status=status.HTTP_404_NOT_FOUND) from simorgh.apps.bpm.models import MaturityImprovementAction # noqa: PLC0415 try: action = MaturityImprovementAction.objects.get(public_id=action_id, assessment__public_id=assessment_id) except MaturityImprovementAction.DoesNotExist: return Response({"detail": "Action not found."}, status=status.HTTP_404_NOT_FOUND) new_status = request.data.get("status") if not new_status: return Response({"detail": "status is required."}, status=status.HTTP_400_BAD_REQUEST) try: updated = services.update_improvement_action_status(action, new_status) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response({"id": str(updated.public_id), "status": updated.status}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def maturity_heatmap(request: Request) -> Response: tenant = _tenant(request) _require_perm(request, "bpm.processmaturityassessment.view") data = selectors.get_org_maturity_heatmap(tenant) return Response({"results": data, "count": len(data)}) # --------------------------------------------------------------------------- # AI Process Advisor (Phase 15) # --------------------------------------------------------------------------- @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_ai_kpi_anomalies(request: Request, public_id) -> Response: """GET /bpm/processes/{id}/ai/kpi-anomalies/ Detect KPIs that miss their target and return AI-generated suggestions. """ from simorgh.apps.bpm import ai_advisor tenant = _tenant(request) _require_perm(request, "bpm.processdefinition.view") try: process = selectors.get_process(public_id, tenant) except Exception: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) user_id = request.user.pk if request.user.is_authenticated else None anomalies = ai_advisor.detect_kpi_anomalies(process, user_id=user_id) data = [ { "kpi_code": a.kpi_code, "kpi_name": a.kpi_name, "current_value": a.current_value, "target_value": a.target_value, "target_operator": a.target_operator, "unit": a.unit, "gap_percent": a.gap_percent, "ai_suggestions": a.ai_suggestions, } for a in anomalies ] return Response({"results": data, "count": len(data)}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_ai_raci_suggest(request: Request, public_id) -> Response: """GET /bpm/processes/{id}/ai/raci-suggest/ Generate RACI assignment suggestions based on PCF best practice. """ from simorgh.apps.bpm import ai_advisor tenant = _tenant(request) _require_perm(request, "bpm.processdefinition.view") try: process = selectors.get_process(public_id, tenant) except Exception: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) user_id = request.user.pk if request.user.is_authenticated else None suggestions = ai_advisor.suggest_raci(process, user_id=user_id) data = [ { "step_number": s.step_number, "step_title": s.step_title, "responsible": s.responsible, "accountable": s.accountable, "consulted": s.consulted, "informed": s.informed, "notes": s.notes, } for s in suggestions ] return Response({"results": data, "count": len(data)}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_ai_gaps(request: Request, public_id) -> Response: """GET /bpm/processes/{id}/ai/gaps/ Identify gaps between org process and PCF standard. """ from simorgh.apps.bpm import ai_advisor tenant = _tenant(request) _require_perm(request, "bpm.processdefinition.view") try: process = selectors.get_process(public_id, tenant) except Exception: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) user_id = request.user.pk if request.user.is_authenticated else None gaps = ai_advisor.analyze_process_gaps(process, user_id=user_id) data = [ { "gap_type": g.gap_type, "title": g.title, "description": g.description, "priority": g.priority, "recommendation": g.recommendation, } for g in gaps ] return Response({"results": data, "count": len(data)}) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def process_ai_recommendations(request: Request, public_id) -> Response: """GET /bpm/processes/{id}/ai/recommendations/ Improvement recommendations combining maturity + KPI gap data. """ from simorgh.apps.bpm import ai_advisor tenant = _tenant(request) _require_perm(request, "bpm.processdefinition.view") try: process = selectors.get_process(public_id, tenant) except Exception: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) user_id = request.user.pk if request.user.is_authenticated else None recs = ai_advisor.get_improvement_recommendations(process, user_id=user_id) data = [ { "rank": r.rank, "title": r.title, "rationale": r.rationale, "impact": r.impact, "owner_role": r.owner_role, "source": r.source, } for r in recs ] return Response({"results": data, "count": len(data)}) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def ai_nl_query(request: Request) -> Response: """POST /bpm/ai/query/ Answer a natural-language question about the tenant's BPM data. Body: {"question": "..."} """ from simorgh.apps.bpm import ai_advisor tenant = _tenant(request) question = (request.data.get("question") or "").strip() if not question: return Response({"detail": "question is required."}, status=status.HTTP_400_BAD_REQUEST) user_id = request.user.pk if request.user.is_authenticated else None result = ai_advisor.run_nl_query(question, tenant, user_id=user_id) return Response({ "question": result.question, "answer": result.answer, "data_snapshot": result.data_snapshot, })