import csv import io import os from django.conf import settings from django.db.models.functions import Lower from django.http import HttpResponse from rest_framework import status from rest_framework.decorators import action from rest_framework.viewsets import ModelViewSet from rest_framework.filters import SearchFilter, OrderingFilter from core.responses import success_response, error_response from core.permissions import DjangoModelPermissionsWithView from .models import MasterAnnualCTC from .serializers import MasterAnnualCTCSerializer class MasterAnnualCTCViewSet(ModelViewSet): queryset = MasterAnnualCTC.objects.all() serializer_class = MasterAnnualCTCSerializer permission_classes = [DjangoModelPermissionsWithView] filter_backends = [SearchFilter, OrderingFilter] search_fields = ["name"] ordering_fields = ["name", "created_at"] ordering = ["-created_at"] def get_queryset(self): return MasterAnnualCTC.objects.select_related("created_by", "updated_by").all() def perform_create(self, serializer): serializer.save( created_by=self.request.user, updated_by=self.request.user ) def perform_update(self, serializer): serializer.save(updated_by=self.request.user) def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = self.get_serializer(queryset, many=True) return success_response(serializer.data) def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) return success_response(serializer.data) def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) self.perform_create(serializer) return success_response( serializer.data, "CTC range created successfully", status_code=status.HTTP_201_CREATED ) def update(self, request, *args, **kwargs): partial = kwargs.pop("partial", False) instance = self.get_object() serializer = self.get_serializer(instance, data=request.data, partial=partial) serializer.is_valid(raise_exception=True) self.perform_update(serializer) return success_response(serializer.data, "CTC range updated successfully") def destroy(self, request, *args, **kwargs): instance = self.get_object() self.perform_destroy(instance) return success_response(None, "CTC range deleted successfully") @action(detail=False, methods=["get"], url_path="template") def download_template(self, request): """Serve the master annual CTC CSV template.""" template_dir = os.path.join(settings.BASE_DIR.parent, "frontend", "templates") template_path = os.path.join(template_dir, "master-annual-ctc.csv") if not os.path.exists(template_dir): os.makedirs(template_dir, exist_ok=True) if not os.path.exists(template_path): with open(template_path, "w", newline="", encoding="utf-8") as f: f.write("CTC Range\n1–2 LPA\n3–5 LPA\n6–8 LPA\n8–10 LPA\n10–15 LPA\n") try: with open(template_path, "r", encoding="utf-8") as f: csv_content = f.read() except Exception as e: return error_response(f"Failed to read template: {str(e)}", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) response = HttpResponse(csv_content, content_type="text/csv") response["Content-Disposition"] = 'attachment; filename="master-annual-ctc.csv"' return response @action(detail=False, methods=["post"], url_path="import") def import_csv(self, request): """Parse, validate, and bulk insert master annual CTC ranges from CSV.""" file_obj = request.FILES.get("file") if not file_obj: return error_response("No file was provided.", status_code=status.HTTP_400_BAD_REQUEST) if not file_obj.name.endswith(".csv"): return error_response("Invalid file type. Only CSV files are allowed.", status_code=status.HTTP_400_BAD_REQUEST) try: content = file_obj.read().decode("utf-8-sig") csv_file = io.StringIO(content) reader = csv.reader(csv_file) rows = list(reader) except Exception as e: return error_response(f"Failed to parse CSV: {str(e)}", status_code=status.HTTP_400_BAD_REQUEST) if not rows: return error_response("The uploaded CSV file is empty.", status_code=status.HTTP_400_BAD_REQUEST) import re as _re def _norm(s): return _re.sub(r"[^a-z0-9]", "", (s or "").lower()) norm_header = [_norm(col) for col in rows[0]] aliases = {"ctcrange", "ctc", "name", "annualctc", "range", "label"} ctc_idx = next((i for i, c in enumerate(norm_header) if c in aliases), None) if ctc_idx is None: return error_response( "Missing required header. The CSV must have a column named 'CTC Range' " "(also accepts 'CTC', 'Name', or 'Range'). Download the template for the correct format.", status_code=status.HTTP_400_BAD_REQUEST, ) errors = [] row_items = [] for idx, row in enumerate(rows[1:], start=2): if not row or all(not cell.strip() for cell in row): continue if len(row) <= ctc_idx: errors.append(f"Row {idx} : CTC range is empty") continue name = row[ctc_idx].strip() if not name: errors.append(f"Row {idx} : CTC range is empty") continue if not _re.match(r"^[a-zA-Z0-9\s.\-–+]+$", name): errors.append(f"Row {idx} : CTC range contains invalid special characters") continue row_items.append((idx, name)) all_names = [name for _, name in row_items] existing_names_in_db = set( MasterAnnualCTC.objects.annotate(name_lower=Lower("name")) .filter(name_lower__in=[n.lower() for n in all_names]) .values_list("name_lower", flat=True) ) seen_in_csv_lower = set() ctcs_to_create = [] for idx, name in row_items: name_lower = name.lower() if name_lower in seen_in_csv_lower: errors.append(f"Row {idx} : Duplicate CTC range in CSV") continue if name_lower in existing_names_in_db: errors.append(f"Row {idx} : CTC range already exists") continue seen_in_csv_lower.add(name_lower) ctcs_to_create.append( MasterAnnualCTC( name=name, status="DRAFT", created_by=request.user, updated_by=request.user, ) ) imported_count = 0 if ctcs_to_create: try: MasterAnnualCTC.objects.bulk_create(ctcs_to_create, batch_size=1000) imported_count = len(ctcs_to_create) except Exception as e: return error_response(f"Database insertion failed: {str(e)}", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) return success_response( data={ "imported": imported_count, "skipped": len(errors), "errors": errors, }, message="CSV import completed." ) @action(detail=False, methods=["patch"], url_path="approve") def bulk_approve(self, request): """Bulk approve selected draft master annual CTC items.""" item_ids = request.data.get("ctcIds") or request.data.get("ids") or request.data.get("skillIds") if not item_ids or not isinstance(item_ids, list): return error_response( "Invalid request. 'ctcIds' must be a non-empty list.", status_code=status.HTTP_400_BAD_REQUEST ) try: item_ids = [int(iid) for iid in item_ids] except (ValueError, TypeError): return error_response( "Invalid ID formats. All IDs must be integers.", status_code=status.HTTP_400_BAD_REQUEST ) try: updated_count = MasterAnnualCTC.objects.filter( id__in=item_ids, status="DRAFT" ).update( status="ACTIVE", updated_by=request.user ) except Exception as e: return error_response(f"Database update failed: {str(e)}", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) return success_response( data={"updated_count": updated_count}, message=f"Successfully approved {updated_count} CTC range(s)." ) @action(detail=False, methods=["delete"], url_path="bulk-delete") def bulk_delete(self, request): """Bulk delete selected draft master annual CTC items.""" item_ids = request.data.get("ctcIds") or request.data.get("ids") or request.data.get("skillIds") if not item_ids or not isinstance(item_ids, list): return error_response( "Invalid request. 'ctcIds' must be a non-empty list.", status_code=status.HTTP_400_BAD_REQUEST ) try: item_ids = [int(iid) for iid in item_ids] except (ValueError, TypeError): return error_response( "Invalid ID formats. All IDs must be integers.", status_code=status.HTTP_400_BAD_REQUEST ) try: deleted_count, _ = MasterAnnualCTC.objects.filter( id__in=item_ids, status="DRAFT" ).delete() except Exception as e: return error_response(f"Database deletion failed: {str(e)}", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) return success_response( data={"deleted_count": deleted_count}, message=f"Successfully deleted {deleted_count} CTC range(s)." )