bika.lims.browser.analysisrequest.workflow.AnalysisRequestWorkflowAction.cloneAR() - Code Metrics - Inspection of "Refactoring of `retract_ar` transition and related..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8d2e48...36dfc2 )

by Ramon

created 2018-09-24 20:02 UTC

bika.lims.browser.analysisrequest.workflow.AnalysisRequestWorkflowAction.cloneAR() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	43
Code Lines	35

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	35
dl	0
loc	43
rs	8.1066
c	0
b	0
f	0
cc	6
nop	2

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from string import Template

import plone
from DateTime import DateTime
from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.utils import _createObjectByType, safe_unicode
from bika.lims import PMF, api
from bika.lims import bikaMessageFactory as _
from bika.lims import interfaces
from bika.lims.browser.analyses.workflow import AnalysesWorkflowAction
from bika.lims.browser.bika_listing import WorkflowAction
from bika.lims.content.analysisspec import ResultsRangeDict
from bika.lims.permissions import *
from bika.lims.utils import changeWorkflowState
from bika.lims.utils import encode_header
from bika.lims.utils import isActive
from bika.lims.utils import t
from bika.lims.utils import tmpID
from bika.lims.workflow import getCurrentState
from bika.lims.workflow import wasTransitionPerformed
from email.Utils import formataddr


