import calendar from datetime import date, timedelta from decimal import Decimal from management.management.utils.utils import create_audit_log from management.models.fascicolo import Fascicolo from management.models.ore_plan import OrePlan from management.models.plan_archivio import TipoDocumento from management.models.pratica import Pratica from management.models.stato_pratica import StatoPratica from management.models.tipo_commessa import TipoCommessa from management.models.utente import Utente from management.repositories.ore_plan import OrePlanRepository from management.serializers.plan_archivio_serializer import PlanArchivioAllegatiListSerializer, PlanArchivioGetItemsSerializer, TipoDocumentoSerializer from management.views import pratica from .base import * from django.db.models import F, Q, Avg, Case, CharField, Count, DurationField, ExpressionWrapper, FloatField, IntegerField, DecimalField, Sum, Value, When from django.db.models.functions import Coalesce, Concat, Trim, TruncMonth from dateutil.relativedelta import relativedelta from workalendar.europe import Italy from django.contrib.contenttypes.models import ContentType class PlanArchivioAPIView(GenericViewSet): authentication_classes = [TokenAuthentication] _repository = PlanArchivioRepository() _repository_ore_plan = OrePlanRepository() @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_items(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) query_set = self._repository.entity.objects.all() date_from_value = None date_to_value = None remaining_filters = [] original_filters = getattr(args, "filters", []) for f in original_filters: if f.field_name == "date_from": date_from_value = f.filter_value elif f.field_name == "date_to": date_to_value = f.filter_value else: remaining_filters.append(f) oggi = datetime.today() if date_from_value and date_to_value: if isinstance(date_from_value, str): d_start_obj = datetime.fromisoformat(date_from_value.split("T")[0]) else: d_start_obj = date_from_value if isinstance(date_to_value, str): d_end_obj = datetime.fromisoformat(date_to_value.split("T")[0]) else: d_end_obj = date_to_value d_end_obj = d_end_obj + relativedelta(months=1) cal = Italy() query_set = query_set.annotate( year_month=ExpressionWrapper( F("anno_di_lavorazione") * 100 + F("mese_di_lavorazione"), output_field=IntegerField() ) ) start_value = d_start_obj.year * 100 + d_start_obj.month end_value = d_end_obj.year * 100 + d_end_obj.month query_set = query_set.filter( year_month__gt=start_value, year_month__lte=end_value ) clean_args = FilterAndSorting() clean_args.filters = remaining_filters clean_args.sortings = getattr(args, "sortings", []) filtered_set = parse_filters_and_sortings(clean_args, query_set, user_id=None) filtered_set = filtered_set.annotate( cliente_title=F("cliente__title"), stato_pratica_title=F("stato_pratica__title"), incaricato_full_name=Concat( F("incaricato__nome"), Value(" "), F("incaricato__cognome"), output_field=CharField() ), tipo_documento = F("documenti__nome") ) oggi = timezone.now().date() stats = filtered_set.aggregate( totale_prese_in_carico=Count('id', filter=Q(preso_in_carico=1)), totale_aperte=Count('id', filter=Q(stato_pratica__codice='TO_DO')), totale_review=Count('id', filter=Q(stato_pratica__codice='REV')), completate=Count('id', filter=Q(stato_pratica__codice='CHI') | Q(stato_pratica__codice='NO_LAV')), totale_scadute=Count('id', filter=Q(data_di_scadenza__lt=oggi)), totale_scadenza_oggi=Count('id', filter=Q(data_di_scadenza=oggi)) ) lines = list(filtered_set.values()) return Response({ "totalCount": len(lines), "lines": lines, "stats": stats }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_items_archivio(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) filtered_set = self._repository.filtered(args).select_related('cliente', 'stato_pratica', 'incaricato', 'documenti') filtered_set = filtered_set.annotate( cliente_title=Coalesce(F("cliente__title"), Value("Cliente non definito")), stato_pratica_title=Coalesce(F("stato_pratica__title"), Value("Stato non definito")), incaricato_full_name=Case( When(incaricato__isnull=True, then=Value("Nessun incaricato")), default=Concat( Coalesce(F("incaricato__nome"), Value("")), Value(" "), Coalesce(F("incaricato__cognome"), Value("")) ), output_field=CharField() ), tipo_documento = F("documenti__nome") ) oggi = timezone.now().date() stats = filtered_set.aggregate( totale_prese_in_carico=Count('id', filter=Q(preso_in_carico=1)), totale_aperte=Count('id', filter=Q(stato_pratica__codice='TO_DO')), totale_review=Count('id', filter=Q(stato_pratica__codice='REV')), completate=Count('id', filter=Q(stato_pratica__codice='CHI') | Q(stato_pratica__codice='NO_LAV')), totale_scadute=Count('id', filter=Q(data_di_scadenza__lt=oggi)), totale_scadenza_oggi=Count('id', filter=Q(data_di_scadenza=oggi)) ) lines = list(filtered_set.values()) return Response({ "totalCount": len(lines), "lines": lines, "stats": stats }) except Exception as e: handle_exception(e) def get_fisc_context(self, date_obj): def first_monday(d): first_day = d.replace(day=1) return first_day + timedelta(days=(0 - first_day.weekday()) % 7) current = date_obj while True: fm = first_monday(current) if current < fm: current = current.replace(day=1) - timedelta(days=1) continue diff_days = (date_obj - fm).days week_num = (diff_days // 7) + 1 return current.year, current.month, week_num @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_items_week(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) query_set = self._repository.entity.objects.all() date_from_value = None date_to_value = None remaining_filters = [] original_filters = getattr(args, "filters", []) for f in original_filters: if f.field_name == "date_from": date_from_value = f.filter_value elif f.field_name == "date_to": date_to_value = f.filter_value else: remaining_filters.append(f) if date_from_value and date_to_value: if isinstance(date_from_value, str): d_start_obj = datetime.fromisoformat(date_from_value.split("T")[0]) + timedelta(days=1) else: d_start_obj = date_from_value + timedelta(days=1) if isinstance(date_to_value, str): d_end_obj = datetime.fromisoformat(date_to_value.split("T")[0]) else: d_end_obj = date_to_value sy, sm, sw = self.get_fisc_context(d_start_obj) ey, em, ew = self.get_fisc_context(d_end_obj) if sm == em and sy == ey: query_set = query_set.filter( anno_di_lavorazione=sy, mese_di_lavorazione=sm, settimana_di_lavorazione=sw ) else: query_set = query_set.filter( Q(anno_di_lavorazione=sy, mese_di_lavorazione=sm, settimana_di_lavorazione=sw) | Q(anno_di_lavorazione=ey, mese_di_lavorazione=em, settimana_di_lavorazione=ew) ) clean_args = FilterAndSorting() clean_args.filters = remaining_filters clean_args.sortings = getattr(args, "sortings", []) filtered_set = parse_filters_and_sortings(clean_args, query_set, user_id=None) filtered_set = filtered_set.annotate( cliente_title=F("cliente__title"), stato_pratica_title=F("stato_pratica__title"), incaricato_full_name=Concat( F("incaricato__nome"), Value(" "), F("incaricato__cognome"), output_field=CharField() ), tipo_documento = F("documenti__nome") ) oggi = timezone.now().date() stats = filtered_set.aggregate( totale_prese_in_carico=Count('id', filter=Q(preso_in_carico=1)), totale_aperte=Count('id', filter=Q(stato_pratica__codice='TO_DO')), totale_review=Count('id', filter=Q(stato_pratica__codice='REV')), completate=Count('id', filter=Q(stato_pratica__codice='CHI') | Q(stato_pratica__codice='NO_LAV')), totale_scadute=Count('id', filter=Q(data_di_scadenza__lt=oggi)), totale_scadenza_oggi=Count('id', filter=Q(data_di_scadenza=oggi)) ) lines = list(filtered_set.values()) return Response({ "totalCount": len(lines), "lines": lines, "stats": stats }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_filters_list(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request._request.body, object_hook=objJsonDecode) filtered_set = self._repository.filtered(args) data_cliente_ids = filtered_set.values_list('cliente', flat=True).distinct() data_pratica_ids = filtered_set.values_list('pratica', flat=True).distinct() data_fascicolo_ids = filtered_set.values_list('fascicolo', flat=True).distinct() data_commessa_ids = filtered_set.values_list('commessa', flat=True).distinct() data_incaricato_ids = filtered_set.values_list('incaricato', flat=True).distinct() data_stato_ids = filtered_set.values_list('stato_pratica', flat=True).distinct() data_documenti_ids = filtered_set.values_list('documenti', flat=True).distinct() data_gestore_ids = filtered_set.values_list('gestore', flat=True).distinct() clienti_data = list( Cliente.objects.filter(id__in=data_cliente_ids) .values('id', 'title', 'gruppo', 'modello') .distinct() ) data_gruppo_ids_from_clienti = {c['gruppo'] for c in clienti_data if c['gruppo']} data_modello_ids_from_clienti = {c['modello'] for c in clienti_data if c['modello']} pratica_data = list( Pratica.objects.filter(id__in=data_pratica_ids) .values('id', 'title') .distinct() ) gruppo_data = list( Gruppo.objects.filter(id__in=data_gruppo_ids_from_clienti) .values('id', 'title') .distinct() ) modello_data = list( Modello.objects.filter(id__in=data_modello_ids_from_clienti) .values('id', 'title') .distinct() ) commessa_data = list(TipoCommessa.objects.filter(id__in=data_commessa_ids).values('id', 'title')) fascicolo_data = list(Fascicolo.objects.filter(id__in=data_fascicolo_ids).values('id', 'title')) data_anno = list(filtered_set.order_by('anno_di_competenza').values_list('anno_di_competenza', flat=True).distinct()) data_anno_lavorazione = list(filtered_set.values_list('anno_di_lavorazione', flat=True).distinct().order_by('-anno_di_lavorazione')) data_gestore = list(Utente.objects.filter(id__in=data_gestore_ids).values('id', 'nome', 'cognome', 'role')) data_incaricato = list(Utente.objects.filter(id__in=data_incaricato_ids).values('id', 'nome', 'cognome', 'role')) data_stato = list(StatoPratica.objects.filter(id__in=data_stato_ids).values('id', 'title')) data_documenti = list(TipoDocumento.objects.filter(id__in=data_documenti_ids).values('id', 'nome')) data_anno_standard = list(filtered_set.order_by('anno_standard').values_list('anno_standard', flat=True).distinct()) data_mese_standard = list(filtered_set.order_by('mese_standard').values_list('mese_standard', flat=True).distinct()) data_periodo_competenza = list(filtered_set.order_by('periodo_di_competenza').values_list('periodo_di_competenza', flat=True).distinct()) data_tipo_pratica = list(filtered_set.order_by().values_list('tipo_pratica', flat=True).distinct()) autorizzazione_fattura_list = list( filtered_set .order_by() .values_list('autorizzazione_fattura', flat=True) .distinct() ) return Response({ "anno_di_competenza_list": data_anno, "anno_di_lavorazione_list": data_anno_lavorazione, "gestore_list": data_gestore, "incaricato_list": data_incaricato, "pratica_list": pratica_data, "cliente_list": clienti_data, "stato_list": data_stato, "tipo_pratica_list": data_tipo_pratica, "fascicolo_list": fascicolo_data, "commessa_list": commessa_data, "modello_list": modello_data, "gruppo_list": gruppo_data, "anno_standard_list": data_anno_standard, "mese_standard_list": data_mese_standard, "periodo_competenza_list": data_periodo_competenza, "periodo_standard_list": None, "autorizzazione_fattura_list": autorizzazione_fattura_list, "documenti_list":data_documenti }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_item_by_id(self, request: Request, pk: int): try: plan_archivio = PlanArchivio.objects.get(id=pk) serializer = PlanArchivioGetItemsSerializer(plan_archivio) return Response(serializer.data, status=status.HTTP_200_OK) except Pratica.DoesNotExist: return Response({'detail': 'Plan archivio non trovato.'}, status=status.HTTP_404_NOT_FOUND) except Exception as e: print(e) return Response({'detail': 'An error occurred.'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_plan_attachments(self, request, pk=None): attachments = PlanArchivioAllegati.objects.filter(plan_id=pk) serializer = PlanArchivioAllegatiListSerializer(attachments, many=True) return Response(serializer.data) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def save_item(self, request: Request): try: with transaction.atomic(): dto = request.data.dict() if hasattr(request.data, 'dict') else dict(request.data) dto["data_di_scadenza"] = dto.get('data_di_scadenza').split('T')[0] status_result, errors = self._repository.save(dto, request.user_id) if not status_result: return Response({'detail': 'An error occurred.'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) return Response(GenericResponseObjectSerializer(GenericResponseObject(status=True)).data) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def take_charge(self, request: Request, pk): try: old_values = { "preso_in_carico": 0 } new_values = { "preso_in_carico": { "old": old_values["preso_in_carico"], "new": 1 } } PlanArchivio.objects.filter(id=pk).update(preso_in_carico=1) create_audit_log( user=request.user_id, activity_type='MODIFICA', entity_name='Pratica', entity_id=pk, entity=PlanArchivio.objects.get(id=pk), old_values=old_values, new_values=new_values ) return Response({"status": True, "message": "Pratica presa in carico"}) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def close_plan(self, request: Request, pk): try: old_values = { "stato_pratica": PlanArchivio.objects.get(id=pk).stato_pratica } new_values = { "stato_pratica": { "old": old_values["stato_pratica"], "new": 7 } } PlanArchivio.objects.filter(id=pk).update(stato_pratica=7) create_audit_log( user=request.user_id, activity_type='MODIFICA', entity_name='Pratica', entity_id=pk, old_values=old_values, new_values=new_values ) return Response({"status": True, "message": "Pratica chiusa"}) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_items_fatture(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) query_set = self._repository.entity.objects.filter( ~Q(stato_pratica__codice='NO_LAV') & ~Q(stato_pratica__codice='CHI') & ~Q(autorizzazione_fattura=4) ) filtered_set = parse_filters_and_sortings(args, query_set, user_id=None) items = PlanArchivioGetItemsSerializer(filtered_set, many=True) tot_spese = filtered_set.aggregate( tot_spese=Sum("spese"), tot_tariffa=Sum("tariffa") ) return Response({ "items": items.data, "stats": tot_spese }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def edit_fattura(self, request: Request, pk): try: pratica = PlanArchivio.objects.get(id=pk) PlanArchivio.objects.filter(id=pk).update(spese=request.data["spesa"], tariffa=request.data["tariffa"]) old_values = { "spese": float(pratica.spese) if pratica.spese is not None else None, "tariffa": float(pratica.tariffa) if pratica.tariffa is not None else None } new_values = { "spese": { "old": old_values["spese"], "new": float(request.data["spesa"]) if request.data["spesa"] is not None else None }, "tariffa": { "old": old_values["tariffa"], "new": float(request.data["tariffa"]) if request.data["tariffa"] is not None else None } } create_audit_log( user=request.user_id, activity_type='MODIFICA FATTURA', entity_name='Pratica', entity=pratica, entity_id=pk, old_values=old_values, new_values=new_values ) return Response({"status": True, "message": "Fattura modificata"}) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def edit_stato_fattura(self, request: Request, pk): try: pratica = PlanArchivio.objects.get(id=pk) PlanArchivio.objects.filter(id=pk).update(autorizzazione_fattura=request.data) old_values = { "autorizzazione_fattura": pratica.autorizzazione_fattura } def map_autorizzazione(value): if value == 3: return "Autorizzata" elif value == 2: return "Sospesa" elif value == 1: return "Non autorizzata" else: return "" new_values = { "autorizzazione_fattura": { "old": map_autorizzazione(old_values["autorizzazione_fattura"]), "new": map_autorizzazione(request.data) } } create_audit_log( user=request.user_id, activity_type='MODIFICA STATO FATTURA', entity_name='Pratica', entity=pratica, entity_id=pk, old_values=old_values, new_values=new_values ) return Response({"status": True, "message": "Fattura modificata"}) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def bulk_edit_stato_fattura(self, request: Request): try: record_ids = request.data.get('record_ids', []) status = request.data.get('status') if not record_ids: return Response( { "status": False, "message": "Nessun record selezionato" }, status=400 ) if status is None: return Response( { "status": False, "message": "Status mancante" }, status=400 ) pratiche = PlanArchivio.objects.filter(id__in=record_ids) if not pratiche.exists(): return Response( { "status": False, "message": "Pratiche non trovate" }, status=404 ) def map_autorizzazione(value): if value == 1: return "Sospesa" elif value == 2: return "Non autorizzata" elif value == 3: return "Autorizzata" elif value == 4: return "Fatturata" return "" updated_count = 0 cliente_ids = list( pratiche.values_list('cliente_id', flat=True).distinct() ) if len(cliente_ids) > 1: return Response( { "status": False, "message": "Le pratiche devono appartenere allo stesso cliente" }, status=400 ) for pratica in pratiche: old_status = pratica.autorizzazione_fattura pratica.autorizzazione_fattura = status pratica.save(update_fields=['autorizzazione_fattura']) old_values = { "autorizzazione_fattura": old_status } new_values = { "autorizzazione_fattura": { "old": map_autorizzazione(old_status), "new": map_autorizzazione(status) } } create_audit_log( user=request.user_id, activity_type='MODIFICA STATO FATTURA MASSIVA', entity_name='Pratica', entity=pratica, entity_id=pratica.id, old_values=old_values, new_values=new_values ) updated_count += 1 return Response({ "status": True, "message": f"{updated_count} fatture aggiornate con successo" }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def bulk_fatturazione(self, request: Request): try: record_ids = request.data.get('record_ids', []) numero_fattura = request.data.get('numero_fattura') data_fattura = request.data.get('data_fattura') if not record_ids: return Response( { "status": False, "message": "Nessun record selezionato" }, status=400 ) if not numero_fattura: return Response( { "status": False, "message": "Numero fattura mancante" }, status=400 ) if not data_fattura: return Response( { "status": False, "message": "Data fattura mancante" }, status=400 ) pratiche = PlanArchivio.objects.filter(id__in=record_ids) if not pratiche.exists(): return Response( { "status": False, "message": "Pratiche non trovate" }, status=404 ) cliente_ids = list( pratiche.values_list('cliente_id', flat=True).distinct() ) if len(cliente_ids) > 1: return Response( { "status": False, "message": "Le pratiche devono appartenere allo stesso cliente" }, status=400 ) updated_count = 0 for pratica in pratiche: old_values = { "autorizzazione_fattura": pratica.autorizzazione_fattura, "numero_fattura": pratica.numero_fattura, "data_fatturazione": str(pratica.data_fatturazione) if pratica.data_fatturazione else None, } stato_chiuso = StatoPratica.objects.get(codice='CHI') pratica.stato_pratica = stato_chiuso pratica.autorizzazione_fattura = 4 pratica.numero_fattura = numero_fattura pratica.data_fatturazione = data_fattura pratica.save( update_fields=[ 'stato_pratica', 'autorizzazione_fattura', 'numero_fattura', 'data_fatturazione', ] ) new_values = { "autorizzazione_fattura": { "old": old_values["autorizzazione_fattura"], "new": 4 }, "numero_fattura": { "old": old_values["numero_fattura"], "new": numero_fattura }, "data_fatturazione": { "old": old_values["data_fatturazione"], "new": data_fattura } } create_audit_log( user=request.user_id, activity_type='FATTURAZIONE MASSIVA', entity_name='Pratica', entity=pratica, entity_id=pratica.id, old_values=old_values, new_values=new_values ) updated_count += 1 return Response({ "status": True, "message": f"{updated_count} pratiche fatturate con successo" }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_items_planner(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) query_set = self._repository.entity.objects.filter( ~Q(stato_pratica__codice='NO_LAV') & ~Q(stato_pratica__codice='CHI') ) date_from_value = None date_to_value = None remaining_filters = [] original_filters = getattr(args, "filters", []) for f in original_filters: if f.field_name == "date_from": date_from_value = f.filter_value elif f.field_name == "date_to": date_to_value = f.filter_value else: remaining_filters.append(f) oggi = datetime.today() if date_from_value and date_to_value: if isinstance(date_from_value, str): d_start_obj = datetime.fromisoformat(date_from_value.split("T")[0]) else: d_start_obj = date_from_value if isinstance(date_to_value, str): d_end_obj = datetime.fromisoformat(date_to_value.split("T")[0]) else: d_end_obj = date_to_value d_end_obj = d_end_obj + relativedelta(months=1) cal = Italy() query_set = query_set.annotate( year_month=ExpressionWrapper( F("anno_di_lavorazione") * 100 + F("mese_di_lavorazione"), output_field=IntegerField() ) ) start_value = d_start_obj.year * 100 + d_start_obj.month end_value = d_end_obj.year * 100 + d_end_obj.month query_set = query_set.filter( year_month__gt=start_value, year_month__lte=end_value ) clean_args = FilterAndSorting() clean_args.filters = remaining_filters clean_args.sortings = getattr(args, "sortings", []) filtered_set = parse_filters_and_sortings(clean_args, query_set, user_id=None) filtered_set = filtered_set.annotate( cliente_title=F("cliente__title"), stato_pratica_title=F("stato_pratica__title"), incaricato_full_name=Concat( F("incaricato__nome"), Value(" "), F("incaricato__cognome"), output_field=CharField() ), tipo_documento = F("documenti__nome") ) oggi = timezone.now().date() stats = filtered_set.aggregate( totale_prese_in_carico=Count('id', filter=Q(preso_in_carico=1)), totale_aperte=Count('id', filter=Q(stato_pratica__codice='TO_DO')), totale_review=Count('id', filter=Q(stato_pratica__codice='REV')), completate=Count('id', filter=Q(stato_pratica__codice='CHI') | Q(stato_pratica__codice='NO_LAV')), totale_scadute=Count('id', filter=Q(data_di_scadenza__lt=oggi)), totale_scadenza_oggi=Count('id', filter=Q(data_di_scadenza=oggi)) ) lines = list(filtered_set.values()) return Response({ "totalCount": len(lines), "lines": lines, "stats": stats }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_planner_report_data(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) data = self._repository_ore_plan.filtered(args).values('settimana_lavorazione').annotate( ore_da_chiudere=Sum( Case( When(~Q(plan_item__stato_pratica__codice='CHI'), then='ore'), default=Value(0), output_field=DecimalField(max_digits=10, decimal_places=2) ) ), ore_chiuse=Sum( Case( When(Q(plan_item__stato_pratica__codice='CHI'), then='ore'), default=Value(0), output_field=DecimalField(max_digits=10, decimal_places=2) ) ), ore_rendicontate=Sum('ore') ).order_by('settimana_lavorazione') report_finale = {i: {'settimana_lavorazione': i, 'ore_da_chiudere': 0, 'ore_chiuse': 0, 'ore_rendicontate': 0} for i in range(1, 6)} anno_lav = None mese_lav = None original_filters = getattr(args, "filters", []) for f in original_filters: if f.field_name == "anno_lavorazione": anno_lav = f.filter_value elif f.field_name == "mese_lavorazione": mese_lav = f.filter_value # mese_lav = f.filter_value if isinstance(f.filter_value, list) else [f.filter_value] for riga in data: context = self.get_report_week_details(anno_lav, riga['settimana_lavorazione']) if context['anno'] == anno_lav and context['mese'] == mese_lav: sett_mese = context['settimana_mese'] if 1 <= sett_mese <= 5: target = report_finale[sett_mese] target['ore_da_chiudere'] += riga['ore_da_chiudere'] target['ore_chiuse'] += riga['ore_chiuse'] target['ore_rendicontate'] += riga['ore_rendicontate'] return Response(list(report_finale.values()), status=status.HTTP_200_OK) except Exception as e: print(e) return Response({'detail': 'An error occurred.'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_report_data_detail(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads(request.body, object_hook=objJsonDecode) data = self._repository_ore_plan.filtered(args).values('settimana_lavorazione').annotate( ore_da_chiudere=Sum( Case( When(~Q(plan_item__stato_pratica__codice='CHI'), then='ore'), default=Value(0), output_field=DecimalField(max_digits=10, decimal_places=2) ) ), ore_chiuse=Sum( Case( When(Q(plan_item__stato_pratica__codice='CHI'), then='ore'), default=Value(0), output_field=DecimalField(max_digits=10, decimal_places=2) ) ), ore_rendicontate=Sum('ore') ).order_by('settimana_lavorazione') anno_lav = None mesi_lav = [] original_filters = getattr(args, "filters", []) for f in original_filters: if f.field_name == "anno_lavorazione": anno_lav = f.filter_value elif f.field_name == "mese_lavorazione": mesi_lav = f.filter_value if isinstance(f.filter_value, list) else [f.filter_value] report_per_mese = {m: {i: {'settimana_lavorazione': i, 'ore_da_chiudere': 0, 'ore_chiuse': 0, 'ore_rendicontate': 0} for i in range(1, 6)} for m in mesi_lav} for riga in data: context = self.get_report_week_details(anno_lav, riga['settimana_lavorazione']) if context['anno'] == anno_lav and context['mese'] in mesi_lav: sett_mese = context['settimana_mese'] if 1 <= sett_mese <= 5: target = report_per_mese[context['mese']][sett_mese] target['ore_da_chiudere'] += (riga['ore_da_chiudere'] or 0) target['ore_chiuse'] += (riga['ore_chiuse'] or 0) target['ore_rendicontate'] += (riga['ore_rendicontate'] or 0) return Response(report_per_mese, status=status.HTTP_200_OK) except Exception as e: print(e) return Response({'detail': 'An error occurred.'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) def get_report_week_details(self, anno, sett_annuale): data_lunedi = datetime.strptime(f'{anno}-W{sett_annuale:02d}-1', "%Y-W%W-%w") anno_riferimento, mese_riferimento, week_num = self.get_fisc_context(data_lunedi) return { 'anno': anno_riferimento, 'mese': mese_riferimento, 'settimana_mese': week_num } @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_report_page(self, request: Request) -> Response: try: args: FilterAndSorting = json.loads( request.body, object_hook=objJsonDecode ) queryset = self._repository.filtered(args) oggi = timezone.now().date() raw_kpi = queryset.aggregate( totale_pratiche=Count("id"), completate=Count( "id", filter=Q(stato_pratica__codice__in=["CHI", "NO_LAV"]) ), prese_in_carico=Count( "id", filter=Q(preso_in_carico=True) ), totale_tariffe=Coalesce( Sum("tariffa"), Value(Decimal("0.00")) ), totale_spese=Coalesce( Sum("spese"), Value(Decimal("0.00")) ), fatturate=Count( "id", filter=Q(Q(autorizzazione_fattura=4) & Q(numero_fattura__isnull=False) & Q(data_fatturazione__isnull=False)) ), ) stato_pratiche = list( queryset .values( "stato_pratica__codice", "stato_pratica__title", "stato_pratica__color" ) .annotate( totale=Count("id") ) .order_by("stato_pratica__title") ) stato_pratiche = [ { "stato": x["stato_pratica__title"], "totale": x["totale"], "color": x["stato_pratica__color"] } for x in stato_pratiche ] kpi = [ { "key": "totale_pratiche", "label": "Totale pratiche", "value": raw_kpi["totale_pratiche"], "icon": "folder", "color": "#44bcc2" }, { "key": "completate", "label": "Completate", "value": raw_kpi["completate"], "icon": "task_alt", "color": "#1a659c" }, { "key": "fatturate", "label": "Fatturate", "value": raw_kpi["fatturate"], "icon": "receipt", "color": "#cc6227" }, { "key": "prese_in_carico", "label": "Prese in carico", "value": raw_kpi["prese_in_carico"], "icon": "assignment_ind", "color": "#f9a825" }, ] economico_mensile = list( queryset .exclude(data_fatturazione__isnull=True) .annotate( mese=TruncMonth("data_fatturazione") ) .values("mese") .annotate( tariffa=Coalesce( Sum("tariffa"), Value(Decimal("0.00")) ), spese=Coalesce( Sum("spese"), Value(Decimal("0.00")) ) ) .order_by("mese") ) prossime_attivita = list( queryset .filter( Q(data_di_scadenza__gte=oggi) & ~Q(stato_pratica__codice="CHI") ) .values( "id", "title", "cliente__title", "data_di_scadenza", "stato_pratica__title" ) .order_by("data_di_scadenza")[:10] ) for item in prossime_attivita: scadenza = item["data_di_scadenza"] item["giorni_alla_scadenza"] = ( scadenza.date() - oggi ).days agg = queryset.aggregate( effettive=Coalesce( Sum("ore_effettive"), Value(Decimal("0.00")), output_field=DecimalField(max_digits=12, decimal_places=2) ), standard=Coalesce( Sum("ore_standard"), Value(Decimal("0.00")), output_field=DecimalField(max_digits=12, decimal_places=2) ) ) efficienza = ( round((agg["effettive"] / agg["standard"]) * 100, 2) if agg["standard"] and agg["effettive"] > 0 else 0 ) return Response({ "kpi": kpi, "stato_pratiche": stato_pratiche, "prossime_attivita": prossime_attivita, "economico_mensile": economico_mensile, "efficienza": efficienza }) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def edit_planner(self, request: Request): try: dto = request.data.dict() if hasattr(request.data, 'dict') else dict(request.data) pratiche = list( PlanArchivio.objects.filter(id__in=dto['ids']) ) PlanArchivio.objects.filter(id__in=dto['ids']).update( anno_di_lavorazione=int(dto['anno_di_lavorazione']), mese_di_lavorazione=dto["mese_di_lavorazione"], settimana_di_lavorazione=dto["settimana_di_lavorazione"], incaricato=dto["incaricato"], documenti=dto["documenti"], stato_pratica=dto["stato"] ) for pratica in pratiche: old_values = {} new_values = {} fields = { "anno_di_lavorazione": int(dto['anno_di_lavorazione']), "mese_di_lavorazione": dto["mese_di_lavorazione"], "settimana_di_lavorazione": dto["settimana_di_lavorazione"], "incaricato": dto["incaricato"], "documenti": dto["documenti"], "stato_pratica": dto["stato"] } for field, new_value in fields.items(): old_value = getattr(pratica, field) if hasattr(old_value, "pk"): old_value = old_value.pk if old_value != new_value: old_values[field] = old_value new_values[field] = { "old": old_value, "new": new_value } if new_values: create_audit_log( user=request.user_id, activity_type='MODIFICA AGENDA ORE', entity_name='Pratica', entity=pratica, entity_id=pratica.id, old_values=old_values, new_values=new_values ) return Response({"status": True, "message": "Pratica modificata"}) except Exception as e: handle_exception(e) @requires_auth(permission=[os.getenv('GESTORE'), os.getenv('INCARICATO')]) def get_tipo_documenti(self, request: Request): try: documenti = TipoDocumento.objects.all() serializer = TipoDocumentoSerializer(documenti, many=True) data = serializer.data return Response(data, status=status.HTTP_200_OK) except Exception as e: print(e) return Response({'detail': 'An error occurred.'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)