bika.lims.browser.workflow.analysisrequest

Complexity

Total Complexity

58

Size/Duplication

Total Lines	399
Duplicated Lines	17.54 %

Importance

Changes

0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	WorkflowActionPrintStickersAdapter.__call__()	0	3	1
A	WorkflowActionReceiveAdapter.is_auto_print_stickers_enabled()	0	4	1
A	WorkflowActionPrintSampleAdapter.set_printed_time()	0	14	3
A	WorkflowActionPrintSampleAdapter.__call__()	0	8	3
A	WorkflowActionScheduleSamplingAdapter.set_sampling_info()	0	16	4
A	WorkflowActionSaveAnalysesAdapter.is_ar_specs_enabled()	0	7	1
A	WorkflowActionRejectAdapter.__call__()	18	18	3
A	WorkflowActionCreatePartitionsAdapter.__call__()	0	3	1
A	WorkflowActionSaveAnalysesAdapter.is_hidden()	0	6	1
A	WorkflowActionSaveAnalysesAdapter.get_specs()	0	10	2
B	WorkflowActionSaveAnalysesAdapter.__call__()	0	48	6
A	WorkflowActionPreserveAdapter.set_preserver_info()	0	17	4
A	WorkflowActionInvalidateAdapter.__call__()	18	18	3
A	WorkflowActionSampleAdapter.set_sampler_info()	0	32	5
A	WorkflowActionSampleAdapter.__call__()	0	16	4
A	WorkflowActionCopyToNewAdapter.__call__()	0	4	1
A	WorkflowActionReceiveAdapter.is_auto_partition_required()	0	8	2
A	WorkflowActionScheduleSamplingAdapter.__call__()	13	13	4
A	WorkflowActionReceiveAdapter.__call__()	0	20	4
A	WorkflowActionPreserveAdapter.__call__()	13	13	4
A	WorkflowActionPublishAdapter.__call__()	0	6	1

# -*- 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-2025 by it's authors.
# Some rights reserved, see README and LICENSE.

from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims.browser.workflow import RequestContextAware
from bika.lims.browser.workflow import WorkflowActionGenericAdapter
from bika.lims.content.analysisspec import ResultsRangeDict
from bika.lims.interfaces import IAnalysisRequest
from bika.lims.interfaces import IWorkflowActionUIDsAdapter
from DateTime import DateTime
from zope.interface import implements


class WorkflowActionCopyToNewAdapter(RequestContextAware):
    """Adapter in charge of Analysis Requests 'copy_to_new' action
    """
    implements(IWorkflowActionUIDsAdapter)

    def __call__(self, action, uids):
        url = "{}/ar_add?ar_count={}&copy_from={}".format(
            self.back_url, len(uids), ",".join(uids))
        return self.redirect(redirect_url=url)


class WorkflowActionPrintStickersAdapter(RequestContextAware):
    """Adapter in charge of Analysis Requests 'print_stickers' action
    """
    implements(IWorkflowActionUIDsAdapter)

    def __call__(self, action, uids):
        url = "{}/sticker?items={}".format(self.back_url, ",".join(uids))
        return self.redirect(redirect_url=url)


class WorkflowActionCreatePartitionsAdapter(RequestContextAware):
    """Adapter in charge of Analysis Requests 'copy_to_new' action
    """
    implements(IWorkflowActionUIDsAdapter)

    def __call__(self, action, uids):
        url = "{}/partition_magic?uids={}".format(self.back_url, ",".join(uids))
        return self.redirect(redirect_url=url)


class WorkflowActionPublishAdapter(RequestContextAware):
    """Adapter in charge of Analysis Requests 'publish'-like actions
    """
    implements(IWorkflowActionUIDsAdapter)

    def __call__(self, action, uids):
        uids = ",".join(uids)
        portal = api.get_portal()
        portal_url = api.get_url(portal)
        url = "{}/samples/publish?items={}".format(portal_url, uids)
        return self.redirect(redirect_url=url)


class WorkflowActionRejectAdapter(WorkflowActionGenericAdapter):

This code seems to be duplicated in your project.

    """Adapter in charge of Analysis Requests 'reject' action
    """

    def __call__(self, action, objects):
        samples = filter(IAnalysisRequest.providedBy, objects)
        if samples:
            # Action reject applies to samples. Redirect to Sample Reject view
            uids = map(api.get_uid, samples)
            uids_str = ",".join(uids)
            url = "{}/reject_samples?uids={}".format(self.back_url, uids_str)
            return self.redirect(redirect_url=url)

        # Generic transition if reject applies to other types (e.g. Analysis)
        transitioned = self.do_action(action, objects)
        if not transitioned:
            return self.redirect(message=_("No changes made."), level="warning")

        # Redirect the user to success page
        ids =  map(api.get_id, transitioned)
        message = _("Rejected items: {}").format(", ".join(ids))
        return self.success(transitioned, message=message)


class WorkflowActionReceiveAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of Analysis Request receive action
    """

    def __call__(self, action, objects):
        transitioned = self.do_action(action, objects)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        auto_partitions = filter(self.is_auto_partition_required, objects)
        if auto_partitions:
            # Redirect to the partitioning view
            uids = ",".join(map(api.get_uid, auto_partitions))
            url = "{}/partition_magic?uids={}".format(self.back_url, uids)
            return self.redirect(redirect_url=url)

        if self.is_auto_print_stickers_enabled():
            # Redirect to the auto-print stickers view
            uids = ",".join(map(api.get_uid, transitioned))
            url = "{}/sticker?autoprint=1&items={}".format(self.back_url, uids)
            return self.redirect(redirect_url=url)

        # Redirect the user to success page
        return self.success(transitioned)

    def is_auto_partition_required(self, brain_or_object):
        """Returns whether the passed in object needs to be partitioned
        """
        obj = api.get_object(brain_or_object)
        if not IAnalysisRequest.providedBy(obj):
            return False
        template = obj.getTemplate()
        return template and template.getAutoPartition()

    def is_auto_print_stickers_enabled(self):
        """Returns whether the auto print of stickers on reception is enabled
        """
        return "receive" in self.context.bika_setup.getAutoPrintStickers()


class WorkflowActionInvalidateAdapter(WorkflowActionGenericAdapter):

This code seems to be duplicated in your project.

    """Adapter in charge of Analysis Request invalidate action
    """

    def __call__(self, action, objects):
        samples = filter(IAnalysisRequest.providedBy, objects)
        if samples:
            # Redirect to the samples invalidation view
            uids = map(api.get_uid, samples)
            uids = ",".join(uids)
            url = "%s/invalidate_samples?uids=%s" % (self.back_url, uids)
            return self.redirect(redirect_url=url)

        # Generic transition if invalidation applies to other types
        transitioned = self.do_action(action, objects)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Redirect the user to success page
        ids = map(api.get_id, transitioned)
        message = _("Invalidated items: {}").format(", ".join(ids))
        return self.success(transitioned, message=message)


class WorkflowActionPrintSampleAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of Analysis Request print_sample action
    """

    def __call__(self, action, objects):
        # Update printed times
        transitioned = filter(lambda obj: self.set_printed_time(obj), objects)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Redirect the user to success page
        return self.success(transitioned)

    def set_printed_time(self, sample):
        """Updates the printed time of the last results report from the sample
        """
        if api.get_workflow_status_of(sample) != "published":
            return False

        reports = sample.objectIds("ARReport")
        if not reports:
            return False

        last_report = sample.get(reports[-1])
        last_report.setDatePrinted(DateTime())
        sample.reindexObject(idxs=["getPrinted"])
        return True


class WorkflowActionSampleAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of Analysis Request sample action
    """

    def __call__(self, action, objects):
        # Assign the Sampler and DateSampled
        for obj in objects:
            try:
                self.set_sampler_info(obj)
            except ValueError as e:
                return self.redirect(message=str(e), level="warning")

        # Trigger "sample" transition
        transitioned = self.do_action(action, objects)
        if not transitioned:
            message = _("Could not transition samples to the sampled state")
            return self.redirect(message=message, level="warning")

        # Redirect the user to success page
        return self.success(transitioned)

    def set_sampler_info(self, sample):
        """Updates the Sampler and the Sample Date with the values provided in
        the request. If neither Sampler nor SampleDate are present in the
        request, returns False
        """
        if sample.getSampler() and sample.getDateSampled():
            # Sampler and Date Sampled already set. This is correct
            return True

        # Try to get the sampler and date sampled from the request.
        # This might happen when the "Sample" transition is triggered from the
        # samples listing view (form keys == column names of the listing)

        # try to get the sampler from the request
        sampler = self.get_form_value("getSampler", sample,
                                      sample.getSampler())
        if not sampler:
            sid = api.get_id(sample)
            raise ValueError(_("Sampler required for sample %s" % sid))

        # try to get the date sampled from the request
        sampled = self.get_form_value("getDateSampled", sample,
                                      sample.getDateSampled())
        if not sampled:
            sid = api.get_id(sample)
            raise ValueError(_("Sample date required for sample %s" % sid))

        # set the field values
        sample.setSampler(sampler)
        sample.setDateSampled(sampled)

        return True


class WorkflowActionPreserveAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of Analysis Request preserve action
    """

    def __call__(self, action, objects):

