financial documents validators

validators_financial_documents.py

from calendar import monthrange
from copy import copy
from datetime import date, timedelta
from typing import Callable

from django.db.models import F, QuerySet, Subquery
from rest_framework.serializers import Serializer
from rest_framework.validators import ValidationError

from a_project.utils.core import date_to_timestamp
from epp.constants.financial_document import (
    DISCONTINUITY_ERROR,
    DUPLICATE_PERIOD_ERROR,
    RAISE_ERROR_BANK_STATEMENT_REQUIRES_ACCOUNT,
    RAISE_ERROR_BANK_STATEMENT_REQUIRES_MONTH_AND_YEAR,
    RAISE_ERROR_FINANCIAL_DOCUMENT_EXISTS,
    RAISE_ERROR_FINANCIAL_DOCUMENT_WITH_BANK_ACCOUNT_EXISTS,
    RECENT_DOCUMENT_ERROR,
)
from epp.models import DocumentType, FinancialDocument
from epp.models.financial_documents import FinancialDocumentsCoreABC, PeriodType
from epp.types.financial_document import FinancialDocumentData

...  # other developer code removed

class PeriodTypeDateFrameTogetherValidator:
    requires_context = True
    missing_message = "This field is required."
    START = "start"
    END = "end"
    frame_names = [START, END]

    MONTH = "month"
    DAY = "day"
    period_type_required_fields = {
        PeriodType.YEAR_TO_DATE.value: {},
        PeriodType.AS_OF_DATE.value: {},
        PeriodType.FULL_YEAR.value: {
            START: (MONTH, DAY),
            END: (MONTH, DAY),
        },
        PeriodType.MONTHLY.value: {
            START: (DAY,),
            END: (DAY,),
        },
        PeriodType.QUARTERLY.value: {
            START: (MONTH, DAY),
            END: (MONTH, DAY),
        },
        PeriodType.UNDEFINED.value: {},
    }
    QUARTERS_START_MONTHS_NUMS = [1, 4, 7, 10]
    YEAR_START_MONTH_NUM = 1
    MONTH_START_DAY_NUM = 1

    QUARTERS_END_MONTHS_NUMS = [3, 6, 9, 12]
    YEAR_END_MONTH_NUM = 12

    def __init__(self, period_type_field: str, date_frame_mapping_fields: dict[str, str], message: str | None = None):
        self.period_type_field = period_type_field
        self._validate_period_mapping(date_frame_mapping_fields)
        self.date_frame_mapping_fields = date_frame_mapping_fields

    def _validate_period_mapping(self, date_frame_mapping_fields: dict[str, str]) -> None:
        if not isinstance(date_frame_mapping_fields, dict):
            raise TypeError("periods_mapping_fields must be a dictionary")
        if other_fields := set(date_frame_mapping_fields.keys()) - set(self.frame_names):
            raise ValueError(f"Invalid fields in periods_mapping_fields: {other_fields}")

    def _get_acceptable_values_for_start(self, period_type: str, date_value: date = None) -> list:
        if period_type == f"{PeriodType.QUARTERLY.value}_{self.MONTH}":
            return self.QUARTERS_START_MONTHS_NUMS
        if period_type == f"{PeriodType.FULL_YEAR.value}_{self.MONTH}":
            return [self.YEAR_START_MONTH_NUM]
        if "day" in period_type:
            return [self.MONTH_START_DAY_NUM]
        return []

    def _get_acceptable_values_for_end(self, period_type: str, date_value: date) -> list:
        if period_type == f"{PeriodType.QUARTERLY.value}_{self.MONTH}":
            return self.QUARTERS_END_MONTHS_NUMS
        if period_type == f"{PeriodType.FULL_YEAR.value}_{self.MONTH}":
            return [self.YEAR_END_MONTH_NUM]
        if "day" in period_type:
            return [monthrange(date_value.year, date_value.month)[1]]
        return []

    def _validate_date_by_date_type(
        self, attrs_to_check: list, period_type: str, get_acceptables_func: Callable, date_value: date
    ) -> None:
        for attr in attrs_to_check:
            key_period_type = f"{period_type}_{attr}"
            acceptable_values = get_acceptables_func(key_period_type, date_value)
            if getattr(date_value, attr) not in acceptable_values:
                raise ValidationError(f"Start date and end date must be in consistence with {period_type}")

    def _validate_dates_order(self, start_date: date, end_date: date) -> None:
        if start_date > end_date:
            raise ValidationError(
                {
                    self.date_frame_mapping_fields[self.START]: "Start date must be less than or equal to end date.",
                    self.date_frame_mapping_fields[self.END]: "End date must be greater than or equal to start date.",
                },
                code="invalid",
            )

    def _check_required_fields(self, attrs: dict, serializer: Serializer):
        start_field_name = self.date_frame_mapping_fields.get(self.START)
        date_value_start = attrs.get(start_field_name) or getattr(serializer.instance, start_field_name, None)

        end_field_name = self.date_frame_mapping_fields.get(self.END)
        date_value_end = attrs.get(end_field_name) or getattr(serializer.instance, end_field_name, None)

        if date_value_start is None and date_value_end is None:
            return

        if any(date_value is None for date_value in [date_value_start, date_value_end]):
            raise ValidationError(
                {
                    self.date_frame_mapping_fields[self.START]: self.missing_message,
                    self.date_frame_mapping_fields[self.END]: self.missing_message,
                },
                code="required",
            )

        self._validate_dates_order(date_value_start, date_value_end)

        period_type = attrs.get(self.period_type_field) or serializer.instance.period_type
        required_periods_dict = self.period_type_required_fields.get(period_type)

        if not required_periods_dict:
            return

        periods_to_check_start = required_periods_dict.get(self.START)
        periods_to_check_end = required_periods_dict.get(self.END)

        self._validate_date_by_date_type(
            attrs_to_check=periods_to_check_start,
            period_type=period_type,
            get_acceptables_func=self._get_acceptable_values_for_start,
            date_value=date_value_start,
        )

        self._validate_date_by_date_type(
            attrs_to_check=periods_to_check_end,
            period_type=period_type,
            get_acceptables_func=self._get_acceptable_values_for_end,
            date_value=date_value_end,
        )

    def _is_period_type_valid(self, attrs: dict, serializer: Serializer) -> None:
        if (
            attrs.get(self.period_type_field) is None
            and getattr(serializer.instance, self.period_type_field, None) is None
        ):
            raise ValidationError(
                {self.period_type_field: self.missing_message},
                code="required",
            )

    def __call__(self, attrs: dict, serializer: Serializer) -> None:
        self._is_period_type_valid(attrs, serializer)
        self._check_required_fields(attrs, serializer)