class AnalysisRequestWorkflowAction(AnalysesWorkflowAction):

    """Workflow actions taken in AnalysisRequest context.

        Sample context workflow actions also redirect here
        Applies to
            Analysis objects
            SamplePartition objects
    """

    def __call__(self):
        form = self.request.form
        plone.protect.CheckAuthenticator(form)
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        if type(action) in (list, tuple):
            action = action[0]
        if type(came_from) in (list, tuple):
            came_from = came_from[0]
        # Call out to the workflow action method
        # Use default bika_listing.py/WorkflowAction for other transitions
        method_name = 'workflow_action_' + action if action else ''
        method = getattr(self, method_name, False)
        if method:
            method()
        else:
            WorkflowAction.__call__(self)

    def notify_ar_retract(self, ar, newar):
        bika_setup = api.get_bika_setup()
        laboratory = bika_setup.laboratory
        lab_address = "<br/>".join(laboratory.getPrintAddress())
        mime_msg = MIMEMultipart('related')
        mime_msg['Subject'] = t(_("Erroneus result publication from ${request_id}",
                                mapping={"request_id": ar.getId()}))
        mime_msg['From'] = formataddr(
            (encode_header(laboratory.getName()),
             laboratory.getEmailAddress()))
        to = []
        contact = ar.getContact()
        if contact:
            to.append(formataddr((encode_header(contact.Title()),
                                  contact.getEmailAddress())))
        for cc in ar.getCCContact():
            formatted = formataddr((encode_header(cc.Title()),
                                   cc.getEmailAddress()))
            if formatted not in to:
                to.append(formatted)

        managers = self.context.portal_groups.getGroupMembers('LabManagers')
        for bcc in managers:
            user = self.portal.acl_users.getUser(bcc)
            if user:
                uemail = user.getProperty('email')
                ufull = user.getProperty('fullname')
                formatted = formataddr((encode_header(ufull), uemail))
                if formatted not in to:
                    to.append(formatted)
        mime_msg['To'] = ','.join(to)
        aranchor = "<a href='%s'>%s</a>" % (ar.absolute_url(),
                                            ar.getId())
        naranchor = "<a href='%s'>%s</a>" % (newar.absolute_url(),
                                             newar.getId())
        addremarks = ('addremarks' in self.request and ar.getRemarks()) and ("<br/><br/>" + _("Additional remarks:") +
                                                                             "<br/>" + ar.getRemarks().split("===")[1].strip() +
                                                                             "<br/><br/>") or ''
        sub_d = dict(request_link=aranchor,
                     new_request_link=naranchor,
                     remarks=addremarks,
                     lab_address=lab_address)
        body = Template("Some errors have been detected in the results report "
                        "published from the Analysis Request $request_link. The Analysis "
                        "Request $new_request_link has been created automatically and the "
                        "previous has been invalidated.<br/>The possible mistake "
                        "has been picked up and is under investigation.<br/><br/>"
                        "$remarks $lab_address").safe_substitute(sub_d)
        msg_txt = MIMEText(safe_unicode(body).encode('utf-8'),
                           _subtype='html')
        mime_msg.preamble = 'This is a multi-part MIME message.'
        mime_msg.attach(msg_txt)
        try:
            host = getToolByName(self.context, 'MailHost')
            host.send(mime_msg.as_string(), immediate=True)
        except Exception as msg:
            message = _('Unable to send an email to alert lab '
                        'client contacts that the Analysis Request has been '
                        'retracted: ${error}',
                        mapping={'error': safe_unicode(msg)})
            self.context.plone_utils.addPortalMessage(message, 'warning')

    def workflow_action_save_partitions_button(self):
        form = self.request.form
        # Sample Partitions or AR Manage Analyses: save Partition Table
        sample = self.context.portal_type == 'Sample' and self.context or\
            self.context.getSample()
        part_prefix = sample.getId() + "-P"
        nr_existing = len(sample.objectIds())
        nr_parts = len(form['PartTitle'][0])
        # add missing parts
        if nr_parts > nr_existing:
            for i in range(nr_parts - nr_existing):
                part = _createObjectByType("SamplePartition", sample, tmpID())
                part.setDateReceived = DateTime()
                part.processForm()
        # remove excess parts
        if nr_existing > nr_parts:
            for i in range(nr_existing - nr_parts):
                part = sample['%s%s' % (part_prefix, nr_existing - i)]
                analyses = part.getAnalyses()
                for a in analyses:
                    a.setSamplePartition(None)
                sample.manage_delObjects(['%s%s' % (part_prefix, nr_existing - i), ])
        # modify part container/preservation
        for part_uid, part_id in form['PartTitle'][0].items():
            part = sample["%s%s" % (part_prefix, part_id.split(part_prefix)[1])]
            part.edit(
                Container=form['getContainer'][0][part_uid],
                Preservation=form['getPreservation'][0][part_uid],
            )
            part.reindexObject()
            # Adding the Security Seal Intact checkbox's value to the container object
            container_uid = form['getContainer'][0][part_uid]
            uc = getToolByName(self.context, 'uid_catalog')
            cbr = uc(UID=container_uid)
            if cbr and len(cbr) > 0:
                container_obj = cbr[0].getObject()
            else:
                continue
            value = form.get('setSecuritySealIntact', {}).get(part_uid, '') == 'on'
            container_obj.setSecuritySealIntact(value)
        objects = WorkflowAction._get_selected_items(self)
        if not objects:
            message = _("No items have been selected")
            self.context.plone_utils.addPortalMessage(message, 'info')
            if self.context.portal_type == 'Sample':
                # in samples his table is on 'Partitions' tab
                self.destination_url = self.context.absolute_url() +\
                    "/partitions"
            else:
                # in ar context this table is on 'ManageAnalyses' tab
                self.destination_url = self.context.absolute_url() +\
                    "/analyses"
            self.request.response.redirect(self.destination_url)

    def workflow_action_save_analyses_button(self):
        form = self.request.form
        workflow = getToolByName(self.context, 'portal_workflow')
        bsc = self.context.bika_setup_catalog
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        # AR Manage Analyses: save Analyses
        ar = self.context
        sample = ar.getSample()
        objects = WorkflowAction._get_selected_items(self)
        if not objects:
            message = _("No analyses have been selected")
            self.context.plone_utils.addPortalMessage(message, 'info')
            self.destination_url = self.context.absolute_url() + "/analyses"
            self.request.response.redirect(self.destination_url)
            return
        Analyses = objects.keys()
        prices = form.get("Price", [None])[0]

        # Hidden analyses?
        # https://jira.bikalabs.com/browse/LIMS-1324
        outs = []
        hiddenans = form.get('Hidden', {})
        for uid in Analyses:
            hidden = hiddenans.get(uid, '')
            hidden = True if hidden == 'on' else False
            outs.append({'uid':uid, 'hidden':hidden})
        ar.setAnalysisServicesSettings(outs)

        specs = {}
        if form.get("min", None):
            for service_uid in Analyses:
                service = objects[service_uid]
                keyword = service.getKeyword()
                specs[service_uid] = {
                    "min": form["min"][0][service_uid],
                    "max": form["max"][0][service_uid],
                    "warn_min": form["warn_min"][0][service_uid],
                    "warn_max": form["warn_max"][0][service_uid],
                    "keyword": keyword,
                    "uid": service_uid,
                }
        else:
            for service_uid in Analyses:
                service = objects[service_uid]
                keyword = service.getKeyword()
                specs[service_uid] = ResultsRangeDict(keyword=keyword,
                                                      uid=service_uid)
        new = ar.setAnalyses(Analyses, prices=prices, specs=specs.values())
        # link analyses and partitions
        # If Bika Setup > Analyses > 'Display individual sample
        # partitions' is checked, no Partitions available.
        # https://github.com/bikalabs/Bika-LIMS/issues/1030
        if 'Partition' in form:
            for service_uid, service in objects.items():
                part_id = form['Partition'][0][service_uid]
                part = sample[part_id]
                analysis = ar[service.getKeyword()]
                analysis.setSamplePartition(part)
                analysis.reindexObject()
                partans = part.getAnalyses()
                partans.append(analysis)
                part.setAnalyses(partans)
                part.reindexObject()

        if new:
            ar_state = getCurrentState(ar)
            if wasTransitionPerformed(ar, 'to_be_verified'):
                # Apply to AR only; we don't want this transition to cascade.
                ar.REQUEST['workflow_skiplist'].append("retract all analyses")
                workflow.doActionFor(ar, 'retract')
                ar.REQUEST['workflow_skiplist'].remove("retract all analyses")
                ar_state = getCurrentState(ar)
            for analysis in new:
                changeWorkflowState(analysis, 'bika_analysis_workflow', ar_state)

        message = PMF("Changes saved.")
        self.context.plone_utils.addPortalMessage(message, 'info')
        self.destination_url = self.context.absolute_url()
        self.request.response.redirect(self.destination_url)

    def workflow_action_preserve(self):
        form = self.request.form
        workflow = getToolByName(self.context, 'portal_workflow')
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        checkPermission = self.context.portal_membership.checkPermission
        # Partition Preservation
        # the partition table shown in AR and Sample views sends it's
        # action button submits here.
        objects = WorkflowAction._get_selected_items(self)
        transitioned = []
        incomplete = []
        for obj_uid, obj in objects.items():
            part = obj
            # can't transition inactive items
            if workflow.getInfoFor(part, 'inactive_state', '') == 'inactive':
                continue
            if not checkPermission(PreserveSample, part):
                continue
            # grab this object's Preserver and DatePreserved from the form
            Preserver = form['getPreserver'][0][obj_uid].strip()
            Preserver = Preserver and Preserver or ''
            DatePreserved = form['getDatePreserved'][0][obj_uid].strip()
            DatePreserved = DatePreserved and DateTime(DatePreserved) or ''
            # write them to the sample
            part.setPreserver(Preserver)
            part.setDatePreserved(DatePreserved)
            # transition the object if both values are present
            if Preserver and DatePreserved:
                workflow.doActionFor(part, action)
                transitioned.append(part.id)
            else:
                incomplete.append(part.id)
            part.reindexObject()
            part.aq_parent.reindexObject()
        message = None
        if len(transitioned) > 1:
            message = _('${items} are waiting to be received.',
                        mapping={'items': safe_unicode(', '.join(transitioned))})
            self.context.plone_utils.addPortalMessage(message, 'info')
        elif len(transitioned) == 1:
            message = _('${item} is waiting to be received.',
                        mapping={'item': safe_unicode(', '.join(transitioned))})
            self.context.plone_utils.addPortalMessage(message, 'info')
        if not message:
            message = _('No changes made.')
            self.context.plone_utils.addPortalMessage(message, 'info')

        if len(incomplete) > 1:
            message = _('${items} are missing Preserver or Date Preserved',
                        mapping={'items': safe_unicode(', '.join(incomplete))})
            self.context.plone_utils.addPortalMessage(message, 'error')
        elif len(incomplete) == 1:
            message = _('${item} is missing Preserver or Preservation Date',
                        mapping={'item': safe_unicode(', '.join(incomplete))})
            self.context.plone_utils.addPortalMessage(message, 'error')

        self.destination_url = self.request.get_header("referer",
                               self.context.absolute_url())
        self.request.response.redirect(self.destination_url)

    def workflow_action_receive(self):
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        items = [self.context,] if came_from == 'workflow_action' \
                else self._get_selected_items().values()
        trans, dest = self.submitTransition(action, came_from, items)
        if trans and 'receive' in self.context.bika_setup.getAutoPrintStickers():
            transitioned = [item.UID() for item in items]
            tmpl = self.context.bika_setup.getAutoStickerTemplate()
            q = "/sticker?autoprint=1&template=%s&items=" % tmpl
            q += ",".join(transitioned)
            self.request.response.redirect(self.context.absolute_url() + q)
        elif trans:
            message = PMF('Changes saved.')
            self.context.plone_utils.addPortalMessage(message, 'info')
            self.destination_url = self.context.absolute_url()
            self.request.response.redirect(self.destination_url)

    def workflow_action_submit(self):
        AnalysesWorkflowAction.workflow_action_submit(self)
        checkPermission = self.context.portal_membership.checkPermission
        if checkPermission(EditResults, self.context):
            self.destination_url = self.context.absolute_url() + "/manage_results"
        else:
            self.destination_url = self.context.absolute_url()
        self.request.response.redirect(self.destination_url)

    def workflow_action_prepublish(self):
        self.workflow_action_publish()

    def workflow_action_republish(self):
        self.workflow_action_publish()

    def workflow_action_print(self):
        # Calls printLastReport method for selected ARs
        uids = self.request.get('uids',[])
        uc = getToolByName(self.context, 'uid_catalog')
        for obj in uc(UID=uids):
            ar=obj.getObject()
            ar.printLastReport()
        referer = self.request.get_header("referer")
        self.request.response.redirect(referer)

    def workflow_action_publish(self):
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        if not isActive(self.context):
            message = _('Item is inactive.')
            self.context.plone_utils.addPortalMessage(message, 'info')
            self.request.response.redirect(self.context.absolute_url())
            return
        # AR publish preview
        uids = self.request.form.get('uids', [self.context.UID()])
        items = ",".join(uids)
        self.request.response.redirect(
            self.context.portal_url() + "/analysisrequests/publish?items="
            + items)

    def workflow_action_verify(self):
        # default bika_listing.py/WorkflowAction, but then go to view screen.
        self.destination_url = self.context.absolute_url()
        action, came_from = WorkflowAction._get_form_workflow_action(self)
        if type(came_from) in (list, tuple):
            came_from = came_from[0]
        return self.workflow_action_default(action='verify', came_from=came_from)

    def workflow_action_invalidate(self):

        # AR should be retracted
        # Can't transition inactive ARs
        if not api.is_active(self.context):
            message = _('Item is inactive.')
            self.context.plone_utils.addPortalMessage(message, 'info')
            self.request.response.redirect(self.context.absolute_url())
            return

        # Retract the AR and get the retest
        api.do_transition_for(self.context, 'invalidate')
        retest = self.context.getRetest()

        # 4. The system immediately alerts the client contacts who ordered
        # the results, per email and SMS, that a possible mistake has been
        # picked up and is under investigation.
        # A much possible information is provided in the email, linking
        # to the AR online.
        bika_setup = api.get_bika_setup()
        if bika_setup.getNotifyOnARRetract():
            self.notify_ar_retract(self.context, retest)

        message = _('${items} invalidated.',
                    mapping={'items': self.context.getId()})
        self.context.plone_utils.addPortalMessage(message, 'warning')
        self.request.response.redirect(retest.absolute_url())

    def workflow_action_copy_to_new(self):
        # Pass the selected AR UIDs in the request, to ar_add.
        objects = WorkflowAction._get_selected_items(self)
        if not objects:
            message = _("No analyses have been selected")
            self.context.plone_utils.addPortalMessage(message, 'info')
            referer = self.request.get_header("referer")
            self.request.response.redirect(referer)
            return
        url = self.context.absolute_url() + "/ar_add" + \
            "?ar_count={0}".format(len(objects)) + \
            "&copy_from={0}".format(",".join(objects.keys()))
        self.request.response.redirect(url)
        return

    def workflow_action_schedule_sampling(self):
        """
        This function prevent the transition if the fields "SamplingDate"
        and "ScheduledSamplingSampler" are uncompleted.
        :returns: bool
        """
        from bika.lims.utils.workflow import schedulesampling
        message = 'Not expected transition.'
        # In Samples Folder we have to get each selected item
        if interfaces.ISamplesFolder.providedBy(self.context):
            select_objs = WorkflowAction._get_selected_items(self)
            message = _('Transition done.')
            for key in select_objs.keys():
                sample = select_objs[key]
                # Getting the sampler
                sch_sampl = self.request.form.get(
                    'getScheduledSamplingSampler', None)[0].get(key) if\
                    self.request.form.get(
                        'getScheduledSamplingSampler', None) else ''
                # Getting the date
                sampl_date = self.request.form.get(
                    'getSamplingDate', None)[0].get(key) if\
                    self.request.form.get(
                        'getSamplingDate', None) else ''
                # Setting both values
                sample.setScheduledSamplingSampler(sch_sampl)
                sample.setSamplingDate(sampl_date)
                # Transitioning the sample
                success, errmsg = schedulesampling.doTransition(sample)
                if errmsg == 'missing':
                    message = _(
                        "'Sampling date' and 'Define the Sampler for the" +
                        " scheduled sampling' must be completed and saved " +
                        "in order to schedule a sampling. Element: %s" %
                        sample.getId())
                elif errmsg == 'cant_trans':
                    message = _(
                        "The item %s can't be transitioned." % sample.getId())
                else:
                    message = _('Transition done.')
                self.context.plone_utils.addPortalMessage(message, 'info')
        else:
            success, errmsg = schedulesampling.doTransition(self.context)
            if errmsg == 'missing':
                message = _(
                    "'Sampling date' and 'Define the Sampler for the" +
                    " scheduled sampling' must be completed and saved in " +
                    "order to schedule a sampling.")
            elif errmsg == 'cant_trans':
                message = _("The item can't be transitioned.")
            else:
                message = _('Transition done.')
            self.context.plone_utils.addPortalMessage(message, 'info')
        # Reload the page in order to show the portal message
        self.request.response.redirect(self.context.absolute_url())
        return success



