bika.lims.utils.analysisrequest.create_retest() - Code Metrics - Inspection of "Fix error when invalidating sample with copies of..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( f5d3d8...6a1f51 )

by Ramon

created 2021-11-02 10:36 UTC

bika.lims.utils.analysisrequest.create_retest() D

↳ Parent: bika.lims.utils.analysisrequest

Complexity

Conditions

Size

Total Lines	83
Code Lines	44

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	44
dl	0
loc	83
rs	4.8
c	0
b	0
f	0
cc	12
nop	1

How to fix Long Method Complexity

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2021 by it's authors.
# Some rights reserved, see README and LICENSE.

import itertools
from string import Template

import six
from Products.Archetypes.config import UID_CATALOG
from Products.CMFPlone.utils import _createObjectByType
from Products.CMFPlone.utils import safe_unicode
from senaite.core.workflow import ANALYSIS_WORKFLOW
from senaite.core.workflow import SAMPLE_WORKFLOW
from zope.interface import alsoProvides
from zope.lifecycleevent import modified

from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.api.mail import compose_email
from bika.lims.api.mail import is_valid_email_address
from bika.lims.api.mail import send_email
from bika.lims.catalog import SETUP_CATALOG
from bika.lims.idserver import renameAfterCreation
from bika.lims.interfaces import IAnalysisRequest
from bika.lims.interfaces import IAnalysisRequestRetest
from bika.lims.interfaces import IAnalysisRequestSecondary
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IReceived
from bika.lims.interfaces import IRoutineAnalysis
from bika.lims.utils import changeWorkflowState
from bika.lims.utils import copy_field_values
from bika.lims.utils import createPdf
from bika.lims.utils import get_link
from bika.lims.utils import tmpID
from bika.lims.workflow import ActionHandlerPool
from bika.lims.workflow import doActionFor
from bika.lims.workflow import push_reindex_to_actions_pool
from bika.lims.workflow.analysisrequest import do_action_to_analyses


def create_analysisrequest(client, request, values, analyses=None,
                           results_ranges=None, prices=None):
    """Creates a new AnalysisRequest (a Sample) object
    :param client: The container where the Sample will be created
    :param request: The current Http Request object
    :param values: A dict, with keys as AnalaysisRequest's schema field names
    :param analyses: List of Services or Analyses (brains, objects, UIDs,
        keywords). Extends the list from values["Analyses"]
    :param results_ranges: List of Results Ranges. Extends the results ranges
        from the Specification object defined in values["Specification"]
    :param prices: Mapping of AnalysisService UID -> price. If not set, prices
        are read from the associated analysis service.
    """
    # Don't pollute the dict param passed in
    values = dict(values.items())

    # Resolve the Service uids of analyses to be added in the Sample. Values
    # passed-in might contain Profiles and also values that are not uids. Also,
    # additional analyses can be passed-in through either values or services
    service_uids = to_services_uids(values=values, services=analyses)

    # Remove the Analyses from values. We will add them manually
    values.update({"Analyses": []})

    # Create the Analysis Request and submit the form
    ar = _createObjectByType('AnalysisRequest', client, tmpID())
    ar.processForm(REQUEST=request, values=values)

    # Set the analyses manually
    ar.setAnalyses(service_uids, prices=prices, specs=results_ranges)

    # Handle hidden analyses from template and profiles
    # https://github.com/senaite/senaite.core/issues/1437
    # https://github.com/senaite/senaite.core/issues/1326
    apply_hidden_services(ar)

    # Handle rejection reasons
    rejection_reasons = resolve_rejection_reasons(values)
    ar.setRejectionReasons(rejection_reasons)

    # Handle secondary Analysis Request
    primary = ar.getPrimaryAnalysisRequest()
    if primary:
        # Mark the secondary with the `IAnalysisRequestSecondary` interface
        alsoProvides(ar, IAnalysisRequestSecondary)

        # Rename the secondary according to the ID server setup
        renameAfterCreation(ar)

        # Set dates to match with those from the primary
        ar.setDateSampled(primary.getDateSampled())
        ar.setSamplingDate(primary.getSamplingDate())
        ar.setDateReceived(primary.getDateReceived())

        # Force the transition of the secondary to received and set the
        # description/comment in the transition accordingly.
        if primary.getDateReceived():
            primary_id = primary.getId()
            comment = "Auto-received. Secondary Sample of {}".format(primary_id)
            changeWorkflowState(ar, SAMPLE_WORKFLOW, "sample_received",
                                action="receive", comments=comment)

            # Mark the secondary as received
            alsoProvides(ar, IReceived)

            # Initialize analyses
            do_action_to_analyses(ar, "initialize")

            # Notify the ar has ben modified
            modified(ar)

            # Reindex the AR
            ar.reindexObject()

            # If rejection reasons have been set, reject automatically
            if rejection_reasons:
                do_rejection(ar)

            # In "received" state already
            return ar

    # Try first with no sampling transition, cause it is the most common config
    success, message = doActionFor(ar, "no_sampling_workflow")
    if not success:
        doActionFor(ar, "to_be_sampled")

    # If rejection reasons have been set, reject the sample automatically
    if rejection_reasons:
        do_rejection(ar)

    return ar


