Finished release 0.3.

Changelog.md

 # Changelog
 
+## [0.2] - 2025-01-27
+- Added new validations
+
 ## [0.2] - 2025-01-09
 - Updated the setting related to the MEDIA ROOT and URL and also the CSRF settings
 

requirements.txt

-Django==5.1.1
+Django==5.0.11
 ruff
 mypy
 tox
 sphinx
 sphinx-autodoc-typehints
+mssql-django

sage_validation/file_validator/apps.py

 """App configuration for file_validator app."""
+
 from django.apps import AppConfig
 
 

sage_validation/file_validator/forms.py

 """Forms for the file_validator app."""
+
 import csv
-from collections.abc import Sequence
+from collections.abc import Iterable, Sequence
 from typing import ClassVar
 
 from django import forms
+from django.core.files.uploadedfile import UploadedFile
+
+from sage_validation.file_validator.models import Meocostcentres, Meovalidsageaccounts, PlAccountCodes, XxData
 
 
 class CSVUploadForm(forms.Form):
@@ -12,68 +16,74 @@ class CSVUploadForm(forms.Form):
     file = forms.FileField(label="Select a CSV file")
 
     required_columns: ClassVar[list] = [
-        "AccountNumber", "CBAccountNumber", "DaysDiscountValid", "DiscountValue",
-        "DiscountPercentage", "DueDate", "GoodsValueInAccountCurrency",
-        "PurControlValueInBaseCurrency", "DocumentToBaseCurrencyRate",
-        "DocumentToAccountCurrencyRate", "PostedDate", "QueryCode",
-        "TransactionReference", "SecondReference", "Source",
-        "SYSTraderTranType", "TransactionDate", "UniqueReferenceNumber",
-        "UserNumber", "TaxValue", "SYSTraderGenerationReasonType",
-        "GoodsValueInBaseCurrency"
+        "AccountNumber",
+        "CBAccountNumber",
+        "DaysDiscountValid",
+        "DiscountValue",
+        "DiscountPercentage",
+        "DueDate",
+        "GoodsValueInAccountCurrency",
+        "PurControlValueInBaseCurrency",
+        "DocumentToBaseCurrencyRate",
+        "DocumentToAccountCurrencyRate",
+        "PostedDate",
+        "QueryCode",
+        "TransactionReference",
+        "SecondReference",
+        "Source",
+        "SYSTraderTranType",
+        "TransactionDate",
+        "UniqueReferenceNumber",
+        "UserNumber",
+        "TaxValue",
+        "SYSTraderGenerationReasonType",
+        "GoodsValueInBaseCurrency",
     ]
 
     repeating_columns: ClassVar[dict] = {
         "NominalAnalysis": [
-            "NominalAnalysisTransactionValue", "NominalAnalysisNominalAccountNumber",
-            "NominalAnalysisNominalCostCentre", "NominalAnalysisNominalDepartment",
-            "NominalAnalysisNominalAnalysisNarrative", "NominalAnalysisTransactionAnalysisCode"
+            "NominalAnalysisTransactionValue",
+            "NominalAnalysisNominalAccountNumber",
+            "NominalAnalysisNominalCostCentre",
+            "NominalAnalysisNominalDepartment",
+            "NominalAnalysisNominalAnalysisNarrative",
+            "NominalAnalysisTransactionAnalysisCode",
         ],
         "TaxAnalysis": [
-            "TaxAnalysisTaxRate", "TaxAnalysisGoodsValueBeforeDiscount",
-            "TaxAnalysisDiscountValue", "TaxAnalysisDiscountPercentage",
-            "TaxAnalysisTaxOnGoodsValue"
-        ]
+            "TaxAnalysisTaxRate",
+            "TaxAnalysisGoodsValueBeforeDiscount",
+            "TaxAnalysisDiscountValue",
+            "TaxAnalysisDiscountPercentage",
+            "TaxAnalysisTaxOnGoodsValue",
+        ],
     }
 
-    def clean_file(self) -> str:
-        """Validate the uploaded file format and contents."""
+    def clean_file(self) -> UploadedFile:
+        """Validate the uploaded file."""
         file = self.cleaned_data["file"]
 
-        if not file.name.endswith(".csv"):
-            err_msg = "File must be in CSV format"
-            raise forms.ValidationError(err_msg)
-
-        try:
-            csv_file = file.read().decode("utf-8").splitlines()
-            reader = csv.DictReader(csv_file)
-            fieldnames = reader.fieldnames if reader.fieldnames is not None else []
-
-            missing_columns = [col for col in self.required_columns if col not in fieldnames]
+        # Step 1: Validate file type
+        self._validate_file_type(file)
 