1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE
4			#
5			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			from email.mime.multipart import MIMEMultipart
9			from email.mime.text import MIMEText
10			from string import Template
11
12			import plone
13			from DateTime import DateTime
14			from Products.CMFCore.utils import getToolByName
15			from Products.CMFPlone.utils import _createObjectByType, safe_unicode
16			from bika.lims import PMF, api
17			from bika.lims import bikaMessageFactory as _
18			from bika.lims import interfaces
19			from bika.lims.browser.analyses.workflow import AnalysesWorkflowAction
20			from bika.lims.browser.bika_listing import WorkflowAction
21			from bika.lims.content.analysisspec import ResultsRangeDict
22			from bika.lims.permissions import *
23			from bika.lims.utils import changeWorkflowState
24			from bika.lims.utils import encode_header
25			from bika.lims.utils import isActive
26			from bika.lims.utils import t
27			from bika.lims.utils import tmpID
28			from bika.lims.workflow import getCurrentState
29			from bika.lims.workflow import wasTransitionPerformed
30			from email.Utils import formataddr
31
32
33			class AnalysisRequestWorkflowAction(AnalysesWorkflowAction):
34
35			"""Workflow actions taken in AnalysisRequest context.
36
37			Sample context workflow actions also redirect here
38			Applies to
39			Analysis objects
40			SamplePartition objects
41			"""
42
43			def __call__(self):
44			form = self.request.form
45			plone.protect.CheckAuthenticator(form)
46			action, came_from = WorkflowAction._get_form_workflow_action(self)
47			if type(action) in (list, tuple):
48			action = action[0]
49			if type(came_from) in (list, tuple):
50			came_from = came_from[0]
51			# Call out to the workflow action method
52			# Use default bika_listing.py/WorkflowAction for other transitions
53			method_name = 'workflow_action_' + action if action else ''
54			method = getattr(self, method_name, False)
55			if method:
56			method()
57			else:
58			WorkflowAction.__call__(self)
59
60			def notify_ar_retract(self, ar, newar):
61			bika_setup = api.get_bika_setup()
62			laboratory = bika_setup.laboratory
63			lab_address = "<br/>".join(laboratory.getPrintAddress())
64			mime_msg = MIMEMultipart('related')
65			mime_msg['Subject'] = t(_("Erroneus result publication from ${request_id}",
66			mapping={"request_id": ar.getId()}))
67			mime_msg['From'] = formataddr(
68			(encode_header(laboratory.getName()),
69			laboratory.getEmailAddress()))
70			to = []
71			contact = ar.getContact()
72			if contact:
73			to.append(formataddr((encode_header(contact.Title()),
74			contact.getEmailAddress())))
75			for cc in ar.getCCContact():
76			formatted = formataddr((encode_header(cc.Title()),
77			cc.getEmailAddress()))
78			if formatted not in to:
79			to.append(formatted)
80
81			managers = self.context.portal_groups.getGroupMembers('LabManagers')
82			for bcc in managers:
83			user = self.portal.acl_users.getUser(bcc)
84			if user:
85			uemail = user.getProperty('email')
86			ufull = user.getProperty('fullname')
87			formatted = formataddr((encode_header(ufull), uemail))
88			if formatted not in to:
89			to.append(formatted)
90			mime_msg['To'] = ','.join(to)
91			aranchor = "<a href='%s'>%s</a>" % (ar.absolute_url(),
92			ar.getId())
93			naranchor = "<a href='%s'>%s</a>" % (newar.absolute_url(),
94			newar.getId())
95			addremarks = ('addremarks' in self.request and ar.getRemarks()) and ("<br/><br/>" + _("Additional remarks:") +
96			"<br/>" + ar.getRemarks().split("===")[1].strip() +
97			"<br/><br/>") or ''
98			sub_d = dict(request_link=aranchor,
99			new_request_link=naranchor,
100			remarks=addremarks,
101			lab_address=lab_address)
102			body = Template("Some errors have been detected in the results report "
103			"published from the Analysis Request $request_link. The Analysis "
104			"Request $new_request_link has been created automatically and the "
105			"previous has been invalidated.<br/>The possible mistake "
106			"has been picked up and is under investigation.<br/><br/>"
107			"$remarks $lab_address").safe_substitute(sub_d)
108			msg_txt = MIMEText(safe_unicode(body).encode('utf-8'),
109			_subtype='html')
110			mime_msg.preamble = 'This is a multi-part MIME message.'
111			mime_msg.attach(msg_txt)
112			try:
113			host = getToolByName(self.context, 'MailHost')
114			host.send(mime_msg.as_string(), immediate=True)
115			except Exception as msg:
116			message = _('Unable to send an email to alert lab '
117			'client contacts that the Analysis Request has been '
118			'retracted: ${error}',
119			mapping={'error': safe_unicode(msg)})
120			self.context.plone_utils.addPortalMessage(message, 'warning')
121
122			def workflow_action_save_partitions_button(self):
123			form = self.request.form
124			# Sample Partitions or AR Manage Analyses: save Partition Table
125			sample = self.context.portal_type == 'Sample' and self.context or\
126			self.context.getSample()
127			part_prefix = sample.getId() + "-P"
128			nr_existing = len(sample.objectIds())
129			nr_parts = len(form['PartTitle'][0])
130			# add missing parts
131			if nr_parts > nr_existing:
132			for i in range(nr_parts - nr_existing):
133			part = _createObjectByType("SamplePartition", sample, tmpID())
134			part.setDateReceived = DateTime()
135			part.processForm()
136			# remove excess parts
137			if nr_existing > nr_parts:
138			for i in range(nr_existing - nr_parts):
139			part = sample['%s%s' % (part_prefix, nr_existing - i)]
140			analyses = part.getAnalyses()
141			for a in analyses:
142			a.setSamplePartition(None)
143			sample.manage_delObjects(['%s%s' % (part_prefix, nr_existing - i), ])
144			# modify part container/preservation
145			for part_uid, part_id in form['PartTitle'][0].items():
146			part = sample["%s%s" % (part_prefix, part_id.split(part_prefix)[1])]
147			part.edit(
148			Container=form['getContainer'][0][part_uid],
149			Preservation=form['getPreservation'][0][part_uid],
150			)
151			part.reindexObject()
152			# Adding the Security Seal Intact checkbox's value to the container object
153			container_uid = form['getContainer'][0][part_uid]
154			uc = getToolByName(self.context, 'uid_catalog')
155			cbr = uc(UID=container_uid)
156			if cbr and len(cbr) > 0:
157			container_obj = cbr[0].getObject()
158			else:
159			continue
160			value = form.get('setSecuritySealIntact', {}).get(part_uid, '') == 'on'
161			container_obj.setSecuritySealIntact(value)
162			objects = WorkflowAction._get_selected_items(self)
163			if not objects:
164			message = _("No items have been selected")
165			self.context.plone_utils.addPortalMessage(message, 'info')
166			if self.context.portal_type == 'Sample':
167			# in samples his table is on 'Partitions' tab
168			self.destination_url = self.context.absolute_url() +\
169			"/partitions"
170			else:
171			# in ar context this table is on 'ManageAnalyses' tab
172			self.destination_url = self.context.absolute_url() +\
173			"/analyses"
174			self.request.response.redirect(self.destination_url)
175
176			def workflow_action_save_analyses_button(self):
177			form = self.request.form
178			workflow = getToolByName(self.context, 'portal_workflow')
179			bsc = self.context.bika_setup_catalog
180			action, came_from = WorkflowAction._get_form_workflow_action(self)
181			# AR Manage Analyses: save Analyses
182			ar = self.context
183			sample = ar.getSample()
184			objects = WorkflowAction._get_selected_items(self)
185			if not objects:
186			message = _("No analyses have been selected")
187			self.context.plone_utils.addPortalMessage(message, 'info')
188			self.destination_url = self.context.absolute_url() + "/analyses"
189			self.request.response.redirect(self.destination_url)
190			return
191			Analyses = objects.keys()
192			prices = form.get("Price", [None])[0]
193
194			# Hidden analyses?
195			# https://jira.bikalabs.com/browse/LIMS-1324
196			outs = []
197			hiddenans = form.get('Hidden', {})
198			for uid in Analyses:
199			hidden = hiddenans.get(uid, '')
200			hidden = True if hidden == 'on' else False
201			outs.append({'uid':uid, 'hidden':hidden})
202			ar.setAnalysisServicesSettings(outs)
203
204			specs = {}
205			if form.get("min", None):
206			for service_uid in Analyses:
207			service = objects[service_uid]
208			keyword = service.getKeyword()
209			specs[service_uid] = {
210			"min": form["min"][0][service_uid],
211			"max": form["max"][0][service_uid],
212			"warn_min": form["warn_min"][0][service_uid],
213			"warn_max": form["warn_max"][0][service_uid],
214			"keyword": keyword,
215			"uid": service_uid,
216			}
217			else:
218			for service_uid in Analyses:
219			service = objects[service_uid]
220			keyword = service.getKeyword()
221			specs[service_uid] = ResultsRangeDict(keyword=keyword,
222			uid=service_uid)
223			new = ar.setAnalyses(Analyses, prices=prices, specs=specs.values())
224			# link analyses and partitions
225			# If Bika Setup > Analyses > 'Display individual sample
226			# partitions' is checked, no Partitions available.
227			# https://github.com/bikalabs/Bika-LIMS/issues/1030
228			if 'Partition' in form:
229			for service_uid, service in objects.items():
230			part_id = form['Partition'][0][service_uid]
231			part = sample[part_id]
232			analysis = ar[service.getKeyword()]
233			analysis.setSamplePartition(part)
234			analysis.reindexObject()
235			partans = part.getAnalyses()
236			partans.append(analysis)
237			part.setAnalyses(partans)
238			part.reindexObject()
239
240			if new:
241			ar_state = getCurrentState(ar)
242			if wasTransitionPerformed(ar, 'to_be_verified'):
243			# Apply to AR only; we don't want this transition to cascade.
244			ar.REQUEST['workflow_skiplist'].append("retract all analyses")
245			workflow.doActionFor(ar, 'retract')
246			ar.REQUEST['workflow_skiplist'].remove("retract all analyses")
247			ar_state = getCurrentState(ar)
248			for analysis in new:
249			changeWorkflowState(analysis, 'bika_analysis_workflow', ar_state)
250
251			message = PMF("Changes saved.")
252			self.context.plone_utils.addPortalMessage(message, 'info')
253			self.destination_url = self.context.absolute_url()
254			self.request.response.redirect(self.destination_url)
255
256			def workflow_action_preserve(self):
257			form = self.request.form
258			workflow = getToolByName(self.context, 'portal_workflow')
259			action, came_from = WorkflowAction._get_form_workflow_action(self)
260			checkPermission = self.context.portal_membership.checkPermission
261			# Partition Preservation
262			# the partition table shown in AR and Sample views sends it's
263			# action button submits here.
264			objects = WorkflowAction._get_selected_items(self)
265			transitioned = []
266			incomplete = []
267			for obj_uid, obj in objects.items():
268			part = obj
269			# can't transition inactive items
270			if workflow.getInfoFor(part, 'inactive_state', '') == 'inactive':
271			continue
272			if not checkPermission(PreserveSample, part):
273			continue
274			# grab this object's Preserver and DatePreserved from the form
275			Preserver = form['getPreserver'][0][obj_uid].strip()
276			Preserver = Preserver and Preserver or ''
277			DatePreserved = form['getDatePreserved'][0][obj_uid].strip()
278			DatePreserved = DatePreserved and DateTime(DatePreserved) or ''
279			# write them to the sample
280			part.setPreserver(Preserver)
281			part.setDatePreserved(DatePreserved)
282			# transition the object if both values are present
283			if Preserver and DatePreserved:
284			workflow.doActionFor(part, action)
285			transitioned.append(part.id)
286			else:
287			incomplete.append(part.id)
288			part.reindexObject()
289			part.aq_parent.reindexObject()
290			message = None
291			if len(transitioned) > 1:
292			message = _('${items} are waiting to be received.',
293			mapping={'items': safe_unicode(', '.join(transitioned))})
294			self.context.plone_utils.addPortalMessage(message, 'info')
295			elif len(transitioned) == 1:
296			message = _('${item} is waiting to be received.',
297			mapping={'item': safe_unicode(', '.join(transitioned))})
298			self.context.plone_utils.addPortalMessage(message, 'info')
299			if not message:
300			message = _('No changes made.')
301			self.context.plone_utils.addPortalMessage(message, 'info')
302
303			if len(incomplete) > 1:
304			message = _('${items} are missing Preserver or Date Preserved',
305			mapping={'items': safe_unicode(', '.join(incomplete))})
306			self.context.plone_utils.addPortalMessage(message, 'error')
307			elif len(incomplete) == 1:
308			message = _('${item} is missing Preserver or Preservation Date',
309			mapping={'item': safe_unicode(', '.join(incomplete))})
310			self.context.plone_utils.addPortalMessage(message, 'error')
311
312			self.destination_url = self.request.get_header("referer",
313			self.context.absolute_url())
314			self.request.response.redirect(self.destination_url)
315
316			def workflow_action_receive(self):
317			action, came_from = WorkflowAction._get_form_workflow_action(self)
318			items = [self.context,] if came_from == 'workflow_action' \
319			else self._get_selected_items().values()
320			trans, dest = self.submitTransition(action, came_from, items)
321			if trans and 'receive' in self.context.bika_setup.getAutoPrintStickers():
322			transitioned = [item.UID() for item in items]
323			tmpl = self.context.bika_setup.getAutoStickerTemplate()
324			q = "/sticker?autoprint=1&template=%s&items=" % tmpl
325			q += ",".join(transitioned)
326			self.request.response.redirect(self.context.absolute_url() + q)
327			elif trans:
328			message = PMF('Changes saved.')
329			self.context.plone_utils.addPortalMessage(message, 'info')
330			self.destination_url = self.context.absolute_url()
331			self.request.response.redirect(self.destination_url)
332
333			def workflow_action_submit(self):
334			AnalysesWorkflowAction.workflow_action_submit(self)
335			checkPermission = self.context.portal_membership.checkPermission
336			if checkPermission(EditResults, self.context):
337			self.destination_url = self.context.absolute_url() + "/manage_results"
338			else:
339			self.destination_url = self.context.absolute_url()
340			self.request.response.redirect(self.destination_url)
341
342			def workflow_action_prepublish(self):
343			self.workflow_action_publish()
344
345			def workflow_action_republish(self):
346			self.workflow_action_publish()
347
348			def workflow_action_print(self):
349			# Calls printLastReport method for selected ARs
350			uids = self.request.get('uids',[])
351			uc = getToolByName(self.context, 'uid_catalog')
352			for obj in uc(UID=uids):
353			ar=obj.getObject()
354			ar.printLastReport()
355			referer = self.request.get_header("referer")
356			self.request.response.redirect(referer)
357
358			def workflow_action_publish(self):
359			action, came_from = WorkflowAction._get_form_workflow_action(self)
360			if not isActive(self.context):
361			message = _('Item is inactive.')
362			self.context.plone_utils.addPortalMessage(message, 'info')
363			self.request.response.redirect(self.context.absolute_url())
364			return
365			# AR publish preview
366			uids = self.request.form.get('uids', [self.context.UID()])
367			items = ",".join(uids)
368			self.request.response.redirect(
369			self.context.portal_url() + "/analysisrequests/publish?items="
370			+ items)
371
372			def workflow_action_verify(self):
373			# default bika_listing.py/WorkflowAction, but then go to view screen.
374			self.destination_url = self.context.absolute_url()
375			action, came_from = WorkflowAction._get_form_workflow_action(self)
376			if type(came_from) in (list, tuple):
377			came_from = came_from[0]
378			return self.workflow_action_default(action='verify', came_from=came_from)
379
380			def workflow_action_invalidate(self):
381
382			# AR should be retracted
383			# Can't transition inactive ARs
384			if not api.is_active(self.context):
385			message = _('Item is inactive.')
386			self.context.plone_utils.addPortalMessage(message, 'info')
387			self.request.response.redirect(self.context.absolute_url())
388			return
389
390			# Retract the AR and get the retest
391			api.do_transition_for(self.context, 'invalidate')
392			retest = self.context.getRetest()
393
394			# 4. The system immediately alerts the client contacts who ordered
395			# the results, per email and SMS, that a possible mistake has been
396			# picked up and is under investigation.
397			# A much possible information is provided in the email, linking
398			# to the AR online.
399			bika_setup = api.get_bika_setup()
400			if bika_setup.getNotifyOnARRetract():
401			self.notify_ar_retract(self.context, retest)
402
403			message = _('${items} invalidated.',
404			mapping={'items': self.context.getId()})
405			self.context.plone_utils.addPortalMessage(message, 'warning')
406			self.request.response.redirect(retest.absolute_url())
407
408			def workflow_action_copy_to_new(self):
409			# Pass the selected AR UIDs in the request, to ar_add.
410			objects = WorkflowAction._get_selected_items(self)
411			if not objects:
412			message = _("No analyses have been selected")
413			self.context.plone_utils.addPortalMessage(message, 'info')
414			referer = self.request.get_header("referer")
415			self.request.response.redirect(referer)
416			return
417			url = self.context.absolute_url() + "/ar_add" + \
418			"?ar_count={0}".format(len(objects)) + \
419			"&copy_from={0}".format(",".join(objects.keys()))
420			self.request.response.redirect(url)
421			return
422
423			def workflow_action_schedule_sampling(self):
424			"""
425			This function prevent the transition if the fields "SamplingDate"
426			and "ScheduledSamplingSampler" are uncompleted.
427			:returns: bool
428			"""
429			from bika.lims.utils.workflow import schedulesampling
430			message = 'Not expected transition.'
431			# In Samples Folder we have to get each selected item
432			if interfaces.ISamplesFolder.providedBy(self.context):
433			select_objs = WorkflowAction._get_selected_items(self)
434			message = _('Transition done.')
435			for key in select_objs.keys():
436			sample = select_objs[key]
437			# Getting the sampler
438			sch_sampl = self.request.form.get(
439			'getScheduledSamplingSampler', None)[0].get(key) if\
440			self.request.form.get(
441			'getScheduledSamplingSampler', None) else ''
442			# Getting the date
443			sampl_date = self.request.form.get(
444			'getSamplingDate', None)[0].get(key) if\
445			self.request.form.get(
446			'getSamplingDate', None) else ''
447			# Setting both values
448			sample.setScheduledSamplingSampler(sch_sampl)
449			sample.setSamplingDate(sampl_date)
450			# Transitioning the sample
451			success, errmsg = schedulesampling.doTransition(sample)
452			if errmsg == 'missing':
453			message = _(
454			"'Sampling date' and 'Define the Sampler for the" +
455			" scheduled sampling' must be completed and saved " +
456			"in order to schedule a sampling. Element: %s" %
457			sample.getId())
458			elif errmsg == 'cant_trans':
459			message = _(
460			"The item %s can't be transitioned." % sample.getId())
461			else:
462			message = _('Transition done.')
463			self.context.plone_utils.addPortalMessage(message, 'info')
464			else:
465			success, errmsg = schedulesampling.doTransition(self.context)
466			if errmsg == 'missing':
467			message = _(
468			"'Sampling date' and 'Define the Sampler for the" +
469			" scheduled sampling' must be completed and saved in " +
470			"order to schedule a sampling.")
471			elif errmsg == 'cant_trans':
472			message = _("The item can't be transitioned.")
473			else:
474			message = _('Transition done.')
475			self.context.plone_utils.addPortalMessage(message, 'info')
476			# Reload the page in order to show the portal message
477			self.request.response.redirect(self.context.absolute_url())
478			return success
			0 ignored issues – show introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report The variable `success` does not seem to be defined in case the `for` loop on line `435` is not entered. Are you sure this can never be the case? Loading history...
479

senaite / senaite.core

Push — master ( 8d2e48...36dfc2 )

bika.lims.browser.analysisrequest.workflow.AnalysisRequestWorkflowAction.cloneAR() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like