import base64
import io
import traceback
from traceback import print_exception
from PIL import Image
from asgiref.sync import sync_to_async
from django.conf import settings
from django.template import TemplateDoesNotExist
from ferrosoft.apps.befundung import email, sftp
from ferrosoft.apps.befundung.api.serializers import (
ExaminationSerializer,
)
from ferrosoft.apps.befundung.models import (
Examination,
create_log,
Photo,
Report,
Settings,
ExaminationRequest,
TransformedPhoto,
)
from ferrosoft.apps.ferrobase.models import zero
from ferrosoft.apps.ferrobase.services.taskqueue import tenant_task
from ferrosoft.report import generate_report
def _refusal_display(examination: Examination) -> tuple[bool, bool]:
"""
Determine whether to show total refusal and partial refusal.
:param examination: Beschreibung
:type examination: Examination
:return: Whether to show total refusal and partial refusal.
:rtype: tuple[bool, bool]
"""
is_total_refusal = examination.refused_material != ""
is_partial_refusal = any(
[refusal.value > zero for refusal in examination.refusals.all()]
)
if not is_total_refusal and not is_partial_refusal:
return False, True
return is_total_refusal, is_partial_refusal
[docs]
@sync_to_async
@tenant_task()
def report_examination(examination_id: str):
report = None
weighing_slip_id = ""
try:
examination: Examination = Examination.objects.get(pk=examination_id)
except Exception as e:
create_log(
"could not process examination with ID %s: %s"
% (
examination_id,
str(e),
)
)
return
else:
weighing_slip_id = examination.weighing_slip_id
# The following code does not use examination model directly,
# but instead its dict representation (as exported by the serializer).
# This is due to the model being introduced after the following
# code already existed. Since the examination dict is used as kwargs
# in various places, it is currently hard to replace it with model instance.
serializer = ExaminationSerializer(examination)
report = Report(
examination=serializer.data,
images=list(
filter(
lambda it: it is not None,
[_transform_photo(photo) for photo in examination.photos.all()],
)
),
)
try:
config = Settings.get_instance()
# Special condition determining whether to show
enable_total_refusal, enable_partial_refusal = _refusal_display(examination)
pdf_blob = generate_report(
config.get_template(),
{
"enable_partial_refusal": enable_partial_refusal,
"enable_total_refusal": enable_total_refusal,
"examination": report.examination,
"photos": report.images,
"config": config,
},
)
email.send_examination_mail(report, config, pdf_blob)
create_log("sent mail with weighing_slip_id=%s" % weighing_slip_id)
sftp.put_examination_report(report, config, pdf_blob)
create_log(
"uploaded file to SFTP server with weighing_slip_id=%s" % weighing_slip_id
)
except Exception as e:
if settings.DEBUG:
if isinstance(e, TemplateDoesNotExist):
for inner_e in e.chain:
print_exception(inner_e)
raise e
else:
create_log(
"failed processing examination report with weighing_slip_id=%s: %s%s"
% (
weighing_slip_id,
str(e),
"" if not settings.DEBUG else "\n" + traceback.format_exc(),
)
)
def _transform_photo(photo: Photo) -> TransformedPhoto | None:
try:
image_contents = photo.image_file.read()
except FileNotFoundError:
return None
else:
image = Image.open(io.BytesIO(image_contents))
media_type = "image/%s" % image.format.lower()
return TransformedPhoto(
media_type,
image_contents,
base64.b64encode(image_contents).decode(),
)
[docs]
@sync_to_async
@tenant_task()
def report_examination_errors(request_id, errors, replay_uri):
exam_report = ExaminationRequest.objects.get(pk=request_id)
weighing_slip_id = exam_report.payload.get("weighing_slip_id", "")
error_report = generate_report(
"befundung/errors.html",
as_html=True,
as_text=True,
context={
"errors": errors,
"replay_uri": replay_uri,
"weighing_slip_id": weighing_slip_id,
},
)
config = Settings.get_instance()
email.send_error_mail(error_report, config)