def apply_hidden_services(sample):
    """
    Applies the hidden setting to the sample analyses in accordance with the
    settings from its template and/or profiles
    :param sample: the sample that contains the analyses
    """
    hidden = list()

    # Get the "hidden" service uids from the template
    template = sample.getTemplate()
    hidden = get_hidden_service_uids(template)

    # Get the "hidden" service uids from profiles
    profiles = sample.getProfiles()
    hid_profiles = map(get_hidden_service_uids, profiles)
    hid_profiles = list(itertools.chain(*hid_profiles))
    hidden.extend(hid_profiles)

    # Update the sample analyses
    analyses = sample.getAnalyses(full_objects=True)
    analyses = filter(lambda an: an.getServiceUID() in hidden, analyses)
    for analysis in analyses:
        analysis.setHidden(True)


def get_hidden_service_uids(profile_or_template):
    """Returns a list of service uids that are set as hidden
    :param profile_or_template: ARTemplate or AnalysisProfile object
    """
    if not profile_or_template:
        return []
    settings = profile_or_template.getAnalysisServicesSettings()
    hidden = filter(lambda ser: ser.get("hidden", False), settings)
    return map(lambda setting: setting["uid"], hidden)


def to_services_uids(services=None, values=None):
    """
    Returns a list of Analysis Services uids
    :param services: A list of service items (uid, keyword, brain, obj, title)
    :param values: a dict, where keys are AR|Sample schema field names.
    :returns: a list of Analyses Services UIDs
    """
    def to_list(value):
        if not value:
            return []
        if isinstance(value, six.string_types):
            return [value]
        if isinstance(value, (list, tuple)):
            return value
        logger.warn("Cannot convert to a list: {}".format(value))
        return []

    services = services or []
    values = values or {}

    # Merge analyses from analyses_serv and values into one list
    uids = to_list(services) + to_list(values.get("Analyses"))

    # Convert them to a list of service uids
    uids = filter(None, map(to_service_uid, uids))

    # Extend with service uids from profiles
    profiles = to_list(values.get("Profiles"))
    if profiles:
        uid_catalog = api.get_tool(UID_CATALOG)
        for brain in uid_catalog(UID=profiles):
            profile = api.get_object(brain)
            uids.extend(profile.getRawService() or [])

    # Get the service uids without duplicates, but preserving the order
    return list(dict.fromkeys(uids).keys())


def to_service_uid(uid_brain_obj_str):
    """Resolves the passed in element to a valid uid. Returns None if the value
    cannot be resolved to a valid uid
    """
    if api.is_uid(uid_brain_obj_str) and uid_brain_obj_str != "0":
        return uid_brain_obj_str

    if api.is_object(uid_brain_obj_str):
        obj = api.get_object(uid_brain_obj_str)

        if IAnalysisService.providedBy(obj):
            return api.get_uid(obj)

        elif IRoutineAnalysis.providedBy(obj):
            return obj.getServiceUID()

        else:
            logger.error("Type not supported: {}".format(obj.portal_type))
            return None

    if isinstance(uid_brain_obj_str, six.string_types):
        # Maybe is a keyword?
        query = dict(portal_type="AnalysisService", getKeyword=uid_brain_obj_str)
        brains = api.search(query, SETUP_CATALOG)
        if len(brains) == 1:
            return api.get_uid(brains[0])

        # Or maybe a title
        query = dict(portal_type="AnalysisService", title=uid_brain_obj_str)
        brains = api.search(query, SETUP_CATALOG)
        if len(brains) == 1:
            return api.get_uid(brains[0])

    return None