-            for section_name, column_list in self.repeating_columns.items():
-                max_repeat = self.get_max_repeat(fieldnames, section_name)
+        # Step 2: Parse file and validate headers
+        csv_file = file.read().decode("utf-8").splitlines()
+        reader = csv.DictReader(csv_file, delimiter=",")
+        fieldnames = reader.fieldnames if reader.fieldnames is not None else []
+        self._validate_headers(fieldnames)
 
-                if max_repeat == 0:
-                    missing_columns.extend([f"{base_col}/1" for base_col in column_list])
-                else:
-                    for repeat in range(1, max_repeat + 1):
-                        missing_columns.extend(
-                            [f"{base_col}/{repeat}" for base_col in column_list if
-                             f"{base_col}/{repeat}" not in fieldnames]
-                        )
-
-            if missing_columns:
-                err_msg = f"Missing required columns: {', '.join(missing_columns)}"
-                raise forms.ValidationError(err_msg)
-
-        except (UnicodeDecodeError, csv.Error) as e:
-            err_msg = f"File could not be processed: {e!s}"
-            raise forms.ValidationError(err_msg) from e
+        error_list = []
+        # Step 3: Validate 'Source' and 'SYSTraderTranType' values
+        data = list(reader)
+        error_list.extend(self._validate_source_and_trader_type(data))
+        error_list.extend(self._validate_nominal_analysis_account(data))
+        error_list.extend(self._validate_nc_cc_dep_combination_against_meo_sage_account(data))
+        if error_list:
+            raise forms.ValidationError(error_list)
 
         return file
 
     @staticmethod
-    def get_max_repeat(fieldnames: Sequence[str], section_prefix: str) -> int:
+    def _get_max_repeat(fieldnames: Sequence[str], section_prefix: str) -> int:
         """Identify the maximum number of repeats for a section."""
         max_repeat = 0
         for field in fieldnames:
@@ -84,3 +94,135 @@ class CSVUploadForm(forms.Form):
                 except ValueError:
                     continue
         return max_repeat