This code seems to be duplicated in your project.

        # Assign the Preserver and DatePreserved
        transitioned = filter(lambda obj: self.set_preserver_info(obj), objects)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Trigger "preserve" transition
        transitioned = self.do_action(action, transitioned)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Redirect the user to success page
        return self.success(transitioned)

    def set_preserver_info(self, sample):
        """Updates the Preserver and the Date Preserved with the values provided
        in the request. If neither Preserver nor DatePreserved are present in
        the request, returns False
        """
        if sample.getPreserver() and sample.getDatePreserved():
            # Preserver and Date Preserved already set. This is correct
            return True
        preserver = self.get_form_value("Preserver", sample,
                                        sample.getPreserver())
        preserved = self.get_form_value("getDatePreserved",
                                        sample.getDatePreserved())
        if not all([preserver, preserved]):
            return False
        sample.setPreserver(preserver)
        sample.setDatePreserver(DateTime(preserved))
        return True


class WorkflowActionScheduleSamplingAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of Analysis request schedule sampling action
    """

    def __call__(self, action, objects):

This code seems to be duplicated in your project.

        # Assign the scheduled Sampler and Sampling Date
        transitioned = filter(lambda obj: self.set_sampling_info(obj), objects)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Trigger "schedule_sampling" transition
        transitioned = self.do_action(action, transitioned)
        if not transitioned:
            return self.redirect(message=_("No changes made"), level="warning")

        # Redirect the user to success page
        return self.success(transitioned)

    def set_sampling_info(self, sample):
        """Updates the scheduled Sampling sampler and the Sampling Date with the
        values provided in the request. If neither Sampling sampler nor Sampling
        Date are present in the request, returns False
        """
        if sample.getScheduledSamplingSampler() and sample.getSamplingDate():
            return True
        sampler = self.get_form_value("getScheduledSamplingSampler", sample,
                                      sample.getScheduledSamplingSampler())
        sampled = self.get_form_value("getSamplingDate",
                                      sample.getSamplingDate())
        if not all([sampler, sampled]):
            return False
        sample.setScheduledSamplingSampler(sampler)
        sample.setSamplingDate(DateTime(sampled))
        return True


class WorkflowActionSaveAnalysesAdapter(WorkflowActionGenericAdapter):
    """Adapter in charge of "save analyses" action in Analysis Request.
    """

    def __call__(self, action, objects):
        """The objects passed in are Analysis Services and the context is the
        Analysis Request
        """
        sample = self.context
        if not IAnalysisRequest.providedBy(sample):
            return self.redirect(message=_("No changes made"), level="warning")

        # NOTE: https://github.com/senaite/senaite.core/issues/1276
        #
        # Explicitly lookup the UIDs from the request, because the default
        # behavior of the method `get_uids` in `WorkflowActionGenericAdapter`
        # falls back to the UID of the current context if no UIDs were
        # submitted, which is in that case an `AnalysisRequest`.
        uids = self.get_uids_from_request()
        services = map(api.get_object, uids)

        # Get form values
        form = self.request.form
        prices = form.get("Price", [None])[0]
        hidden = map(lambda o: {
            "uid": api.get_uid(o), "hidden": self.is_hidden(o)
        }, services)

        # Do not overwrite default result ranges set through sample
        # specification field unless the edition of specs at analysis
        # level is explicitely allowed
        specs = []
        if self.is_ar_specs_enabled:
            specs = map(lambda service: self.get_specs(service), services)

        # Set new analyses to the sample
        sample.setAnalysisServicesSettings(hidden)
        sample.setAnalyses(uids, prices=prices, specs=specs, hidden=hidden)

        # Just in case new analyses have been added while the Sample was in a
        # "non-open" state (e.g. "to_be_verified")
        self.do_action("rollback_to_receive", [sample])

        # Reindex the analyses that have been added
        for analysis in sample.objectValues("Analysis"):
            analysis.reindexObject()

        # Reindex the Sample
        sample.reindexObject()

        # Redirect the user to success page
        self.success([sample])

    @property
    def is_ar_specs_enabled(self):
        """Returns whether the assignment of specs at analysis level within
        sample context is enabled or not
        """
        setup = api.get_setup()
        return setup.getEnableARSpecs()

    def is_hidden(self, service):
        """Returns whether the request Hidden param for the given obj is True
        """
        uid = api.get_uid(service)
        hidden_ans = self.request.form.get("Hidden", {})
        return hidden_ans.get(uid, "") == "on"

    def get_specs(self, service):
        """Returns the analysis specs available in the request for the given uid
        """
        uid = api.get_uid(service)
        keyword = service.getKeyword()
        specs = ResultsRangeDict(keyword=keyword, uid=uid).copy()
        for key in specs.keys():
            specs_value = self.request.form.get(key, [{}])[0].get(uid, None)
            specs[key] = specs_value or specs.get(key)
        return specs