def create_retest(ar):
    """Creates a retest (Analysis Request) from an invalidated Analysis Request
    :param ar: The invalidated Analysis Request
    :type ar: IAnalysisRequest
    :rtype: IAnalysisRequest
    """
    if not ar:
        raise ValueError("Source Analysis Request cannot be None")

    if not IAnalysisRequest.providedBy(ar):
        raise ValueError("Type not supported: {}".format(repr(type(ar))))

    if ar.getRetest():
        # Do not allow the creation of another retest!
        raise ValueError("Retest already set")

    if not ar.isInvalid():
        # Analysis Request must be in 'invalid' state
        raise ValueError("Cannot do a retest from an invalid Analysis Request"
                         .format(repr(ar)))

    # Open the actions pool
    actions_pool = ActionHandlerPool.get_instance()
    actions_pool.queue_pool()

    # Create the Retest (Analysis Request)
    ignore = ['Analyses', 'DatePublished', 'Invalidated', 'Sample']
    retest = _createObjectByType("AnalysisRequest", ar.aq_parent, tmpID())
    copy_field_values(ar, retest, ignore_fieldnames=ignore)

    # Mark the retest with the `IAnalysisRequestRetest` interface
    alsoProvides(retest, IAnalysisRequestRetest)

    # Assign the source to retest
    retest.setInvalidated(ar)

    # Rename the retest according to the ID server setup
    renameAfterCreation(retest)

    # Copy the analyses from the source
    intermediate_states = ['retracted',]
    for an in ar.getAnalyses(full_objects=True):
        # skip retests
        if an.isRetest():
            continue

        if api.get_workflow_status_of(an) in intermediate_states:
            # Exclude intermediate analyses
            continue

        # Original sample might have multiple copies of same analysis
        keyword = an.getKeyword()
        analyses = retest.getAnalyses(full_objects=True)
        analyses = filter(lambda ret: ret.getKeyword() == keyword, analyses)

        if analyses:
            keyword = '{}-{}'.format(keyword, len(analyses))

        # Create the analysis retest
        nan = _createObjectByType("Analysis", retest, keyword)

        # Make a copy
        ignore_fieldnames = ['DataAnalysisPublished']
        copy_field_values(an, nan, ignore_fieldnames=ignore_fieldnames)
        nan.unmarkCreationFlag()
        push_reindex_to_actions_pool(nan)

    # Transition the retest to "sample_received"!
    changeWorkflowState(retest, SAMPLE_WORKFLOW, 'sample_received')
    alsoProvides(retest, IReceived)

    # Initialize analyses
    for analysis in retest.getAnalyses(full_objects=True):
        if not IRoutineAnalysis.providedBy(analysis):
            continue
        changeWorkflowState(analysis, ANALYSIS_WORKFLOW, "unassigned")

    # Reindex and other stuff
    push_reindex_to_actions_pool(retest)
    push_reindex_to_actions_pool(retest.aq_parent)

    # Resume the actions pool
    actions_pool.resume()
    return retest


def create_partition(analysis_request, request, analyses, sample_type=None,
                     container=None, preservation=None, skip_fields=None,
                     internal_use=True):
    """
    Creates a partition for the analysis_request (primary) passed in
    :param analysis_request: uid/brain/object of IAnalysisRequest type
    :param request: the current request object
    :param analyses: uids/brains/objects of IAnalysis type
    :param sampletype: uid/brain/object of SampleType
    :param container: uid/brain/object of Container
    :param preservation: uid/brain/object of Preservation
    :param skip_fields: names of fields to be skipped on copy from primary
    :return: the new partition
    """
    partition_skip_fields = [
        "Analyses",
        "Attachment",
        "Client",
        "DetachedFrom",
        "Profile",
        "Profiles",
        "RejectionReasons",
        "Remarks",
        "ResultsInterpretation",
        "ResultsInterpretationDepts",
        "Sample",
        "Template",
        "creation_date",
        "modification_date",
        "ParentAnalysisRequest",
        "PrimaryAnalysisRequest",
        # default fields
        "id",
        "description",
        "allowDiscussion",
        "subject",
        "location",
        "contributors",
        "creators",
        "effectiveDate",
        "expirationDate",
        "language",
        "rights",
        "creation_date",
        "modification_date",
    ]
    if skip_fields:
        partition_skip_fields.extend(skip_fields)
        partition_skip_fields = list(set(partition_skip_fields))

    # Copy field values from the primary analysis request
    ar = api.get_object(analysis_request)
    record = fields_to_dict(ar, partition_skip_fields)

    # Update with values that are partition-specific
    record.update({
        "InternalUse": internal_use,
        "ParentAnalysisRequest": api.get_uid(ar),
    })
    if sample_type is not None:
        record["SampleType"] = sample_type and api.get_uid(sample_type) or ""
    if container is not None:
        record["Container"] = container and api.get_uid(container) or ""
    if preservation is not None:
        record["Preservation"] = preservation and api.get_uid(preservation) or ""

    # Create the Partition
    client = ar.getClient()
    analyses = list(set(map(api.get_object, analyses)))
    services = map(lambda an: an.getAnalysisService(), analyses)

    # Populate the root's ResultsRanges to partitions
    results_ranges = ar.getResultsRange() or []
    partition = create_analysisrequest(client,
                                       request=request,
                                       values=record,
                                       analyses=services,
                                       results_ranges=results_ranges)

    # Reindex Parent Analysis Request
    ar.reindexObject(idxs=["isRootAncestor"])

    # Manually set the Date Received to match with its parent. This is
    # necessary because crar calls to processForm, so DateReceived is not
    # set because the partition has not been received yet
    partition.setDateReceived(ar.getDateReceived())
    partition.reindexObject(idxs="getDateReceived")

    # Always set partition to received state
    changeWorkflowState(partition, SAMPLE_WORKFLOW, "sample_received")
    alsoProvides(partition, IReceived)

    # And initialize the analyses the partition contains. This is required
    # here because the transition "initialize" of analyses rely on a guard,
    # so the initialization can only be performed when the sample has been
    # received (DateReceived is set)
    for analysis in partition.getAnalyses(full_objects=True):
        doActionFor(analysis, "initialize")
        analysis.reindexObject()
    return partition