+
+    @staticmethod
+    def _validate_file_type(file: UploadedFile) -> None:
+        """Validate that the uploaded file is a CSV."""
+        if not file.name.endswith(".csv"):
+            msg = "File must be in CSV format."
+            raise forms.ValidationError(msg)
+
+    def _validate_headers(self, fieldnames: Sequence[str]) -> None:
+        """Validate required and repeating columns in the headers."""
+        missing_columns = [col for col in self.required_columns if col not in fieldnames]
+
+        for section_name, column_list in self.repeating_columns.items():
+            max_repeat = self._get_max_repeat(fieldnames, section_name)
+            if max_repeat == 0:
+                missing_columns.extend([f"{base_col}/1" for base_col in column_list])
+            else:
+                for repeat in range(1, max_repeat + 1):
+                    missing_columns.extend([
+                        f"{base_col}/{repeat}" for base_col in column_list if f"{base_col}/{repeat}" not in fieldnames
+                    ])
+
+        if missing_columns:
+            msg = f"Missing required columns: {', '.join(missing_columns)}"
+            raise forms.ValidationError(msg)
+
+    @staticmethod
+    def _validate_source_and_trader_type(data: Iterable[dict]) -> list:
+        """Validate that 'Source' is always 80 and 'SYSTraderTranType' is always 4."""
+        errors = []
+
+        for index, row in enumerate(data, start=1):
+            if row.get("Source") != "80":
+                errors.append(f"Row {index}: 'Source' must be 80, but found {row.get('Source')}.")
+
+            if row.get("SYSTraderTranType") != "4":
+                errors.append(f"Row {index}: 'SYSTraderTranType' must be 4, but found {row.get('SYSTraderTranType')}.")
+
+        return errors
+
+    @staticmethod
+    def _validate_nominal_analysis_account(data: Iterable[dict]) -> list[str]:
+        """Validate that 'AccountNumber' matches the name in 'NominalAnalysisNominalAnalysisNarrative/1'.
+
+        This only checks the first group of NominalAnalysis columns. A list of codes/names
+        is fetched from the database for validation (from the 'PL Account Codes' table).
+
+        Args:
+            data (Iterable[dict]): The rows of data to validate.
+
+        Returns:
+            List[str]: A list of error messages, if any.
+
+        """
+        errors = []
+
+        account_code_map = {
+            obj.pl_account_code: obj.pl_account_name
+            for obj in PlAccountCodes.objects.using("meo").all()  # type: ignore[attr-defined]
+        }
+
+        for index, row in enumerate(data, start=1):
+            account_code = row.get("AccountNumber")
+            nominal = row.get("NominalAnalysisNominalAnalysisNarrative/1")
+
+            # Skip rows without 'AccountNumber' or 'NominalAnalysisNominalAnalysisNarrative/1'
+            if not account_code or not nominal:
+                continue
+
+            pl_account_name = account_code_map.get(account_code)
+            if pl_account_name is None:
+                errors.append(f"Row {index}: 'AccountNumber' {account_code} does not exist in PL Account Codes.")
+            elif pl_account_name not in nominal:
+                errors.append(
+                    f"Row {index}: 'AccountNumber' must match '{pl_account_name}' in "
+                    f"'NominalAnalysisNominalAnalysisNarrative/1', but found '{nominal}'."
+                )
+
+        return errors
+
+    @staticmethod
+    def _validate_nc_cc_dep_combination_against_meo_sage_account(data: Iterable[dict]) -> list[str]:
+        """Validate that the combination of 'AccountCostCentre', 'AccountDepartment', and 'AccountNumber' exists in MEO.
+
+        Args:
+            data (Iterable[dict]): The rows of data to validate.
+
+        Returns:
+            List[str]: A list of error messages, if any.
+
+        """
+        errors = []
+
+        cost_centre_map = {
+            obj.cc: obj.cctype for obj in Meocostcentres.objects.using("meo").all()
+        }
+
+        xx_data_map = {
+            obj.xx_value: (obj.project, obj.overhead) for obj in XxData.objects.using("meo").all()
+        }
+
+        for index, row in enumerate(data, start=1):
+            cc = row.get("NominalAnalysisNominalCostCentre/1")
+            dep = row.get("NominalAnalysisNominalDepartment/1")
+            nominal_account_name = row.get("NominalAnalysisNominalAccountNumber/1")
+
+            if not cc or not dep or not nominal_account_name:
+                continue
+
+            cc_type = cost_centre_map.get(cc)
+            if not cc_type:
+                errors.append(f"Row {index}: 'NominalAnalysisNominalCostCentre/1' ({cc}) is not a valid cost centre.")
+                continue
+
+            xx_data = xx_data_map.get(nominal_account_name)
+            if not xx_data:
+                errors.append(
+                    f"Row {index}: 'NominalAnalysisNominalAccountNumber/1' ({nominal_account_name}) is not valid.")
+                continue
+
+            nc = xx_data[0] if cc_type == "Project" else xx_data[1]
+
+            if not Meovalidsageaccounts.objects.using("meo").filter(
+                    accountcostcentre=cc, accountdepartment=dep, accountnumber=nc
+            ).exists():
+                errors.append(
+                    f"Row {index}: The combination of 'NominalAnalysisNominalCostCentre/1' ({cc}), "
+                    f"'NominalAnalysisNominalDepartment/1' ({dep}), and 'NominalAnalysisNominalAccountNumber/1' "
+                    f"({nominal_account_name}) does not exist in MEO valid Sage accounts."
+                )
+
+        return errors

sage_validation/file_validator/models.py