def validate_documents_set_on_gaps(
    documents: QuerySet[FinancialDocumentsCoreABC], data_to_validate: dict
) -> tuple[bool, dict]:
    subcontractor = data_to_validate.get("subcontractor")

    dates_to_validate: list[dict] = data_to_validate.get("dates_to_validate")
    dates_to_validate.sort(
        key=lambda elem: (date_to_timestamp(elem["end_date"]), date_to_timestamp(elem["start_date"]))
    )

    closest_document_subquery = (
        documents.filter(subcontractor=subcontractor, end_date__lte=dates_to_validate[0]["start_date"])
        .order_by("-end_date")
        .values("end_date")[:1]
    )
    existing_documents = (
        documents.filter(subcontractor=subcontractor)
        .annotate(closest_end_date=Subquery(closest_document_subquery))
        .annotate(uuid=F("file__uuid"))
    )

    all_dates = copy(dates_to_validate)  # intentionally used copy() to keep pointers to dicts in dates_to_validate
    existing_documents_data = existing_documents.filter(end_date__gte=F("closest_end_date")).values(
        "uuid", "period_type", "start_date", "end_date"
    )
    all_dates += existing_documents_data

    all_dates.sort(
        key=lambda elem: (-date_to_timestamp(elem["end_date"]), date_to_timestamp(elem["start_date"]))
    )  # from most recent to older
    is_valid = True

    prev_doc = {}
    for cursor_ind, document in enumerate(all_dates):
        document["errors"] = []
        if cursor_ind == 0:
            if date.today() - document["end_date"] > timedelta(days=30):
                is_valid = False
                document["errors"].append(RECENT_DOCUMENT_ERROR)
            prev_doc = document
            continue
        if prev_doc["start_date"] - document["end_date"] > timedelta(days=1):
            is_valid = False
            document["errors"].append(
                (DISCONTINUITY_ERROR[0], DISCONTINUITY_ERROR[1].format(to_date=prev_doc["start_date"]))
            )
            prev_doc["errors"].append(
                (DISCONTINUITY_ERROR[0], DISCONTINUITY_ERROR[1].format(to_date=document["end_date"]))
            )

        if (
            document["period_type"] == prev_doc["period_type"]
            and document["start_date"] == prev_doc["start_date"]
            and document["end_date"] == prev_doc["end_date"]
        ):
            is_valid = False
            document["errors"].append(DUPLICATE_PERIOD_ERROR)
            prev_doc["errors"].append(DUPLICATE_PERIOD_ERROR)
        prev_doc = document

    return is_valid, data_to_validate