def fields_to_dict(obj, skip_fields=None):
    """
    Generates a dictionary with the field values of the object passed in, where
    keys are the field names. Skips computed fields
    """
    data = {}
    obj = api.get_object(obj)
    for field_name, field in api.get_fields(obj).items():
        if skip_fields and field_name in skip_fields:
            continue
        if field.type == "computed":
            continue
        data[field_name] = field.get(obj)
    return data


def resolve_rejection_reasons(values):
    """Resolves the rejection reasons from the submitted values to the format
    supported by Sample's Rejection Reason field
    """
    rejection_reasons = values.get("RejectionReasons")
    if not rejection_reasons:
        return []

    # Predefined reasons selected?
    selected = rejection_reasons[0] or {}
    if selected.get("checkbox") == "on":
        selected = selected.get("multiselection") or []
    else:
        selected = []

    # Other reasons set?
    other = values.get("RejectionReasons.textfield")
    if other:
        other = other[0] or {}
        other = other.get("other", "")
    else:
        other = ""

    # If neither selected nor other reasons are set, return empty
    if any([selected, other]):
        return [{"selected": selected, "other": other}]

    return []


def do_rejection(sample, notify=None):
    """Rejects the sample and if succeeds, generates the rejection pdf and
    sends a notification email. If notify is None, the notification email will
    only be sent if the setting in Setup is enabled
    """
    sample_id = api.get_id(sample)
    if not sample.getRejectionReasons():
        logger.warn("Cannot reject {} w/o rejection reasons".format(sample_id))
        return

    success, msg = doActionFor(sample, "reject")
    if not success:
        logger.warn("Cannot reject the sample {}".format(sample_id))
        return

    # Generate a pdf with the rejection reasons
    pdf = get_rejection_pdf(sample)

    # Attach the PDF to the sample
    filename = "{}-rejected.pdf".format(sample_id)
    attachment = sample.createAttachment(pdf, filename=filename)
    pdf_file = attachment.getAttachmentFile()

    # Do we need to send a notification email?
    if notify is None:
        setup = api.get_setup()
        notify = setup.getNotifyOnSampleRejection()

    if notify:
        # Compose and send the email
        mime_msg = get_rejection_mail(sample, pdf_file)
        if mime_msg:
            # Send the email
            send_email(mime_msg)


def get_rejection_pdf(sample):
    """Generates a pdf with sample rejection reasons
    """
    # Avoid circular dependencies
    from bika.lims.browser.analysisrequest.reject import \
        AnalysisRequestRejectPdfView

    # Render the html's rejection document
    tpl = AnalysisRequestRejectPdfView(sample, api.get_request())
    html = tpl.template()
    html = safe_unicode(html).encode("utf-8")

    # Generate the pdf
    return createPdf(htmlreport=html)


def get_rejection_mail(sample, rejection_pdf=None):
    """Generates an email to sample contacts with rejection reasons
    """
    # Get the reasons
    reasons = sample.getRejectionReasons()
    reasons = reasons and reasons[0] or {}
    reasons = reasons.get("selected", []) + [reasons.get("other")]
    reasons = filter(None, reasons)
    reasons = "<br/>- ".join(reasons)

    # Render the email body
    setup = api.get_setup()
    lab_address = setup.laboratory.getPrintAddress()
    email_body = Template(setup.getEmailBodySampleRejection())
    email_body = email_body.safe_substitute({
        "lab_address": "<br/>".join(lab_address),
        "reasons": reasons and "<br/>-{}".format(reasons) or "",
        "sample_id": api.get_id(sample),
        "sample_link": get_link(api.get_url(sample), api.get_id(sample))
    })

    def to_valid_email_address(contact):
        if not contact:
            return None
        address = contact.getEmailAddress()
        if not is_valid_email_address(address):
            return None
        return address

    # Get the recipients
    _to = [sample.getContact()] + sample.getCCContact()
    _to = map(to_valid_email_address, _to)
    _to = filter(None, _to)

    if not _to:
        # Cannot send an e-mail without recipient!
        logger.warn("No valid recipients for {}".format(api.get_id(sample)))
        return None

    lab = api.get_setup().laboratory
    attachments = rejection_pdf and [rejection_pdf] or []

    return compose_email(
        from_addr=lab.getEmailAddress(),
        to_addr=_to,
        subj=_("%s has been rejected") % api.get_id(sample),
        body=email_body,
        attachments=attachments)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2021 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import itertools