-"""Models for the file_validator app."""
+"""Models for MEO DB tables and views."""
+from django.db import models
+
+
+class PlAccountCodes(models.Model):
+    """Represents profit and loss (PL) account codes.
+
+    This model maps the PL account names to their respective codes.
+    """
+
+    objects = None
+    pl_account_name = models.TextField(db_column="PL_Account_Name")
+    pl_account_code = models.CharField(db_column="PL_Account_Code", primary_key=True, max_length=50)
+
+    class Meta:
+        """Metaclass for the PlAccountCodes model."""
+
+        managed = False
+        db_table = "PL_Account_Codes"
+
+
+class Meocostcentres(models.Model):
+    """Represents cost centres.
+
+    This model contains data related to cost centres,
+    including their names, types, and unique IDs.
+    """
+
+    cc = models.CharField(db_column="CC", max_length=3)
+    cc_name = models.CharField(db_column="CC_Name", max_length=50, blank=True, null=True)
+    cctype = models.CharField(db_column="CCType", max_length=8)
+    id = models.IntegerField(db_column="ID", primary_key=True)
+
+    class Meta:
+        """Metaclass for the Meocostcentres model."""
+
+        managed = False
+        db_table = "meoCostCentres"
+
+
+class Meonominal(models.Model):
+    """View for MEO nominal codes."""
+
+    nom = models.CharField(db_column="Nom", max_length=5)
+    nomname = models.CharField(db_column="NomName", max_length=60, blank=True, null=True)
+    nomid = models.IntegerField(db_column="NomID")
+
+    class Meta:
+        """Metaclass for the Meonominal model."""
+
+        managed = False
+        db_table = "meoNominal"
+
+
+class Meovalidsageaccounts(models.Model):
+    """View for MEO valid Sage accounts."""
+
+    accountname = models.CharField(db_column="AccountName", max_length=60)
+    accountnumber = models.CharField(db_column="AccountNumber", max_length=8, blank=True, null=True)
+    accountcostcentre = models.CharField(db_column="AccountCostCentre", max_length=3, blank=True, null=True)
+    accountdepartment = models.CharField(db_column="AccountDepartment", max_length=3, blank=True, null=True)
+
+    class Meta:
+        """Metaclass for the Meovalidsageaccounts model."""
+
+        managed = False
+        db_table = "meoValidSageAccounts"
+
+
+class Meovalidsuppliers(models.Model):
+    """View for MEO valid suppliers."""
+
+    supplieraccountnumber = models.CharField(db_column="SupplierAccountNumber", max_length=8)
+    supplieraccountname = models.CharField(db_column="SupplierAccountName", max_length=60)
+
+    class Meta:
+        """Metaclass for the Meovalidsuppliers model."""
+
+        managed = False
+        db_table = "meoValidSuppliers"
+
+
+class Meovalidvat(models.Model):
+    """View for MEO valid VAT codes."""
+
+    tax_code = models.SmallIntegerField(db_column="Tax code")
+    tax_code_label = models.CharField(db_column="tax code label", max_length=60)
+    rate = models.DecimalField(db_column="Rate", max_digits=18, decimal_places=2)
+    inputnominalaccount = models.CharField(db_column="InputNominalAccount", max_length=8, blank=True, null=True)
+
+    class Meta:
+        """Metaclass for the Meovalidvat model."""
+
+        managed = False
+        db_table = "meoValidVAT"
+
+
+class VwPlAcctCodes(models.Model):
+    """View for profit and loss account codes."""
+
+    pl_account_name = models.TextField(db_column="PL_Account_Name")
+    pl_account_code = models.CharField(db_column="PL_Account_Code", max_length=50)
+
+    class Meta:
+        """Metaclass for the VwPlAcctCodes model."""
+
+        managed = False
+        db_table = "vw_PL_Acct_Codes"
+
+
+class VwXxData(models.Model):
+    """provides a read-only view of XX data entries.
+
+    The entries are typically used for categorizing costs and activities.
+    """
+
+    description = models.CharField(db_column="Description", max_length=50)
+    xx_value = models.CharField(db_column="xx_Value", max_length=50)
+    project = models.SmallIntegerField(db_column="Project")
+    overhead = models.IntegerField(db_column="Overhead")
+
+    class Meta:
+        """Meta class for the VwXxData model."""
+
+        managed = False
+        db_table = "vw_xx-data"
+
+
+class XxData(models.Model):
+    """Model for XX data entries."""
+
+    description = models.CharField(db_column="Description", max_length=50)
+    xx_value = models.CharField(db_column="xx_Value", primary_key=True, max_length=50)
+    project = models.SmallIntegerField(db_column="Project")
+    overhead = models.IntegerField(db_column="Overhead")
+
+    class Meta:
+        """Metaclass for the XxData model."""
+
+        managed = False
+        db_table = "xx-data"

sage_validation/file_validator/urls.py

 """Urls for the file_validator app."""
+
 from django.urls import path
 
 from sage_validation.file_validator.views import CSVUploadView

sage_validation/file_validator/views.py

 """Views for the file_validator app."""
+
 from django.http import HttpRequest, HttpResponse, JsonResponse
 from django.shortcuts import render
 from django.urls import reverse_lazy

sage_validation/settings.py

@@ -78,7 +78,19 @@ DATABASES = {
     "default": {
         "ENGINE": "django.db.backends.sqlite3",
         "NAME": BASE_DIR / "db.sqlite3",
-    }
+    },
+    "meo": {
+        'ENGINE': 'mssql',
+        'NAME': os.getenv("MSSQL_DB_NAME", ""),
+        'USER': os.getenv("MSSQL_DB_USER", ""),
+        'PASSWORD': os.getenv("MSSQL_DB_PASSWORD", ""),
+        'HOST': os.getenv("MSSQL_DB_HOST", "localhost"),
+        'PORT': os.getenv("MSSQL_DB_PORT", ""),
+
+        'OPTIONS': {
+            'driver': 'ODBC Driver 18 for SQL Server',
+        },
+    },
 }
 
 # Password validation

setup.py

@@ -3,11 +3,12 @@ from setuptools import find_packages, setup
 
 setup(
     name="sage-validation",
-    version="0.2",
+    version="0.3",
     packages=find_packages(),
     include_package_data=True,
     install_requires=[
-        "Django==5.1.1",
+        "Django==5.0.11",
+        "mssql-django",
     ],
     extras_require={
         "prod": [