import csv
import datetime
import logging
import os
import re
import shutil
import tempfile
import uuid
from decimal import Decimal, DecimalException
from typing import List, Tuple, Callable, Type
from django.db import IntegrityError
from django.utils.translation import gettext as _
from ferrosoft import strparse
from ferrosoft.apps.dataimport.api.serializers import ImportJobSerializer
from ferrosoft.apps.dataimport.models import (
ImportJob,
MappingCollection,
FieldMapping,
FieldType,
LogEntry,
LogHighlight,
FormatParameters,
DateFormatType,
UnknownEnumName,
)
from ferrosoft.apps.ferrobase import encodings
from ferrosoft.apps.ferrobase.services.taskqueue import background_task_tenant
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_transaction
[docs]
def import_job_lines(job: ImportJob, processor_state):
"""Process file lines from job as CSV, run process_import_line on each line.
:param ImportJob job: Job to process.
:param processor_state: Result from processor before job. Can be anything, is passed along to model processor.
"""
line_count = 0
group_name = job.queue_group_name()
# Copy contents to a temporary file, because Django File doesn't support the options
# that are required by this function. It must be named, to allow it to be opened several
# times, because for encoding detection the file must be read in binary mode, while for
# CSV parsing it is read in text mode.
buffer_file = tempfile.NamedTemporaryFile(mode="w+b", delete=False)
try:
shutil.copyfileobj(job.import_file.content, buffer_file)
buffer_file.close()
# If no encoding is specified, detect it.
text_encoding = job.text_encoding
if text_encoding is None:
with open(buffer_file.name, "rb") as f:
text_encoding = encodings.detect_text_encoding(f)
with open(
buffer_file.name, "r", encoding=text_encoding, newline=""
) as import_file:
# Ensure header names are stripped of leading and trailing whitespace.
header = [
h.strip() for h in import_file.readline().split(job.field_separator)
]
reader = csv.DictReader(
import_file, fieldnames=header, delimiter=job.field_separator
)
# Convert job to dict, to allow trivial pickling by the task queue.
# Avoid passing complex objects such as Django model instance.
job_data = ImportJobSerializer(job).data
for i, line in enumerate(reader):
background_task_tenant(
"ferrosoft.apps.dataimport.queue.process_import_line",
job_data,
i + 1,
line,
processor_state,
result_group=group_name,
task_stream="dataimport_lines",
)
line_count += 1
finally:
# Save line count for informational purposes.
job.line_count = line_count
job.save(update_fields=["line_count"])
buffer_file.close()
os.unlink(buffer_file.name)
def _convert_value(value: str, field: FieldMapping, format_params: FormatParameters):
match FieldType(field.field_type):
case FieldType.BOOLEAN:
return strparse.parse_boolean(value)
case FieldType.DECIMAL:
if value == "":
return None
# Numbers may use a decimal separator other than dot. Replace it with dot.
if format_params.decimal_separator != ".":
value = value.replace(format_params.decimal_separator, ".")
try:
return Decimal(value)
except DecimalException:
raise ValueError(_("Decimal number is incorrectly formatted"))
case FieldType.DATE:
if value == "":
return None
if format_params.date_format == DateFormatType.ISO_8601:
return datetime.datetime.fromisoformat(value)
if format_params.date_format_codes == "":
raise ValueError("Date format codes empty")
return datetime.datetime.strptime(value, format_params.date_format_codes)
case FieldType.DATE_TIME:
if value == "":
return None
if format_params.date_format == DateFormatType.ISO_8601:
return datetime.datetime.fromisoformat(value)
if format_params.date_time_format_codes == "":
raise ValueError("Date time format codes empty")
return datetime.datetime.strptime(
value, format_params.date_time_format_codes
)
case FieldType.INTEGER:
return int(value)
case FieldType.TEXT:
enum_conv = field.get_enum_converter()
if enum_conv is not None:
try:
return enum_conv.convert(value).value
except UnknownEnumName as e:
if field.required:
raise e
logging.warning("Falling back to null: %s", str(e))
return None
return value
case _:
raise ValueError(_("Unsupported field type"))
def _resolve_value(value: str, mapping: FieldMapping):
related_model = mapping.get_related_model()
# Filter related model by specified field.
# Allow None value to be returned.
return related_model.objects.filter(**{mapping.related_field: value}).first()
_extract_pattern = re.compile(r"^.*[Uu]nique.*\((.+?)\)=\((.+)\)", re.DOTALL)
_field_pattern = re.compile(r"^[\"\s]*(.+?)[\"\s]*$")
[docs]
def create_or_get_model[T](model_type: Type[T], creator: Callable[[], T]) -> T:
"""
Create model or get it if it exists already.
If creator function raises IntegrityError, get model
by the unique fields reported in the exception.
Args:
model_type: Model class type.
creator: Function creating the model and returning it.
Returns: Model instance.
"""
try:
with tenant_db_transaction():
return creator()
except IntegrityError as e:
unique_fields = extract_unique_fields(e)
if unique_fields is None:
raise e
return model_type.objects.get(**unique_fields)
def _create_model_optimistically(model_type, model_data: dict):
return create_or_get_model(
model_type, lambda: model_type.objects.create(**model_data)
)
[docs]
def convert_line(
line_no: int,
line: dict,
job: ImportJob,
field_mappings: List[FieldMapping],
collection: MappingCollection,
) -> Tuple[object | None, List[LogEntry]]:
"""
Convert line to model instance and save the latter.
Args:
line_no: Current line number
line: Current line
job: ImportJob where the current line belongs to.
field_mappings: Field mappings associated with the collection.
collection: Mapping collection associated with the job.
Returns: Created model (or None) and list of LogEntry objects created by this function.
"""
log_entries = []
model = None
try:
# Get model type class that can be instantiated with populated model_data.
model_type = collection.get_model()
model_data = {}
# Populate model_data by traversing field_mappings.
for mapping in field_mappings:
field_type = FieldType(mapping.field_type)
# Handle missing field. It is treated only as a warning, because None
# values could be desired for fields which allow None. Primary key field
# is allowed silently, because it is usually a generated value.
if mapping.input_field_name not in line and mapping.required:
log_entries.append(
LogEntry.create(
job,
line_no,
_("Field not found: %(name)s")
% {"name": mapping.input_field_name},
highlight=LogHighlight.WARNING,
)
)
# Get value from supplied CSV line. CSV lines only ever return string.
value: str | None | uuid.UUID = line.get(mapping.input_field_name, None)
original_value = value
if value == "NULL":
# Treat this special value as None, because CSV doesn't support None values.
value = None
elif field_type == FieldType.PRIMARY_KEY:
# Generate primary key if not supplied.
value = uuid.uuid4() if value is None else uuid.UUID(value)
elif field_type == FieldType.FOREIGN_KEY:
# Resolve foreign key.
value = _resolve_value(value, mapping)
elif isinstance(value, str):
# For any other field types, convert value to expected type.
value = _convert_value(
value,
mapping,
job.format_parameters,
)
if mapping.required and value is None:
if field_type == FieldType.FOREIGN_KEY:
raise ValueError(
_("Required %(name)s %(value)s not found")
% {
"name": mapping.input_field_name,
"value": original_value,
}
)
else:
raise ValueError(
_("Required field %(name)s is blank")
% {"name": mapping.input_field_name}
)
# Data is ready.
model_data[mapping.output_field_name] = value
# Save the model without further ado.
model = _create_model_optimistically(model_type, model_data)
except Exception as e:
log_entries.append(
LogEntry.create(
job,
line_no,
str(e),
highlight=LogHighlight.DANGER,
)
)
else:
context = {
"model": collection.get_model_without_app(),
"id": str(model.pk),
"repr": str(model),
}
log_entries.append(
LogEntry.create(
job,
line_no,
_("Created %(model)s %(repr)s with ID %(id)s") % context,
highlight=LogHighlight.SUCCESS,
context=context | {"type": "model_created"},
)
)
return model, log_entries