22			from string import Template
23
24			import six
25			from Products.Archetypes.config import UID_CATALOG
26			from Products.CMFPlone.utils import _createObjectByType
27			from Products.CMFPlone.utils import safe_unicode
28			from senaite.core.workflow import ANALYSIS_WORKFLOW
29			from senaite.core.workflow import SAMPLE_WORKFLOW
30			from zope.interface import alsoProvides
31			from zope.lifecycleevent import modified
32
33			from bika.lims import api
34			from bika.lims import bikaMessageFactory as _
35			from bika.lims import logger
36			from bika.lims.api.mail import compose_email
37			from bika.lims.api.mail import is_valid_email_address
38			from bika.lims.api.mail import send_email
39			from bika.lims.catalog import SETUP_CATALOG
40			from bika.lims.idserver import renameAfterCreation
41			from bika.lims.interfaces import IAnalysisRequest
42			from bika.lims.interfaces import IAnalysisRequestRetest
43			from bika.lims.interfaces import IAnalysisRequestSecondary
44			from bika.lims.interfaces import IAnalysisService
45			from bika.lims.interfaces import IReceived
46			from bika.lims.interfaces import IRoutineAnalysis
47			from bika.lims.utils import changeWorkflowState
48			from bika.lims.utils import copy_field_values
49			from bika.lims.utils import createPdf
50			from bika.lims.utils import get_link
51			from bika.lims.utils import tmpID
52			from bika.lims.workflow import ActionHandlerPool
53			from bika.lims.workflow import doActionFor
54			from bika.lims.workflow import push_reindex_to_actions_pool
55			from bika.lims.workflow.analysisrequest import do_action_to_analyses
56
57
58			def create_analysisrequest(client, request, values, analyses=None,
59			results_ranges=None, prices=None):
60			"""Creates a new AnalysisRequest (a Sample) object
61			:param client: The container where the Sample will be created
62			:param request: The current Http Request object
63			:param values: A dict, with keys as AnalaysisRequest's schema field names
64			:param analyses: List of Services or Analyses (brains, objects, UIDs,
65			keywords). Extends the list from values["Analyses"]
66			:param results_ranges: List of Results Ranges. Extends the results ranges
67			from the Specification object defined in values["Specification"]
68			:param prices: Mapping of AnalysisService UID -> price. If not set, prices
69			are read from the associated analysis service.
70			"""
71			# Don't pollute the dict param passed in
72			values = dict(values.items())
73
74			# Resolve the Service uids of analyses to be added in the Sample. Values
75			# passed-in might contain Profiles and also values that are not uids. Also,
76			# additional analyses can be passed-in through either values or services
77			service_uids = to_services_uids(values=values, services=analyses)
78
79			# Remove the Analyses from values. We will add them manually
80			values.update({"Analyses": []})
81
82			# Create the Analysis Request and submit the form
83			ar = _createObjectByType('AnalysisRequest', client, tmpID())
84			ar.processForm(REQUEST=request, values=values)
85
86			# Set the analyses manually
87			ar.setAnalyses(service_uids, prices=prices, specs=results_ranges)
88
89			# Handle hidden analyses from template and profiles
90			# https://github.com/senaite/senaite.core/issues/1437
91			# https://github.com/senaite/senaite.core/issues/1326
92			apply_hidden_services(ar)
93
94			# Handle rejection reasons
95			rejection_reasons = resolve_rejection_reasons(values)
96			ar.setRejectionReasons(rejection_reasons)
97
98			# Handle secondary Analysis Request
99			primary = ar.getPrimaryAnalysisRequest()
100			if primary:
101			# Mark the secondary with the `IAnalysisRequestSecondary` interface
102			alsoProvides(ar, IAnalysisRequestSecondary)
103
104			# Rename the secondary according to the ID server setup
105			renameAfterCreation(ar)
106
107			# Set dates to match with those from the primary
108			ar.setDateSampled(primary.getDateSampled())
109			ar.setSamplingDate(primary.getSamplingDate())
110			ar.setDateReceived(primary.getDateReceived())
111
112			# Force the transition of the secondary to received and set the
113			# description/comment in the transition accordingly.
114			if primary.getDateReceived():
115			primary_id = primary.getId()
116			comment = "Auto-received. Secondary Sample of {}".format(primary_id)
117			changeWorkflowState(ar, SAMPLE_WORKFLOW, "sample_received",
118			action="receive", comments=comment)
119
120			# Mark the secondary as received
121			alsoProvides(ar, IReceived)
122
123			# Initialize analyses
124			do_action_to_analyses(ar, "initialize")
125
126			# Notify the ar has ben modified
127			modified(ar)
128
129			# Reindex the AR
130			ar.reindexObject()
131
132			# If rejection reasons have been set, reject automatically
133			if rejection_reasons:
134			do_rejection(ar)
135
136			# In "received" state already
137			return ar
138
139			# Try first with no sampling transition, cause it is the most common config
140			success, message = doActionFor(ar, "no_sampling_workflow")
141			if not success:
142			doActionFor(ar, "to_be_sampled")
143
144			# If rejection reasons have been set, reject the sample automatically
145			if rejection_reasons:
146			do_rejection(ar)
147
148			return ar
149
150
151			def apply_hidden_services(sample):
152			"""
153			Applies the hidden setting to the sample analyses in accordance with the
154			settings from its template and/or profiles
155			:param sample: the sample that contains the analyses
156			"""
157			hidden = list()
158
159			# Get the "hidden" service uids from the template
160			template = sample.getTemplate()
161			hidden = get_hidden_service_uids(template)
162
163			# Get the "hidden" service uids from profiles
164			profiles = sample.getProfiles()
165			hid_profiles = map(get_hidden_service_uids, profiles)
166			hid_profiles = list(itertools.chain(*hid_profiles))
167			hidden.extend(hid_profiles)
168
169			# Update the sample analyses
170			analyses = sample.getAnalyses(full_objects=True)
171			analyses = filter(lambda an: an.getServiceUID() in hidden, analyses)
172			for analysis in analyses:
173			analysis.setHidden(True)
174
175
176			def get_hidden_service_uids(profile_or_template):
177			"""Returns a list of service uids that are set as hidden
178			:param profile_or_template: ARTemplate or AnalysisProfile object
179			"""
180			if not profile_or_template:
181			return []
182			settings = profile_or_template.getAnalysisServicesSettings()
183			hidden = filter(lambda ser: ser.get("hidden", False), settings)
184			return map(lambda setting: setting["uid"], hidden)
185
186
187			def to_services_uids(services=None, values=None):
188			"""
189			Returns a list of Analysis Services uids
190			:param services: A list of service items (uid, keyword, brain, obj, title)
191			:param values: a dict, where keys are AR\|Sample schema field names.
192			:returns: a list of Analyses Services UIDs
193			"""
194			def to_list(value):
195			if not value:
196			return []
197			if isinstance(value, six.string_types):
198			return [value]
199			if isinstance(value, (list, tuple)):
200			return value
201			logger.warn("Cannot convert to a list: {}".format(value))
202			return []
203
204			services = services or []
205			values = values or {}
206
207			# Merge analyses from analyses_serv and values into one list
208			uids = to_list(services) + to_list(values.get("Analyses"))
209
210			# Convert them to a list of service uids
211			uids = filter(None, map(to_service_uid, uids))
212
213			# Extend with service uids from profiles
214			profiles = to_list(values.get("Profiles"))
215			if profiles:
216			uid_catalog = api.get_tool(UID_CATALOG)
217			for brain in uid_catalog(UID=profiles):
218			profile = api.get_object(brain)
219			uids.extend(profile.getRawService() or [])
220
221			# Get the service uids without duplicates, but preserving the order
222			return list(dict.fromkeys(uids).keys())
223
224
225			def to_service_uid(uid_brain_obj_str):
226			"""Resolves the passed in element to a valid uid. Returns None if the value
227			cannot be resolved to a valid uid
228			"""
229			if api.is_uid(uid_brain_obj_str) and uid_brain_obj_str != "0":
230			return uid_brain_obj_str
231
232			if api.is_object(uid_brain_obj_str):
233			obj = api.get_object(uid_brain_obj_str)
234
235			if IAnalysisService.providedBy(obj):
236			return api.get_uid(obj)
237
238			elif IRoutineAnalysis.providedBy(obj):
239			return obj.getServiceUID()
240
241			else:
242			logger.error("Type not supported: {}".format(obj.portal_type))
243			return None
244
245			if isinstance(uid_brain_obj_str, six.string_types):
246			# Maybe is a keyword?
247			query = dict(portal_type="AnalysisService", getKeyword=uid_brain_obj_str)
248			brains = api.search(query, SETUP_CATALOG)
249			if len(brains) == 1:
250			return api.get_uid(brains[0])
251
252			# Or maybe a title
253			query = dict(portal_type="AnalysisService", title=uid_brain_obj_str)
254			brains = api.search(query, SETUP_CATALOG)
255			if len(brains) == 1:
256			return api.get_uid(brains[0])
257
258			return None
259
260
261			def create_retest(ar):
262			"""Creates a retest (Analysis Request) from an invalidated Analysis Request
263			:param ar: The invalidated Analysis Request
264			:type ar: IAnalysisRequest
265			:rtype: IAnalysisRequest
266			"""
267			if not ar:
268			raise ValueError("Source Analysis Request cannot be None")
269
270			if not IAnalysisRequest.providedBy(ar):
271			raise ValueError("Type not supported: {}".format(repr(type(ar))))
272
273			if ar.getRetest():
274			# Do not allow the creation of another retest!
275			raise ValueError("Retest already set")
276
277			if not ar.isInvalid():
278			# Analysis Request must be in 'invalid' state
279			raise ValueError("Cannot do a retest from an invalid Analysis Request"
280			.format(repr(ar)))
281
282			# Open the actions pool
283			actions_pool = ActionHandlerPool.get_instance()
284			actions_pool.queue_pool()
285
286			# Create the Retest (Analysis Request)
287			ignore = ['Analyses', 'DatePublished', 'Invalidated', 'Sample']
288			retest = _createObjectByType("AnalysisRequest", ar.aq_parent, tmpID())
289			copy_field_values(ar, retest, ignore_fieldnames=ignore)
290
291			# Mark the retest with the `IAnalysisRequestRetest` interface
292			alsoProvides(retest, IAnalysisRequestRetest)
293
294			# Assign the source to retest
295			retest.setInvalidated(ar)
296
297			# Rename the retest according to the ID server setup
298			renameAfterCreation(retest)
299
300			# Copy the analyses from the source
301			intermediate_states = ['retracted',]
302			for an in ar.getAnalyses(full_objects=True):
303			# skip retests
304			if an.isRetest():
305			continue
306
307			if api.get_workflow_status_of(an) in intermediate_states:
308			# Exclude intermediate analyses
309			continue
310
311			# Original sample might have multiple copies of same analysis
312			keyword = an.getKeyword()
313			analyses = retest.getAnalyses(full_objects=True)
314			analyses = filter(lambda ret: ret.getKeyword() == keyword, analyses)
			0 ignored issues – show introduced 2021-11-02 09:59 UTC by Report Bug Copy Issue Report The variable `keyword` does not seem to be defined for all execution paths. Loading history...
