Skip to content
Snippets Groups Projects
Commit ff892831 authored by Neda Moeini's avatar Neda Moeini
Browse files

Add column validation to the file validator and make linter happy.

parent 23c60845
No related branches found
No related tags found
No related merge requests found
venv venv
.idea .idea
*.pyc *.pyc
\ No newline at end of file /staticfiles/
.env
\ No newline at end of file
"""Form for uploading CSV files.""" """Forms for the file_validator app."""
import csv
from collections.abc import Sequence
from typing import ClassVar
from django import forms from django import forms
class CSVUploadForm(forms.Form): class CSVUploadForm(forms.Form):
"""Form for uploading CSV files.""" """Form for uploading CSV files only."""
file = forms.FileField(label="Select a CSV file") 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"
]
repeating_columns: ClassVar[dict] = {
"NominalAnalysis": [
"NominalAnalysisTransactionValue", "NominalAnalysisNominalAccountNumber",
"NominalAnalysisNominalCostCentre", "NominalAnalysisNominalDepartment",
"NominalAnalysisNominalAnalysisNarrative", "NominalAnalysisTransactionAnalysisCode"
],
"TaxAnalysis": [
"TaxAnalysisTaxRate", "TaxAnalysisGoodsValueBeforeDiscount",
"TaxAnalysisDiscountValue", "TaxAnalysisDiscountPercentage",
"TaxAnalysisTaxOnGoodsValue"
]
}
def clean_file(self) -> str: def clean_file(self) -> str:
"""Check if the file is a CSV file.""" """Validate the uploaded file format and contents."""
file = self.cleaned_data["file"] file = self.cleaned_data["file"]
# Check if the file is a CSV
if not file.name.endswith(".csv"): if not file.name.endswith(".csv"):
err_msg = "Only CSV files are allowed" err_msg = "File must be in CSV format"
raise forms.ValidationError(err_msg) 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]
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:
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
return file return file
@staticmethod
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:
if field.startswith(section_prefix):
try:
repeat_number = int(field.split("/")[-1])
max_repeat = max(max_repeat, repeat_number)
except ValueError:
continue
return max_repeat
{% extends 'base.html' %} {% extends 'base.html' %}
{% block title %}CSV Upload{% endblock %} {% block title %}File Upload{% endblock %}
{% block content %} {% block content %}
<div class="bg-white p-10 rounded-lg shadow-lg w-11/12 md:w-1/2 lg:w-1/3 mx-auto"> <div class="bg-white p-10 rounded-lg shadow-lg w-11/12 md:w-1/2 lg:w-1/3 mx-auto">
<h2 class="text-2xl font-bold mb-6 text-gray-800 text-center">Upload CSV</h2> <h2 class="text-2xl font-bold mb-6 text-gray-800 text-center">Upload CSV File</h2>
<form id="uploadForm" enctype="multipart/form-data" class="space-y-6"> <form id="uploadForm" enctype="multipart/form-data" class="space-y-6">
<div> <div>
<label for="fileInput" class="block text-sm font-medium text-gray-700 mb-2">Choose CSV File</label> <label for="fileInput" class="block text-sm font-medium text-gray-700 mb-2">Choose CSV File</label>
<input type="file" name="file" id="fileInput" accept=".csv" class="w-full border border-gray-300 p-3 rounded-lg focus:outline-none focus:ring focus:ring-blue-400"> <input type="file" name="file" id="fileInput" accept=".csv"
class="w-full border border-gray-300 p-3 rounded-lg focus:outline-none focus:ring focus:ring-blue-400">
</div> </div>
<button type="submit" class="w-full bg-blue-600 text-white py-3 rounded-lg hover:bg-blue-700 focus:outline-none focus:ring focus:ring-blue-400">Upload</button> <button type="submit"
class="w-full bg-blue-600 text-white py-3 rounded-lg hover:bg-blue-700 focus:outline-none focus:ring focus:ring-blue-400">
Upload
</button>
</form> </form>
<!-- Error Display --> <!-- Error Display -->
...@@ -39,38 +43,56 @@ ...@@ -39,38 +43,56 @@
form.addEventListener('submit', async function (e) { form.addEventListener('submit', async function (e) {
e.preventDefault(); e.preventDefault();
const formData = new FormData();
formData.append('file', fileInput.files[0]);
const response = await fetch('', {
method: 'POST',
body: formData,
headers: {
'X-CSRFToken': '{{ csrf_token }}',
}
});
const result = await response.json();
// Clear previous messages // Clear previous messages
errorList.innerHTML = ''; errorList.innerHTML = '';
successMessage.innerHTML = ''; successMessage.innerHTML = '';
errorSection.classList.add('hidden'); errorSection.classList.add('hidden');
successSection.classList.add('hidden'); successSection.classList.add('hidden');
if (response.ok && result.status === 'success') { const formData = new FormData();
// Show success message formData.append('file', fileInput.files[0]);
successMessage.innerText = result.message;
successSection.classList.remove('hidden'); try {
} else if (response.status === 400 && result.status === 'error') { const response = await fetch('', {
// Show errors method: 'POST',
result.errors.forEach(error => { body: formData,
const li = document.createElement('li'); headers: {
li.textContent = error; 'X-CSRFToken': '{{ csrf_token }}',
errorList.appendChild(li); }
}); });
const result = await response.json();
if (response.ok && result.status === 'success') {
successMessage.innerText = result.message;
successSection.classList.remove('hidden');
} else if (response.status === 400 && result.status === 'error') {
// Handle form errors from the backend
if (Array.isArray(result.errors)) {
result.errors.forEach(errorObj => {
for (const [field, messages] of Object.entries(errorObj)) {
messages.forEach(message => {
const li = document.createElement('li');
li.textContent = `${field}: ${message}`;
errorList.appendChild(li);
});
}
});
} else {
const li = document.createElement('li');
li.textContent = result.errors;
errorList.appendChild(li);
}
errorSection.classList.remove('hidden');
}
} catch (error) {
// Handle unexpected errors
const li = document.createElement('li');
li.textContent = 'An unexpected error occurred. Please try again.';
errorList.appendChild(li);
errorSection.classList.remove('hidden'); errorSection.classList.remove('hidden');
} }
}); });
</script> </script>
{% endblock %} {% endblock %}
\ No newline at end of file
"""Contains the views for the file_validator app.""" """Views for the file_validator app."""
import csv from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import render
from django.http import JsonResponse
from django.urls import reverse_lazy from django.urls import reverse_lazy
from django.views.generic.edit import FormView from django.views.generic.edit import FormView
from file_validator.forms import CSVUploadForm from file_validator.forms import CSVUploadForm
def index_view(request: HttpRequest) -> HttpResponse:
"""Render the index page."""
return render(request, "index.html")
class CSVUploadView(FormView): class CSVUploadView(FormView):
"""View for uploading a CSV file.""" """View for uploading a CSV file."""
...@@ -18,40 +22,14 @@ class CSVUploadView(FormView): ...@@ -18,40 +22,14 @@ class CSVUploadView(FormView):
def get_context_data(self, **kwargs: dict) -> dict: def get_context_data(self, **kwargs: dict) -> dict:
"""Render the form with no error message on GET request.""" """Render the form with no error message on GET request."""
context = super().get_context_data(**kwargs) context = super().get_context_data(**kwargs)
context["error"] = None # No error message on GET request context["error"] = None
context["message"] = None # No success message on GET request context["message"] = None
return context return context
def form_valid(self, form: CSVUploadForm) -> JsonResponse: def form_valid(self, form: CSVUploadForm) -> JsonResponse: # noqa: ARG002
"""Handle the CSV validation and passes appropriate error or success messages to the template.""" """Handle the CSV validation and passes appropriate success messages to the template."""
file_obj = form.cleaned_data["file"] return JsonResponse({"status": "success", "message": "File is valid"})
try:
# Read and decode the CSV file
csv_file = file_obj.read().decode("utf-8").splitlines()
reader = csv.DictReader(csv_file)
# Example validation: Check for required columns
# TODO: Add the validation logic here. This is just a sample.
required_columns = ["name"]
fieldnames = reader.fieldnames if reader.fieldnames is not None else []
missing_columns = [col for col in required_columns if col not in fieldnames]
if missing_columns:
# If there are missing columns, return error message as JSON
return JsonResponse(
{"status": "error", "errors": [f"Missing columns: {", ".join(missing_columns)}"]}, status=400
)
return JsonResponse({"status": "success", "message": "CSV file is valid"})
except (UnicodeDecodeError, csv.Error) as e:
# If there"s an error (e.g., invalid file content), return the error message as JSON
return JsonResponse({"status": "error", "errors": [str(e)]}, status=400)
def form_invalid(self, form: CSVUploadForm) -> JsonResponse: def form_invalid(self, form: CSVUploadForm) -> JsonResponse:
"""Handle the form when it is invalid (e.g., wrong file type). """Handle the form when it is invalid (e.g., wrong file type or validation errors)."""
It renders the form again with errors.
"""
return JsonResponse({"status": "error", "errors": [form.errors]}, status=400) return JsonResponse({"status": "error", "errors": [form.errors]}, status=400)
...@@ -6,7 +6,7 @@ import sys ...@@ -6,7 +6,7 @@ import sys
def main(): def main():
"""Run administrative tasks.""" """Run administrative tasks."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings') os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'root.settings')
try: try:
from django.core.management import execute_from_command_line from django.core.management import execute_from_command_line
except ImportError as exc: except ImportError as exc:
......
""" """ASGI config for sage_validation project.
ASGI config for sage_validation project.
It exposes the ASGI callable as a module-level variable named ``application``. It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see For more information on this file, see
https://docs.djangoproject.com/en/5.1/howto/deployment/asgi/ https://docs.djangoproject.com/en/5.1/howto/deployment/asgi/
""" """
import os import os
from django.core.asgi import get_asgi_application from django.core.asgi import get_asgi_application
......
""" """Django settings for sage_validation project.
Django settings for sage_validation project.
Generated by 'django-admin startproject' using Django 5.1.1. Generated by 'django-admin startproject' using Django 5.1.1.
...@@ -9,7 +8,7 @@ https://docs.djangoproject.com/en/5.1/topics/settings/ ...@@ -9,7 +8,7 @@ https://docs.djangoproject.com/en/5.1/topics/settings/
For the full list of settings and their values, see For the full list of settings and their values, see
https://docs.djangoproject.com/en/5.1/ref/settings/ https://docs.djangoproject.com/en/5.1/ref/settings/
""" """
import os
from pathlib import Path from pathlib import Path
# Build paths inside the project like this: BASE_DIR / 'subdir'. # Build paths inside the project like this: BASE_DIR / 'subdir'.
...@@ -19,7 +18,7 @@ BASE_DIR = Path(__file__).resolve().parent.parent ...@@ -19,7 +18,7 @@ BASE_DIR = Path(__file__).resolve().parent.parent
# See https://docs.djangoproject.com/en/5.1/howto/deployment/checklist/ # See https://docs.djangoproject.com/en/5.1/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret! # SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = "django-insecure-9tdba&yktxzclzokj^=uxfsmisgeo8(6!p3koa8ndy8s^3x@@y" SECRET_KEY = os.getenv("SECRET_KEY")
# SECURITY WARNING: don't run with debug turned on in production! # SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True DEBUG = True
...@@ -60,7 +59,7 @@ ROOT_URLCONF = "root.urls" ...@@ -60,7 +59,7 @@ ROOT_URLCONF = "root.urls"
TEMPLATES = [ TEMPLATES = [
{ {
"BACKEND": "django.template.backends.django.DjangoTemplates", "BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [BASE_DIR / 'templates'], "DIRS": [BASE_DIR / "templates"],
"APP_DIRS": True, "APP_DIRS": True,
"OPTIONS": { "OPTIONS": {
"context_processors": [ "context_processors": [
...@@ -127,3 +126,8 @@ INTERNAL_IPS = [ ...@@ -127,3 +126,8 @@ INTERNAL_IPS = [
"127.0.0.1", "127.0.0.1",
] ]
TAILWIND_APP_NAME = "theme" TAILWIND_APP_NAME = "theme"
STATIC_ROOT = BASE_DIR / "staticfiles"
STATICFILES_DIRS = [
BASE_DIR / "static",
]
""" """WSGI config for sage_validation project.
WSGI config for sage_validation project.
It exposes the WSGI callable as a module-level variable named ``application``. It exposes the WSGI callable as a module-level variable named ``application``.
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment