#1839

Committed 26 Jun 2024 09:52AM UTC coverage: 84.749% (-0.1%) from 84.896%

Build # #1839

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #1528 from akvo/feature/1524-hh-consent-question

Feature/1524 hh consent question

Run Details

3056 of 3742 branches covered (81.67%)

Branch coverage included in aggregate %.

6852 of 7949 relevant lines covered (86.2%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

48.03

backend/api/v1/v1_jobs/job.py

import logging
import os
from django_q.models import Task

import pandas as pd
from django.utils import timezone
from django_q.tasks import async_task
from django.db.models import Subquery, Max
from api.v1.v1_jobs.administrations_bulk_upload import (
    seed_administration_data,
    validate_administrations_bulk_upload,
)
from utils.upload_entities import (
    validate_entity_file,
    validate_entity_data,
)
from api.v1.v1_profile.constants import UserRoleTypes
from api.v1.v1_forms.models import Forms, QuestionOptions
from api.v1.v1_forms.constants import SubmissionTypes, QuestionTypes
from api.v1.v1_jobs.constants import JobStatus, JobTypes

# from api.v1.v1_jobs.functions import HText
from api.v1.v1_jobs.models import Jobs
from api.v1.v1_jobs.seed_data import seed_excel_data
from api.v1.v1_jobs.validate_upload import validate
from api.v1.v1_profile.models import Administration, EntityData
from api.v1.v1_users.models import SystemUser
from utils import storage
from utils.email_helper import send_email, EmailTypes
from utils.export_form import (
    generate_definition_sheet,
    get_question_names,
    blank_data_template,
    meta_columns,
)
from utils.functions import update_date_time_format
from utils.storage import upload
from utils.custom_generator import generate_sqlite

logger = logging.getLogger(__name__)


def download_data(
    form: Forms, administration_ids, download_type="all", submission_type=None
):
    filter_data = {}
    if administration_ids and submission_type != SubmissionTypes.certification:
        filter_data["administration_id__in"] = administration_ids
    if administration_ids and submission_type == SubmissionTypes.certification:
        filter_adm_key = "created_by__user_access__administration_id__in"
        filter_data[filter_adm_key] = administration_ids
    if submission_type:
        filter_data["submission_type"] = submission_type
    else:
        filter_data["submission_type__in"] = [
            SubmissionTypes.registration,
            SubmissionTypes.monitoring,
        ]
    data = form.form_form_data.filter(**filter_data)
    if download_type == "recent":
        latest_per_uuid = (
            data.values("uuid")
            .annotate(latest_created=Max("created"))
            .values("latest_created")
        )
        data = data.filter(created__in=Subquery(latest_per_uuid))
    return [d.to_data_frame for d in data.order_by("id")]


def get_answer_label(answer_values, question_id):
    if answer_values is None or answer_values != answer_values:
        return answer_values
    answer_value = answer_values.split("|")
    answer_label = []
    for value in answer_value:
        options = QuestionOptions.objects.filter(
            question_id=question_id, value=value
        ).first()
        if options:
            answer_label.append(options.label)
    return "|".join(answer_label)


def generate_data_sheet(
    writer: pd.ExcelWriter,
    form: Forms,
    administration_ids=None,
    submission_type: SubmissionTypes = None,
    download_type: str = "all",
    use_label: bool = False,
) -> None:
    questions = get_question_names(form=form)
    data = download_data(
        form=form,
        administration_ids=administration_ids,
        submission_type=submission_type,
        download_type=download_type,
    )
    if len(data):
        df = pd.DataFrame(data)
        new_columns = {}
        for question in questions:
            question_id, question_name, question_type = question
            if question_name not in df:
                df[question_name] = None
            if use_label:
                if question_type in [
                    QuestionTypes.option,
                    QuestionTypes.multiple_option,
                ]:
                    new_columns[question_name] = df[question_name].apply(
                        lambda x: get_answer_label(x, question_id)
                    )
        question_names = [question[1] for question in questions]
        if use_label:
            new_df = pd.DataFrame(new_columns)
            df.drop(columns=list(new_df), inplace=True)
            df = pd.concat([df, new_df], axis=1)
        # Reorder columns
        df = df[meta_columns + question_names]
        df.to_excel(writer, sheet_name="data", index=False)
        generate_definition_sheet(form=form, writer=writer)
    else:
        blank_data_template(form=form, writer=writer)


def job_generate_data_download(job_id, **kwargs):
    job = Jobs.objects.get(pk=job_id)
    file_path = "./tmp/{0}".format(job.result)
    if os.path.exists(file_path):
        os.remove(file_path)
    administration_ids = False
    administration_name = "All Administration Level"
    if kwargs.get("administration"):
        administration = Administration.objects.get(
            pk=kwargs.get("administration")
        )
        if administration.path:
            filter_path = "{0}{1}.".format(
                administration.path, administration.id
            )
        else:
            filter_path = f"{administration.id}."
        administration_ids = list(
            Administration.objects.filter(
                path__startswith=filter_path
            ).values_list("id", flat=True)
        )
        administration_ids.append(administration.id)

        administration_name = list(
            Administration.objects.filter(
                path__startswith=filter_path
            ).values_list("name", flat=True)
        )
    form = Forms.objects.get(pk=job.info.get("form_id"))
    download_type = kwargs.get("download_type")
    submission_type = kwargs.get("submission_type")
    writer = pd.ExcelWriter(file_path, engine="xlsxwriter")

    generate_data_sheet(
        writer=writer,
        form=form,
        administration_ids=administration_ids,
        submission_type=submission_type,
        download_type=download_type,
        use_label=job.info.get("use_label"),
    )

    context = [
        {"context": "Form Name", "value": form.name},
        {
            "context": "Download Date",
            "value": update_date_time_format(job.created),
        },
        {
            "context": "Administration",
            "value": ",".join(administration_name)
            if isinstance(administration_name, list)
            else administration_name,
        },
    ]

    context = (
        pd.DataFrame(context).groupby(["context", "value"], sort=False).first()
    )
    context.to_excel(writer, sheet_name="context", startrow=0, header=False)
    workbook = writer.book
    worksheet = writer.sheets["context"]
    f = workbook.add_format(
        {
            "align": "left",
            "bold": False,
            "border": 0,
        }
    )
    worksheet.set_column("A:A", 20, f)
    worksheet.set_column("B:B", 30, f)
    merge_format = workbook.add_format(
        {
            "bold": True,
            "border": 1,
            "align": "center",
            "valign": "vcenter",
            "fg_color": "#45add9",
            "color": "#ffffff",
        }
    )
    worksheet.merge_range("A1:B1", "Context", merge_format)
    writer.save()
    url = upload(file=file_path, folder="download")
    return url


def job_generate_data_download_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    if task.success:
        job.status = JobStatus.done
        job.available = timezone.now()
    else:
        job.status = JobStatus.failed
    job.save()


def seed_data_job(job_id):
    try:
        job = Jobs.objects.get(pk=job_id)
        seed_excel_data(job)
    except (ValueError, OSError, NameError, TypeError):
        print("Error seed data job")
        return False
    except Exception as unknown_error:
        print("Unknown error seed data job", unknown_error)
        return False
    return True


def seed_data_job_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    is_super_admin = job.user.user_access.role == UserRoleTypes.super_admin
    if task.result:
        job.status = JobStatus.done
        job.available = timezone.now()
        form_id = job.info.get("form")
        form = Forms.objects.filter(pk=int(form_id)).first()
        file = job.info.get("file")
        storage.download(f"upload/{file}")
        df = pd.read_excel(f"./tmp/{file}", sheet_name="data")
        subject = (
            "New Data Uploaded"
            if is_super_admin
            else "New Request @{0}".format(job.user.get_full_name())
        )
        data = {
            "subject": subject,
            "title": "New Data Submission",
            "send_to": [job.user.email],
            "listing": [
                {
                    "name": "Upload Date",
                    "value": job.created.strftime("%m-%d-%Y, %H:%M:%S"),
                },
                {"name": "Questionnaire", "value": form.name},
                {"name": "Number of Records", "value": df.shape[0]},
            ],
            "is_super_admin": is_super_admin,
        }
        send_email(context=data, type=EmailTypes.new_request)
    else:
        job.status = JobStatus.failed
    job.save()


def validate_excel(job_id):
    job = Jobs.objects.get(pk=job_id)
    storage.download(f"upload/{job.info.get('file')}")
    data = validate(
        job.info.get("form"),
        job.info.get("administration"),
        f"./tmp/{job.info.get('file')}",
    )

    if len(data):
        form_id = job.info.get("form")
        form = Forms.objects.filter(pk=int(form_id)).first()
        file = job.info.get("file")
        df = pd.read_excel(f"./tmp/{file}", sheet_name="data")
        error_list = pd.DataFrame(data)
        error_list = error_list[
            list(filter(lambda x: x != "error", list(error_list)))
        ]
        error_file = f"./tmp/error-{job_id}.csv"
        error_list.to_csv(error_file, index=False)
        data = {
            "send_to": [job.user.email],
            "listing": [
                {
                    "name": "Upload Date",
                    "value": job.created.strftime("%m-%d-%Y, %H:%M:%S"),
                },
                {"name": "Questionnaire", "value": form.name},
                {"name": "Number of Records", "value": df.shape[0]},
            ],
        }
        send_email(
            context=data,
            type=EmailTypes.upload_error,
            path=error_file,
            content_type="text/csv",
        )
        return False
    return True


def validate_excel_result(task):
    job = Jobs.objects.get(task_id=task.id)
    job.attempt = job.attempt + 1
    job_info = job.info
    if task.result:
        job.status = JobStatus.done
        job.available = timezone.now()
        job.save()
        job_info.update({"ref_job_id": job.id})
        new_job = Jobs.objects.create(
            result=job.info.get("file"),
            type=JobTypes.seed_data,
            status=JobStatus.on_progress,
            user=job.user,
            info=job_info,
        )
        task_id = async_task(
            "api.v1.v1_jobs.job.seed_data_job",
            new_job.id,
            hook="api.v1.v1_jobs.job.seed_data_job_result",
        )
        new_job.task_id = task_id
        new_job.save()
    else:
        job.status = JobStatus.failed
        job.save()


def handle_administrations_bulk_upload(filename, user_id, upload_time):
    user = SystemUser.objects.get(id=user_id)
    storage.download(f"upload/{filename}")
    file_path = f"./tmp/{filename}"
    errors = validate_administrations_bulk_upload(file_path)
    xlsx = pd.ExcelFile(file_path)
    if "data" not in xlsx.sheet_names:
        logger.error(f"Sheet 'data' not found in {filename}")
        send_email(
            context={
                "send_to": [user.email],
                "listing": [
                    {
                        "name": "Upload Date",
                        "value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
                    },
                    {
                        "name": "Questionnaire",
                        "value": "Administrative List",
                    },
                    {
                        "name": "Error",
                        "value": 'Sheet "data" not found',
                    },
                ],
            },
            type=EmailTypes.upload_error,
        )
        return
    df = pd.read_excel(file_path, sheet_name="data")
    email_context = {
        "send_to": [user.email],
        "listing": [
            {
                "name": "Upload Date",
                "value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
            },
            {
                "name": "Questionnaire",
                "value": "Administrative List",
            },
            {"name": "Number of Records", "value": df.shape[0]},
        ],
    }
    if len(errors):
        logger.error(errors)
        error_file = (
            "./tmp/administration-error-"
            f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
        )
        error_list = pd.DataFrame(errors)
        error_list.to_csv(error_file, index=False)
        send_email(
            context=email_context,
            type=EmailTypes.upload_error,
            path=error_file,
            content_type="text/csv",
        )
        return
    seed_administration_data(file_path)
    generate_sqlite(Administration)
    send_email(context=email_context, type=EmailTypes.administration_upload)


def handle_master_data_bulk_upload_failure(task: Task):
    if task.success:
        return
    logger.error(
        {
            "error": "Failed running background job",
            "id": task.id,
            "name": task.name,
            "started": task.started,
            "stopped": task.stopped,
            "args": task.args,
            "kwargs": task.kwargs,
            "body": task.result,
        }
    )


def handle_entities_error_upload(
    errors: list,
    email_context: dict,
    user: SystemUser,
    upload_time: timezone.datetime,
):
    logger.error(errors)
    error_file = (
        "./tmp/entity-error-"
        f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
    )
    error_list = pd.DataFrame(errors)
    error_list.to_csv(error_file, index=False)
    send_email(
        context=email_context,
        type=EmailTypes.upload_error,
        path=error_file,
        content_type="text/csv",
    )


def handle_entities_bulk_upload(filename, user_id, upload_time):
    user = SystemUser.objects.get(id=user_id)
    storage.download(f"upload/{filename}")
    file_path = f"./tmp/{filename}"
    errors = validate_entity_file(file_path)
    email_context = {
        "send_to": [user.email],
        "listing": [
            {
                "name": "Upload Date",
                "value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
            },
            {
                "name": "Questionnaire",
                "value": "Entity List",
            },
        ],
    }
    if len(errors):
        handle_entities_error_upload(errors, email_context, user, upload_time)
        return
    errors = validate_entity_data(file_path)
    if len(errors):
        handle_entities_error_upload(errors, email_context, user, upload_time)
        return
    generate_sqlite(EntityData)

1	import logging	1✔
2	import os	1✔
3	from django_q.models import Task	1✔
4
5	import pandas as pd	1✔
6	from django.utils import timezone	1✔
7	from django_q.tasks import async_task	1✔
8	from django.db.models import Subquery, Max	1✔
9	from api.v1.v1_jobs.administrations_bulk_upload import (	1✔
10	seed_administration_data,
11	validate_administrations_bulk_upload,
12	)
13	from utils.upload_entities import (	1✔
14	validate_entity_file,
15	validate_entity_data,
16	)
17	from api.v1.v1_profile.constants import UserRoleTypes	1✔
18	from api.v1.v1_forms.models import Forms, QuestionOptions	1✔
19	from api.v1.v1_forms.constants import SubmissionTypes, QuestionTypes	1✔
20	from api.v1.v1_jobs.constants import JobStatus, JobTypes	1✔
21
22	# from api.v1.v1_jobs.functions import HText
23	from api.v1.v1_jobs.models import Jobs	1✔
24	from api.v1.v1_jobs.seed_data import seed_excel_data	1✔
25	from api.v1.v1_jobs.validate_upload import validate	1✔
26	from api.v1.v1_profile.models import Administration, EntityData	1✔
27	from api.v1.v1_users.models import SystemUser	1✔
28	from utils import storage	1✔
29	from utils.email_helper import send_email, EmailTypes	1✔
30	from utils.export_form import (	1✔
31	generate_definition_sheet,
32	get_question_names,
33	blank_data_template,
34	meta_columns,
35	)
36	from utils.functions import update_date_time_format	1✔
37	from utils.storage import upload	1✔
38	from utils.custom_generator import generate_sqlite	1✔
39
40	logger = logging.getLogger(__name__)	1✔
41
42
43	def download_data(	1✔
44	form: Forms, administration_ids, download_type="all", submission_type=None
45	):
46	filter_data = {}	1✔
47	if administration_ids and submission_type != SubmissionTypes.certification:	1✔
48	filter_data["administration_id__in"] = administration_ids	1✔
49	if administration_ids and submission_type == SubmissionTypes.certification:	1✔
50	filter_adm_key = "created_by__user_access__administration_id__in"	1✔
51	filter_data[filter_adm_key] = administration_ids	1✔
52	if submission_type:	1✔
53	filter_data["submission_type"] = submission_type	1✔
54	else:
55	filter_data["submission_type__in"] = [	1✔
56	SubmissionTypes.registration,
57	SubmissionTypes.monitoring,
58	]
59	data = form.form_form_data.filter(**filter_data)	1✔
60	if download_type == "recent":	1✔
61	latest_per_uuid = (	1✔
62	data.values("uuid")
63	.annotate(latest_created=Max("created"))
64	.values("latest_created")
65	)
66	data = data.filter(created__in=Subquery(latest_per_uuid))	1✔
67	return [d.to_data_frame for d in data.order_by("id")]	1✔
68
69
70	def get_answer_label(answer_values, question_id):	1✔
71	if answer_values is None or answer_values != answer_values:	×
72	return answer_values	×
73	answer_value = answer_values.split("\|")	×
74	answer_label = []	×
75	for value in answer_value:	×
76	options = QuestionOptions.objects.filter(	×
77	question_id=question_id, value=value
78	).first()
79	if options:	×
80	answer_label.append(options.label)	×
81	return "\|".join(answer_label)	×
82
83
84	def generate_data_sheet(	1✔
85	writer: pd.ExcelWriter,
86	form: Forms,
87	administration_ids=None,
88	submission_type: SubmissionTypes = None,
89	download_type: str = "all",
90	use_label: bool = False,
91	) -> None:
92	questions = get_question_names(form=form)	1✔
93	data = download_data(	1✔
94	form=form,
95	administration_ids=administration_ids,
96	submission_type=submission_type,
97	download_type=download_type,
98	)
99	if len(data):	1✔
100	df = pd.DataFrame(data)	1✔
101	new_columns = {}	1✔
102	for question in questions:	1✔
103	question_id, question_name, question_type = question	1✔
104	if question_name not in df:	1!
105	df[question_name] = None	×
106	if use_label:	1!
107	if question_type in [	×
108	QuestionTypes.option,
109	QuestionTypes.multiple_option,
110	]:
111	new_columns[question_name] = df[question_name].apply(	×
112	lambda x: get_answer_label(x, question_id)
113	)
114	question_names = [question[1] for question in questions]	1✔
115	if use_label:	1!
116	new_df = pd.DataFrame(new_columns)	×
117	df.drop(columns=list(new_df), inplace=True)	×
118	df = pd.concat([df, new_df], axis=1)	×
119	# Reorder columns
120	df = df[meta_columns + question_names]	1✔
121	df.to_excel(writer, sheet_name="data", index=False)	1✔
122	generate_definition_sheet(form=form, writer=writer)	1✔
123	else:
124	blank_data_template(form=form, writer=writer)	1✔
125
126
127	def job_generate_data_download(job_id, **kwargs):	1✔
128	job = Jobs.objects.get(pk=job_id)	1✔
129	file_path = "./tmp/{0}".format(job.result)	1✔
130	if os.path.exists(file_path):	1!
131	os.remove(file_path)	×
132	administration_ids = False	1✔
133	administration_name = "All Administration Level"	1✔
134	if kwargs.get("administration"):	1✔
135	administration = Administration.objects.get(	1✔
136	pk=kwargs.get("administration")
137	)
138	if administration.path:	1!
139	filter_path = "{0}{1}.".format(	1✔
140	administration.path, administration.id
141	)
142	else:
143	filter_path = f"{administration.id}."	×
144	administration_ids = list(	1✔
145	Administration.objects.filter(
146	path__startswith=filter_path
147	).values_list("id", flat=True)
148	)
149	administration_ids.append(administration.id)	1✔
150
151	administration_name = list(	1✔
152	Administration.objects.filter(
153	path__startswith=filter_path
154	).values_list("name", flat=True)
155	)
156	form = Forms.objects.get(pk=job.info.get("form_id"))	1✔
157	download_type = kwargs.get("download_type")	1✔
158	submission_type = kwargs.get("submission_type")	1✔
159	writer = pd.ExcelWriter(file_path, engine="xlsxwriter")	1✔
160
161	generate_data_sheet(	1✔
162	writer=writer,
163	form=form,
164	administration_ids=administration_ids,
165	submission_type=submission_type,
166	download_type=download_type,
167	use_label=job.info.get("use_label"),
168	)
169
170	context = [	1✔
171	{"context": "Form Name", "value": form.name},
172	{
173	"context": "Download Date",
174	"value": update_date_time_format(job.created),
175	},
176	{
177	"context": "Administration",
178	"value": ",".join(administration_name)
179	if isinstance(administration_name, list)
180	else administration_name,
181	},
182	]
183
184	context = (	1✔
185	pd.DataFrame(context).groupby(["context", "value"], sort=False).first()
186	)
187	context.to_excel(writer, sheet_name="context", startrow=0, header=False)	1✔
188	workbook = writer.book	1✔
189	worksheet = writer.sheets["context"]	1✔
190	f = workbook.add_format(	1✔
191	{
192	"align": "left",
193	"bold": False,
194	"border": 0,
195	}
196	)
197	worksheet.set_column("A:A", 20, f)	1✔
198	worksheet.set_column("B:B", 30, f)	1✔
199	merge_format = workbook.add_format(	1✔
200	{
201	"bold": True,
202	"border": 1,
203	"align": "center",
204	"valign": "vcenter",
205	"fg_color": "#45add9",
206	"color": "#ffffff",
207	}
208	)
209	worksheet.merge_range("A1:B1", "Context", merge_format)	1✔
210	writer.save()	1✔
211	url = upload(file=file_path, folder="download")	1✔
212	return url	1✔
213
214
215	def job_generate_data_download_result(task):	1✔
216	job = Jobs.objects.get(task_id=task.id)	×
217	job.attempt = job.attempt + 1	×
218	if task.success:	×
219	job.status = JobStatus.done	×
220	job.available = timezone.now()	×
221	else:
222	job.status = JobStatus.failed	×
223	job.save()	×
224
225
226	def seed_data_job(job_id):	1✔
227	try:	×
228	job = Jobs.objects.get(pk=job_id)	×
229	seed_excel_data(job)	×
230	except (ValueError, OSError, NameError, TypeError):	×
231	print("Error seed data job")	×
232	return False	×
233	except Exception as unknown_error:	×
234	print("Unknown error seed data job", unknown_error)	×
235	return False	×
236	return True	×
237
238
239	def seed_data_job_result(task):	1✔
240	job = Jobs.objects.get(task_id=task.id)	×
241	job.attempt = job.attempt + 1	×
242	is_super_admin = job.user.user_access.role == UserRoleTypes.super_admin	×
243	if task.result:	×
244	job.status = JobStatus.done	×
245	job.available = timezone.now()	×
246	form_id = job.info.get("form")	×
247	form = Forms.objects.filter(pk=int(form_id)).first()	×
248	file = job.info.get("file")	×
249	storage.download(f"upload/{file}")	×
250	df = pd.read_excel(f"./tmp/{file}", sheet_name="data")	×
251	subject = (	×
252	"New Data Uploaded"
253	if is_super_admin
254	else "New Request @{0}".format(job.user.get_full_name())
255	)
256	data = {	×
257	"subject": subject,
258	"title": "New Data Submission",
259	"send_to": [job.user.email],
260	"listing": [
261	{
262	"name": "Upload Date",
263	"value": job.created.strftime("%m-%d-%Y, %H:%M:%S"),
264	},
265	{"name": "Questionnaire", "value": form.name},
266	{"name": "Number of Records", "value": df.shape[0]},
267	],
268	"is_super_admin": is_super_admin,
269	}
270	send_email(context=data, type=EmailTypes.new_request)	×
271	else:
272	job.status = JobStatus.failed	×
273	job.save()	×
274
275
276	def validate_excel(job_id):	1✔
277	job = Jobs.objects.get(pk=job_id)	×
278	storage.download(f"upload/{job.info.get('file')}")	×
279	data = validate(	×
280	job.info.get("form"),
281	job.info.get("administration"),
282	f"./tmp/{job.info.get('file')}",
283	)
284
285	if len(data):	×
286	form_id = job.info.get("form")	×
287	form = Forms.objects.filter(pk=int(form_id)).first()	×
288	file = job.info.get("file")	×
289	df = pd.read_excel(f"./tmp/{file}", sheet_name="data")	×
290	error_list = pd.DataFrame(data)	×
291	error_list = error_list[	×
292	list(filter(lambda x: x != "error", list(error_list)))
293	]
294	error_file = f"./tmp/error-{job_id}.csv"	×
295	error_list.to_csv(error_file, index=False)	×
296	data = {	×
297	"send_to": [job.user.email],
298	"listing": [
299	{
300	"name": "Upload Date",
301	"value": job.created.strftime("%m-%d-%Y, %H:%M:%S"),
302	},
303	{"name": "Questionnaire", "value": form.name},
304	{"name": "Number of Records", "value": df.shape[0]},
305	],
306	}
307	send_email(	×
308	context=data,
309	type=EmailTypes.upload_error,
310	path=error_file,
311	content_type="text/csv",
312	)
313	return False	×
314	return True	×
315
316
317	def validate_excel_result(task):	1✔
318	job = Jobs.objects.get(task_id=task.id)	×
319	job.attempt = job.attempt + 1	×
320	job_info = job.info	×
321	if task.result:	×
322	job.status = JobStatus.done	×
323	job.available = timezone.now()	×
324	job.save()	×
325	job_info.update({"ref_job_id": job.id})	×
326	new_job = Jobs.objects.create(	×
327	result=job.info.get("file"),
328	type=JobTypes.seed_data,
329	status=JobStatus.on_progress,
330	user=job.user,
331	info=job_info,
332	)
333	task_id = async_task(	×
334	"api.v1.v1_jobs.job.seed_data_job",
335	new_job.id,
336	hook="api.v1.v1_jobs.job.seed_data_job_result",
337	)
338	new_job.task_id = task_id	×
339	new_job.save()	×
340	else:
341	job.status = JobStatus.failed	×
342	job.save()	×
343
344
345	def handle_administrations_bulk_upload(filename, user_id, upload_time):	1✔
346	user = SystemUser.objects.get(id=user_id)	×
347	storage.download(f"upload/{filename}")	×
348	file_path = f"./tmp/{filename}"	×
349	errors = validate_administrations_bulk_upload(file_path)	×
350	xlsx = pd.ExcelFile(file_path)	×
351	if "data" not in xlsx.sheet_names:	×
352	logger.error(f"Sheet 'data' not found in {filename}")	×
353	send_email(	×
354	context={
355	"send_to": [user.email],
356	"listing": [
357	{
358	"name": "Upload Date",
359	"value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
360	},
361	{
362	"name": "Questionnaire",
363	"value": "Administrative List",
364	},
365	{
366	"name": "Error",
367	"value": 'Sheet "data" not found',
368	},
369	],
370	},
371	type=EmailTypes.upload_error,
372	)
373	return	×
374	df = pd.read_excel(file_path, sheet_name="data")	×
375	email_context = {	×
376	"send_to": [user.email],
377	"listing": [
378	{
379	"name": "Upload Date",
380	"value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
381	},
382	{
383	"name": "Questionnaire",
384	"value": "Administrative List",
385	},
386	{"name": "Number of Records", "value": df.shape[0]},
387	],
388	}
389	if len(errors):	×
390	logger.error(errors)	×
391	error_file = (	×
392	"./tmp/administration-error-"
393	f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
394	)
395	error_list = pd.DataFrame(errors)	×
396	error_list.to_csv(error_file, index=False)	×
397	send_email(	×
398	context=email_context,
399	type=EmailTypes.upload_error,
400	path=error_file,
401	content_type="text/csv",
402	)
403	return	×
404	seed_administration_data(file_path)	×
405	generate_sqlite(Administration)	×
406	send_email(context=email_context, type=EmailTypes.administration_upload)	×
407
408
409	def handle_master_data_bulk_upload_failure(task: Task):	1✔
410	if task.success:	×
411	return	×
412	logger.error(	×
413	{
414	"error": "Failed running background job",
415	"id": task.id,
416	"name": task.name,
417	"started": task.started,
418	"stopped": task.stopped,
419	"args": task.args,
420	"kwargs": task.kwargs,
421	"body": task.result,
422	}
423	)
424
425
426	def handle_entities_error_upload(	1✔
427	errors: list,
428	email_context: dict,
429	user: SystemUser,
430	upload_time: timezone.datetime,
431	):
432	logger.error(errors)	1✔
433	error_file = (	1✔
434	"./tmp/entity-error-"
435	f"{upload_time.strftime('%Y%m%d%H%M%S')}-{user.id}.csv"
436	)
437	error_list = pd.DataFrame(errors)	1✔
438	error_list.to_csv(error_file, index=False)	1✔
439	send_email(	1✔
440	context=email_context,
441	type=EmailTypes.upload_error,
442	path=error_file,
443	content_type="text/csv",
444	)
445
446
447	def handle_entities_bulk_upload(filename, user_id, upload_time):	1✔
448	user = SystemUser.objects.get(id=user_id)	1✔
449	storage.download(f"upload/{filename}")	1✔
450	file_path = f"./tmp/{filename}"	1✔
451	errors = validate_entity_file(file_path)	1✔
452	email_context = {	1✔
453	"send_to": [user.email],
454	"listing": [
455	{
456	"name": "Upload Date",
457	"value": upload_time.strftime("%m-%d-%Y, %H:%M:%S"),
458	},
459	{
460	"name": "Questionnaire",
461	"value": "Entity List",
462	},
463	],
464	}
465	if len(errors):	1!
466	handle_entities_error_upload(errors, email_context, user, upload_time)	1✔
467	return	1✔
468	errors = validate_entity_data(file_path)	×
469	if len(errors):	×
470	handle_entities_error_upload(errors, email_context, user, upload_time)	×
471	return	×
472	generate_sqlite(EntityData)	×

akvo / rtmis / #1839

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous