mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-07-09 02:44:10 -04:00
ugh
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
import csv
|
||||
from io import StringIO
|
||||
import re
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
@@ -28,6 +30,16 @@ class CsvFileValidator:
|
||||
self.message = message
|
||||
|
||||
def __call__(self, form, field):
|
||||
|
||||
data = Spreadsheet.from_file_form(form).as_dict
|
||||
data = data["data"]
|
||||
if data is not None and data != "":
|
||||
csv_file = StringIO(data)
|
||||
reader = csv.reader(csv_file)
|
||||
first_line = next(reader, None)
|
||||
if first_line is None or all(cell.strip() == "" for cell in first_line):
|
||||
raise ValidationError(f"No headers on row 1 in {field.data.filename}")
|
||||
|
||||
if not Spreadsheet.can_handle(field.data.filename):
|
||||
raise ValidationError(
|
||||
"{} is not a spreadsheet that Notify can read".format(
|
||||
|
||||
@@ -46,7 +46,7 @@ from app.utils import (
|
||||
PermanentRedirect,
|
||||
hilite,
|
||||
should_skip_template_page,
|
||||
# unicode_truncate,
|
||||
unicode_truncate,
|
||||
)
|
||||
from app.utils.csv import Spreadsheet, get_errors_for_csv
|
||||
from app.utils.templates import get_template
|
||||
@@ -54,7 +54,7 @@ from app.utils.user import user_has_permissions
|
||||
from notifications_utils import SMS_CHAR_COUNT_LIMIT
|
||||
from notifications_utils.insensitive_dict import InsensitiveDict
|
||||
from notifications_utils.recipients import RecipientCSV, first_column_headings
|
||||
# from notifications_utils.sanitise_text import SanitiseASCII
|
||||
from notifications_utils.sanitise_text import SanitiseASCII
|
||||
|
||||
|
||||
def get_example_csv_fields(column_headers, use_example_as_example, submitted_fields):
|
||||
@@ -132,26 +132,25 @@ def send_messages(service_id, template_id):
|
||||
form = CsvUploadForm()
|
||||
if form.validate_on_submit():
|
||||
try:
|
||||
data = Spreadsheet.from_file_form(form).as_dict
|
||||
raise Exception(f"Data in send: {data}")
|
||||
# upload_id = s3upload(
|
||||
# service_id,
|
||||
# Spreadsheet.from_file_form(form).as_dict,
|
||||
# )
|
||||
# file_name_metadata = unicode_truncate(
|
||||
# SanitiseASCII.encode(form.file.data.filename), 1600
|
||||
# )
|
||||
# set_metadata_on_csv_upload(
|
||||
# service_id, upload_id, original_file_name=file_name_metadata
|
||||
# )
|
||||
# return redirect(
|
||||
# url_for(
|
||||
# ".check_messages",
|
||||
# service_id=service_id,
|
||||
# upload_id=upload_id,
|
||||
# template_id=template.id,
|
||||
# )
|
||||
# )
|
||||
|
||||
upload_id = s3upload(
|
||||
service_id,
|
||||
Spreadsheet.from_file_form(form).as_dict,
|
||||
)
|
||||
file_name_metadata = unicode_truncate(
|
||||
SanitiseASCII.encode(form.file.data.filename), 1600
|
||||
)
|
||||
set_metadata_on_csv_upload(
|
||||
service_id, upload_id, original_file_name=file_name_metadata
|
||||
)
|
||||
return redirect(
|
||||
url_for(
|
||||
".check_messages",
|
||||
service_id=service_id,
|
||||
upload_id=upload_id,
|
||||
template_id=template.id,
|
||||
)
|
||||
)
|
||||
except (UnicodeDecodeError, BadZipFile, XLRDError):
|
||||
flash(
|
||||
"Could not read {}. Try using a different file format.".format(
|
||||
|
||||
@@ -32,7 +32,6 @@ class Spreadsheet:
|
||||
self._csv_data = converted.getvalue()
|
||||
return self._csv_data
|
||||
|
||||
|
||||
@classmethod
|
||||
def can_handle(cls, filename):
|
||||
return cls.get_extension(filename) in cls.ALLOWED_FILE_EXTENSIONS
|
||||
@@ -46,7 +45,6 @@ class Spreadsheet:
|
||||
rows = file_content.read().decode("utf-8").splitlines()
|
||||
return "\r\n".join(rows)
|
||||
|
||||
|
||||
@classmethod
|
||||
def from_rows(cls, rows, filename=""):
|
||||
return cls(rows=rows, filename=filename)
|
||||
|
||||
Reference in New Issue
Block a user