315			if analyses:
316			keyword = '{}-{}'.format(keyword, len(analyses))
317
318			# Create the analysis retest
319			nan = _createObjectByType("Analysis", retest, keyword)
320
321			# Make a copy
322			ignore_fieldnames = ['DataAnalysisPublished']
323			copy_field_values(an, nan, ignore_fieldnames=ignore_fieldnames)
324			nan.unmarkCreationFlag()
325			push_reindex_to_actions_pool(nan)
326
327			# Transition the retest to "sample_received"!
328			changeWorkflowState(retest, SAMPLE_WORKFLOW, 'sample_received')
329			alsoProvides(retest, IReceived)
330
331			# Initialize analyses
332			for analysis in retest.getAnalyses(full_objects=True):
333			if not IRoutineAnalysis.providedBy(analysis):
334			continue
335			changeWorkflowState(analysis, ANALYSIS_WORKFLOW, "unassigned")
336
337			# Reindex and other stuff
338			push_reindex_to_actions_pool(retest)
339			push_reindex_to_actions_pool(retest.aq_parent)
340
341			# Resume the actions pool
342			actions_pool.resume()
343			return retest
344
345
346			def create_partition(analysis_request, request, analyses, sample_type=None,
347			container=None, preservation=None, skip_fields=None,
348			internal_use=True):
349			"""
350			Creates a partition for the analysis_request (primary) passed in
351			:param analysis_request: uid/brain/object of IAnalysisRequest type
352			:param request: the current request object
353			:param analyses: uids/brains/objects of IAnalysis type
354			:param sampletype: uid/brain/object of SampleType
355			:param container: uid/brain/object of Container
356			:param preservation: uid/brain/object of Preservation
357			:param skip_fields: names of fields to be skipped on copy from primary
358			:return: the new partition
359			"""
360			partition_skip_fields = [
361			"Analyses",
362			"Attachment",
363			"Client",
364			"DetachedFrom",
365			"Profile",
366			"Profiles",
367			"RejectionReasons",
368			"Remarks",
369			"ResultsInterpretation",
370			"ResultsInterpretationDepts",
371			"Sample",
372			"Template",
373			"creation_date",
374			"modification_date",
375			"ParentAnalysisRequest",
376			"PrimaryAnalysisRequest",
377			# default fields
378			"id",
379			"description",
380			"allowDiscussion",
381			"subject",
382			"location",
383			"contributors",
384			"creators",
385			"effectiveDate",
386			"expirationDate",
387			"language",
388			"rights",
389			"creation_date",
390			"modification_date",
391			]
392			if skip_fields:
393			partition_skip_fields.extend(skip_fields)
394			partition_skip_fields = list(set(partition_skip_fields))
395
396			# Copy field values from the primary analysis request
397			ar = api.get_object(analysis_request)
398			record = fields_to_dict(ar, partition_skip_fields)
399
400			# Update with values that are partition-specific
401			record.update({
402			"InternalUse": internal_use,
403			"ParentAnalysisRequest": api.get_uid(ar),
404			})
405			if sample_type is not None:
406			record["SampleType"] = sample_type and api.get_uid(sample_type) or ""
407			if container is not None:
408			record["Container"] = container and api.get_uid(container) or ""
409			if preservation is not None:
410			record["Preservation"] = preservation and api.get_uid(preservation) or ""
411
412			# Create the Partition
413			client = ar.getClient()
414			analyses = list(set(map(api.get_object, analyses)))
415			services = map(lambda an: an.getAnalysisService(), analyses)
416
417			# Populate the root's ResultsRanges to partitions
418			results_ranges = ar.getResultsRange() or []
419			partition = create_analysisrequest(client,
420			request=request,
421			values=record,
422			analyses=services,
423			results_ranges=results_ranges)
424
425			# Reindex Parent Analysis Request
426			ar.reindexObject(idxs=["isRootAncestor"])
427
428			# Manually set the Date Received to match with its parent. This is
429			# necessary because crar calls to processForm, so DateReceived is not
430			# set because the partition has not been received yet
431			partition.setDateReceived(ar.getDateReceived())
432			partition.reindexObject(idxs="getDateReceived")
433
434			# Always set partition to received state
435			changeWorkflowState(partition, SAMPLE_WORKFLOW, "sample_received")
436			alsoProvides(partition, IReceived)
437
438			# And initialize the analyses the partition contains. This is required
439			# here because the transition "initialize" of analyses rely on a guard,
440			# so the initialization can only be performed when the sample has been
441			# received (DateReceived is set)
442			for analysis in partition.getAnalyses(full_objects=True):
443			doActionFor(analysis, "initialize")
444			analysis.reindexObject()
445			return partition
446
447
448			def fields_to_dict(obj, skip_fields=None):
449			"""
450			Generates a dictionary with the field values of the object passed in, where
451			keys are the field names. Skips computed fields
452			"""
453			data = {}
454			obj = api.get_object(obj)
455			for field_name, field in api.get_fields(obj).items():
456			if skip_fields and field_name in skip_fields:
457			continue
458			if field.type == "computed":
459			continue
460			data[field_name] = field.get(obj)
461			return data
462
463
464			def resolve_rejection_reasons(values):
465			"""Resolves the rejection reasons from the submitted values to the format
466			supported by Sample's Rejection Reason field
467			"""
468			rejection_reasons = values.get("RejectionReasons")
469			if not rejection_reasons:
470			return []
471
472			# Predefined reasons selected?
473			selected = rejection_reasons[0] or {}
474			if selected.get("checkbox") == "on":
475			selected = selected.get("multiselection") or []
476			else:
477			selected = []
478
479			# Other reasons set?
480			other = values.get("RejectionReasons.textfield")
481			if other:
482			other = other[0] or {}
483			other = other.get("other", "")
484			else:
485			other = ""
486
487			# If neither selected nor other reasons are set, return empty
488			if any([selected, other]):
489			return [{"selected": selected, "other": other}]
490
491			return []
492
493
494			def do_rejection(sample, notify=None):
495			"""Rejects the sample and if succeeds, generates the rejection pdf and
496			sends a notification email. If notify is None, the notification email will
497			only be sent if the setting in Setup is enabled
498			"""
499			sample_id = api.get_id(sample)
500			if not sample.getRejectionReasons():
501			logger.warn("Cannot reject {} w/o rejection reasons".format(sample_id))
502			return
503
504			success, msg = doActionFor(sample, "reject")
505			if not success:
506			logger.warn("Cannot reject the sample {}".format(sample_id))
507			return
508
509			# Generate a pdf with the rejection reasons
510			pdf = get_rejection_pdf(sample)
511
512			# Attach the PDF to the sample
513			filename = "{}-rejected.pdf".format(sample_id)
514			attachment = sample.createAttachment(pdf, filename=filename)
515			pdf_file = attachment.getAttachmentFile()
516
517			# Do we need to send a notification email?
518			if notify is None:
519			setup = api.get_setup()
520			notify = setup.getNotifyOnSampleRejection()
521
522			if notify:
523			# Compose and send the email
524			mime_msg = get_rejection_mail(sample, pdf_file)
525			if mime_msg:
526			# Send the email
527			send_email(mime_msg)
528
529
530			def get_rejection_pdf(sample):
531			"""Generates a pdf with sample rejection reasons
532			"""
533			# Avoid circular dependencies
534			from bika.lims.browser.analysisrequest.reject import \
535			AnalysisRequestRejectPdfView
536
537			# Render the html's rejection document
538			tpl = AnalysisRequestRejectPdfView(sample, api.get_request())
539			html = tpl.template()
540			html = safe_unicode(html).encode("utf-8")
541
542			# Generate the pdf
543			return createPdf(htmlreport=html)
544
545
546			def get_rejection_mail(sample, rejection_pdf=None):
547			"""Generates an email to sample contacts with rejection reasons
548			"""
549			# Get the reasons
550			reasons = sample.getRejectionReasons()
551			reasons = reasons and reasons[0] or {}
552			reasons = reasons.get("selected", []) + [reasons.get("other")]
553			reasons = filter(None, reasons)
554			reasons = "<br/>- ".join(reasons)
555
556			# Render the email body
557			setup = api.get_setup()
558			lab_address = setup.laboratory.getPrintAddress()
559			email_body = Template(setup.getEmailBodySampleRejection())
560			email_body = email_body.safe_substitute({
561			"lab_address": "<br/>".join(lab_address),
562			"reasons": reasons and "<br/>-{}".format(reasons) or "",
563			"sample_id": api.get_id(sample),
564			"sample_link": get_link(api.get_url(sample), api.get_id(sample))
565			})
566
567			def to_valid_email_address(contact):
568			if not contact:
569			return None
570			address = contact.getEmailAddress()
571			if not is_valid_email_address(address):
572			return None
573			return address
574
575			# Get the recipients
576			_to = [sample.getContact()] + sample.getCCContact()
577			_to = map(to_valid_email_address, _to)
578			_to = filter(None, _to)
579
580			if not _to:
581			# Cannot send an e-mail without recipient!
582			logger.warn("No valid recipients for {}".format(api.get_id(sample)))
583			return None
584
585			lab = api.get_setup().laboratory
586			attachments = rejection_pdf and [rejection_pdf] or []
587
588			return compose_email(
589			from_addr=lab.getEmailAddress(),
590			to_addr=_to,
591			subj=_("%s has been rejected") % api.get_id(sample),
592			body=email_body,
593			attachments=attachments)
594

senaite / senaite.core

Push — 2.x ( f5d3d8...6a1f51 )

bika.lims.utils.analysisrequest.create_retest() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like