bika.lims.browser.analysisrequest.view.AnalysisRequestViewView.get_custom_fields() - 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

AnalysisRequestViewView.get_custom_fields() A

↳ Parent: bika.lims.browser.analysisrequest.view

Complexity

Conditions

Size

Total Lines	28
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	19
dl	0
loc	28
rs	9.45
c	0
b	0
f	0
cc	4
nop	1

# -*- 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 AccessControl import getSecurityManager
from DateTime import DateTime
from Products.CMFCore.utils import getToolByName
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from bika.lims import bikaMessageFactory as _
from bika.lims import api
from bika.lims.browser import BrowserView
from bika.lims.browser.analyses import AnalysesView
from bika.lims.browser.analyses import QCAnalysesView
from bika.lims.browser.header_table import HeaderTableView
from bika.lims.browser.sample import SamplePartitionsView
from bika.lims.config import POINTS_OF_CAPTURE
from bika.lims.permissions import *
from bika.lims.utils import isActive
from bika.lims.utils import t, check_permission
from bika.lims.utils import to_utf8
from bika.lims.workflow import doActionFor
from bika.lims.workflow import wasTransitionPerformed
from plone.app.layout.globals.interfaces import IViewView
from zope.interface import implements


class AnalysisRequestViewView(BrowserView):

    """ AR View form
        The AR fields are printed in a table, using analysisrequest_view.py
    """

    implements(IViewView)
    template = ViewPageTemplateFile("templates/analysisrequest_view.pt")
    messages = []

    def __init__(self, context, request):
        self.init__ = super(AnalysisRequestViewView, self).__init__(context,
                                                                    request)
        self.icon = self.portal_url + "/++resource++bika.lims.images/analysisrequest_big.png"
        self.messages = []

    def __call__(self):
        ar = self.context
        workflow = getToolByName(self.context, 'portal_workflow')
        if 'transition' in self.request.form:
            doActionFor(self.context, self.request.form['transition'])

        # If the analysis request has been received and hasn't been yet
        # verified yet, redirect the user to manage_results view, but only if
        # the user has privileges to Edit(Field)Results, cause otherwise she/he
        # will receive an InsufficientPrivileges error!
        if (self.request.PATH_TRANSLATED.endswith(self.context.id) and
            check_permission(EditResults, self.context) and
            check_permission(EditFieldResults, self.context) and
            wasTransitionPerformed(self.context, 'receive') and
            not wasTransitionPerformed(self.context, 'verify')):
            # Redirect to manage results view if not cancelled
            if api.get_workflow_status_of(ar, 'cancellation_state') != \
                    "cancelled":
                manage_results_url = "/".join([self.context.absolute_url(),
                                               'manage_results'])
                self.request.response.redirect(manage_results_url)
                return

        # Contacts get expanded for view
        contact = self.context.getContact()
        contacts = []
        for cc in self.context.getCCContact():
            contacts.append(cc)
        if contact in contacts:
            contacts.remove(contact)
        ccemails = []
        for cc in contacts:
            ccemails.append("%s &lt;<a href='mailto:%s'>%s</a>&gt;"
                % (cc.Title(), cc.getEmailAddress(), cc.getEmailAddress()))
        # CC Emails become mailto links
        emails = self.context.getCCEmails()
        if isinstance(emails, str):
            emails = emails and [emails, ] or []
        cc_emails = []
        cc_hrefs = []
        for cc in emails:
            cc_emails.append(cc)
            cc_hrefs.append("<a href='mailto:%s'>%s</a>" % (cc, cc))
        # render header table
        self.header_table = HeaderTableView(self.context, self.request)()
        # Create Partitions View for this ARs sample
        p = SamplePartitionsView(self.context.getSample(), self.request)
        p.show_column_toggles = False
        self.parts = p.contents_table()
        # Create Field and Lab Analyses tables
        self.tables = {}
        for poc in POINTS_OF_CAPTURE:
            if self.context.getAnalyses(getPointOfCapture=poc):
                t = self.createAnalysesView(ar,
                                 self.request,
                                 getPointOfCapture=poc,
                                 show_categories=self.context.bika_setup.getCategoriseAnalysisServices(),
                                 getRequestUID=self.context.UID())
                t.allow_edit = True
                t.form_id = "%s_analyses" % poc
                t.review_states[0]['transitions'] = [{'id': 'submit'},
                                                     {'id': 'retract'},
                                                     {'id': 'verify'}]
                t.show_workflow_action_buttons = True
                t.show_select_column = True
                if getSecurityManager().checkPermission(EditFieldResults, self.context) \
                   and poc == 'field':
                    t.review_states[0]['columns'].remove('DueDate')
                self.tables[POINTS_OF_CAPTURE.getValue(poc)] = t.contents_table()
        # Create QC Analyses View for this AR
        show_cats = self.context.bika_setup.getCategoriseAnalysisServices()
        qcview = self.createQCAnalyesView(ar,
                                self.request,
                                show_categories=show_cats)
        qcview.allow_edit = False
        qcview.show_select_column = False
        qcview.show_workflow_action_buttons = False
        qcview.form_id = "%s_qcanalyses"
        qcview.review_states[0]['transitions'] = [{'id': 'submit'},
                                                  {'id': 'retract'},
                                                  {'id': 'verify'}]
        self.qctable = qcview.contents_table()

        # Create the ResultsInterpretation by department view
        from resultsinterpretation import ARResultsInterpretationView
        self.riview = ARResultsInterpretationView(ar, self.request)

        # If a general retracted is done, rise a waring
        if workflow.getInfoFor(ar, 'review_state') == 'sample_received':
            allstatus = list()
            for analysis in ar.getAnalyses():
                status = workflow.getInfoFor(analysis.getObject(), 'review_state')
                if status not in ['retracted','to_be_verified','verified']:
                    allstatus = []
                    break
                else:
                    allstatus.append(status)
            if len(allstatus) > 0:
                message = "General Retract Done.  Submit this AR manually."
                self.addMessage(message, 'warning')

        # If is a retracted AR, show the link to child AR and show a warn msg
        if workflow.getInfoFor(ar, 'review_state') == 'invalid':
            childar = ar.getRetest() or None
            message = _('These results have been withdrawn and are '
                        'listed here for trace-ability purposes. Please follow '
                        'the link to the retest')
            if childar:
                message = (message + " %s.") % childar.getId()
            else:
                message = message + "."
            self.addMessage(message, 'warning')
        # If is an AR automatically generated due to a Retraction, show it's
        # parent AR information
        invalidated = ar.getInvalidated()
        if invalidated:
            message = _('This Analysis Request has been '
                        'generated automatically due to '
                        'the retraction of the Analysis '
                        'Request ${retracted_request_id}.',
                        mapping={'retracted_request_id': invalidated.getId()})
            self.addMessage(message, 'info')
        self.renderMessages()
        return self.template()

    def getAttachments(self):
        attachments = []
        ar_atts = self.context.getAttachment()
        analyses = self.context.getAnalyses(full_objects=True)
        for att in ar_atts:
            fsize = 0
            file = att.getAttachmentFile()
            if file:
                fsize = file.get_size()
            if fsize < 1024:
                fsize = '%s b' % fsize
            else:
                fsize = '%s Kb' % (fsize / 1024)
            attachments.append({
                'keywords': att.getAttachmentKeys(),
                'analysis': '',
                'size': fsize,
                'name': file.filename,
                'Icon': file.icon,
                'type': att.getAttachmentType().UID() if att.getAttachmentType() else '',
                'absolute_url': att.absolute_url(),
                'UID': att.UID(),
                'report_option': att.getReportOption(),
            })

        for analysis in analyses:
            an_atts = analysis.getAttachment()
            for att in an_atts:
                file = att.getAttachmentFile()
                fsize = file.get_size() if file else 0
                if fsize < 1024:
                    fsize = '%s b' % fsize
                else:
                    fsize = '%s Kb' % (fsize / 1024)
                attachments.append({
                    'keywords': att.getAttachmentKeys(),
                    'analysis': analysis.Title(),
                    'size': fsize,
                    'name': file.filename,
                    'Icon': file.icon,
                    'type': att.getAttachmentType().UID() if att.getAttachmentType() else '',
                    'absolute_url': att.absolute_url(),
                    'UID': att.UID(),
                    'report_option': att.getReportOption(),
                })
        return attachments

    def addMessage(self, message, msgtype='info'):
        self.messages.append({'message': message, 'msgtype': msgtype})

    def renderMessages(self):
        for message in self.messages:
            self.context.plone_utils.addPortalMessage(
                message['message'], message['msgtype'])

    def createAnalysesView(self, context, request, **kwargs):
        return AnalysesView(context, request, **kwargs)

    def createQCAnalyesView(self, context, request, **kwargs):
        return QCAnalysesView(context, request, **kwargs)

    def tabindex(self):
        i = 0
        while True:
            i += 1
            yield i

    def now(self):
        return DateTime()

    def getMemberDiscountApplies(self):
        client = self.context.portal_type == 'Client' and self.context or self.context.aq_parent
        return client and client.portal_type == 'Client' and client.getMemberDiscountApplies() or False

    def analysisprofiles(self):
        """ Return applicable client and Lab AnalysisProfile records
        """
        res = []
        profiles = []
        client = self.context.portal_type == 'AnalysisRequest' \
            and self.context.aq_parent or self.context
        for profile in client.objectValues("AnalysisProfile"):
            if isActive(profile):
                profiles.append((profile.Title(), profile))
        profiles.sort(lambda x, y: cmp(x[0], y[0]))
        res += profiles
        profiles = []
        for profile in self.context.bika_setup.bika_analysisprofiles.objectValues("AnalysisProfile"):
            if isActive(profile):
                lab = t(_('Lab'))
                title = to_utf8(profile.Title())
                profiles.append(("%s: %s" % (lab, title), profile))
        profiles.sort(lambda x, y: cmp(x[0], y[0]))
        res += profiles
        return res

    def artemplates(self):
        """ Return applicable client and Lab ARTemplate records
        """
        res = []
        templates = []
        client = self.context.portal_type == 'AnalysisRequest' \
            and self.context.aq_parent or self.context
        for template in client.objectValues("ARTemplate"):
            if isActive(template):
                templates.append((template.Title(), template))
        templates.sort(lambda x, y: cmp(x[0], y[0]))
        res += templates
        templates = []
        for template in self.context.bika_setup.bika_artemplates.objectValues("ARTemplate"):
            if isActive(template):
                lab = t(_('Lab'))
                title = to_utf8(template.Title())
                templates.append(("%s: %s" % (lab, title), template))
        templates.sort(lambda x, y: cmp(x[0], y[0]))
        res += templates
        return res

    def samplingdeviations(self):
        """ SamplingDeviation vocabulary for AR Add
        """
        bsc = getToolByName(self.context, 'bika_setup_catalog')
        res = [(sd.getObject().Title(), sd.getObject())
               for sd in bsc(portal_type='SamplingDeviation',
                             inactive_state='active')]
        res.sort(lambda x, y: cmp(x[0], y[0]))
        return res

    def sampleconditions(self):
        """ SampleConditions vocabulary for AR Add
        """
        bsc = getToolByName(self.context, 'bika_setup_catalog')
        res = [(sd.getObject().Title(), sd.getObject())
               for sd in bsc(portal_type='SampleConditions',
                             inactive_state='active')]
        res.sort(lambda x, y: cmp(x[0], y[0]))
        return res

    def containertypes(self):
        """ DefaultContainerType vocabulary for AR Add
        """
        bsc = getToolByName(self.context, 'bika_setup_catalog')
        res = [(o.getObject().Title(), o.getObject())
               for o in bsc(portal_type='ContainerType')]
        res.sort(lambda x, y: cmp(x[0], y[0]))
        return res

    def SelectedServices(self):
        """ return information about services currently selected in the
            context AR.
            [[PointOfCapture, category uid, service uid],
             [PointOfCapture, category uid, service uid], ...]
        """
        bac = getToolByName(self.context, 'bika_analysis_catalog')
        res = []
        for analysis in bac(portal_type="Analysis",
                            getRequestUID=self.context.UID()):
            analysis = analysis.getObject()
            res.append([analysis.getPointOfCapture(),
                        analysis.getCategoryUID(),
                        analysis.getServiceUID()])
        return res

    def getRestrictedCategories(self):
        # we are in portal_factory AR context right now
        parent = self.context.aq_parent
        if hasattr(parent, "getRestrictedCategories"):
            return parent.getRestrictedCategories()
        return []

    def Categories(self):
        """ Dictionary keys: poc
            Dictionary values: (Category UID,category Title)
        """
        bsc = getToolByName(self.context, 'bika_setup_catalog')
        cats = {}
        restricted = [u.UID() for u in self.getRestrictedCategories()]
        for service in bsc(portal_type="AnalysisService",
                           inactive_state='active'):
            cat = (service.getCategoryUID, service.getCategoryTitle)
            if restricted and cat[0] not in restricted:
                continue
            poc = service.getPointOfCapture
            if poc not in cats:
                cats[poc] = []
            if cat not in cats[poc]:
                cats[poc].append(cat)
        return cats

    def getDefaultCategories(self):
        # we are in portal_factory AR context right now
        parent = self.context.aq_parent
        if hasattr(parent, "getDefaultCategories"):
            return parent.getDefaultCategories()
        return []

    def DefaultCategories(self):
        """ Used in AR add context, to return list of UIDs for
        automatically-expanded categories.
        """
        cats = self.getDefaultCategories()
        return [cat.UID() for cat in cats]

    def getDefaultSpec(self):

        """ Returns 'lab' or 'client' to set the initial value of the
            specification radios
        """
        mt = getToolByName(self.context, 'portal_membership')
        pg = getToolByName(self.context, 'portal_groups')
        member = mt.getAuthenticatedMember()
        member_groups = [pg.getGroupById(group.id).getGroupName()
                         for group in pg.getGroupsByUserId(member.id)]
        default_spec = ('Clients' in member_groups) and 'client' or 'lab'
        return default_spec

    def getAnalysisProfileTitle(self):
        """Grab the context's current AnalysisProfile Title if any
        """
        return self.context.getProfile() and \
               self.context.getProfile().Title() or ''

    def getARTemplateTitle(self):
        """Grab the context's current ARTemplate Title if any
        """
        return self.context.getTemplate() and \
               self.context.getTemplate().Title() or ''

    def get_requested_analyses(self):
        #
        # title=Get requested analyses
        #
        result = []
        cats = {}
        for analysis in self.context.getAnalyses(full_objects=True):
            if analysis.review_state == 'not_requested':
                continue
            category_name = analysis.getCategoryTitle()
            if not category_name in cats:
                cats[category_name] = {}
            cats[category_name][analysis.Title()] = analysis
        cat_keys = cats.keys()
        cat_keys.sort(lambda x, y: cmp(x.lower(), y.lower()))
        for cat_key in cat_keys:
            analyses = cats[cat_key]
            analysis_keys = analyses.keys()
            analysis_keys.sort(lambda x, y: cmp(x.lower(), y.lower()))
            for analysis_key in analysis_keys:
                result.append(analyses[analysis_key])
        return result

    def get_analyses_not_requested(self):
        #
        # title=Get analyses which have not been requested by the client
        #

        result = []
        for analysis in self.context.getAnalyses():
            if analysis.review_state == 'not_requested':
                result.append(analysis)
        return result

    def get_analysisrequest_verifier(self, analysisrequest):
        """Get the name of the member who last verified this AR
        """
        wtool = getToolByName(self.context, 'portal_workflow')
        mtool = getToolByName(self.context, 'portal_membership')
        verifier = None
        try:
            review_history = wtool.getInfoFor(analysisrequest, 'review_history')
        except:
            return 'access denied'
        review_history = [review for review in review_history if review.get('action', '')]
        if not review_history:
            return 'no history'
        for items in review_history:
            action = items.get('action')
            if action != 'verify':
                continue
            actor = items.get('actor')
            member = mtool.getMemberById(actor)
            if not member:
                verifier = actor
                continue
            verifier = member.getProperty('fullname')
            if verifier is None or verifier == '':
                verifier = actor
        return verifier

    def get_custom_fields(self):
        """ Returns a dictionary with custom fields to be rendered after
            header_table with this structure:
            {<fieldid>:{title:<title>, value:<html>}
        """
        custom = {}
        ar = self.context
        workflow = getToolByName(self.context, 'portal_workflow')
        # If is a retracted AR, show the link to child AR and show a warn msg
        if workflow.getInfoFor(ar, 'review_state') == 'invalid':
            childar = ar.getRetest() or None
            anchor = childar and ("<a href='%s'>%s</a>" % (childar.absolute_url(), childar.getId())) or None
            if anchor:
                custom['ChildAR'] = {
                    'title': t(_("AR for retested results")),
                    'value': anchor
                }
        # If is an AR automatically generated due to a Retraction, show it's
        # parent AR information
        invalidated = ar.getInvalidated()
        if invalidated:
            anchor = "<a href='%s'>%s</a>" % (invalidated.absolute_url(),
                                              invalidated.getId())
            custom['ParentAR'] = {
                'title': t(_("Invalid AR retested")),
                'value': anchor
            }
        return custom


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 AccessControl import getSecurityManager
9		from DateTime import DateTime
10		from Products.CMFCore.utils import getToolByName
11		from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
12		from bika.lims import bikaMessageFactory as _
13		from bika.lims import api
14		from bika.lims.browser import BrowserView
15		from bika.lims.browser.analyses import AnalysesView
16		from bika.lims.browser.analyses import QCAnalysesView
17		from bika.lims.browser.header_table import HeaderTableView
18		from bika.lims.browser.sample import SamplePartitionsView
19		from bika.lims.config import POINTS_OF_CAPTURE
20		from bika.lims.permissions import *
21		from bika.lims.utils import isActive
22		from bika.lims.utils import t, check_permission
23		from bika.lims.utils import to_utf8
24		from bika.lims.workflow import doActionFor
25		from bika.lims.workflow import wasTransitionPerformed
26		from plone.app.layout.globals.interfaces import IViewView
27		from zope.interface import implements
28
29
30		class AnalysisRequestViewView(BrowserView):
31
32		""" AR View form
33		The AR fields are printed in a table, using analysisrequest_view.py
34		"""
35
36		implements(IViewView)
37		template = ViewPageTemplateFile("templates/analysisrequest_view.pt")
38		messages = []
39
40		def __init__(self, context, request):
41		self.init__ = super(AnalysisRequestViewView, self).__init__(context,
42		request)
43		self.icon = self.portal_url + "/++resource++bika.lims.images/analysisrequest_big.png"
44		self.messages = []
45
46		def __call__(self):
47		ar = self.context
48		workflow = getToolByName(self.context, 'portal_workflow')
49		if 'transition' in self.request.form:
50		doActionFor(self.context, self.request.form['transition'])
51
52		# If the analysis request has been received and hasn't been yet
53		# verified yet, redirect the user to manage_results view, but only if
54		# the user has privileges to Edit(Field)Results, cause otherwise she/he
55		# will receive an InsufficientPrivileges error!
56		if (self.request.PATH_TRANSLATED.endswith(self.context.id) and
57		check_permission(EditResults, self.context) and
58		check_permission(EditFieldResults, self.context) and
59		wasTransitionPerformed(self.context, 'receive') and
60		not wasTransitionPerformed(self.context, 'verify')):
61		# Redirect to manage results view if not cancelled
62		if api.get_workflow_status_of(ar, 'cancellation_state') != \
63		"cancelled":
64		manage_results_url = "/".join([self.context.absolute_url(),
65		'manage_results'])
66		self.request.response.redirect(manage_results_url)
67		return
68
69		# Contacts get expanded for view
70		contact = self.context.getContact()
71		contacts = []
72		for cc in self.context.getCCContact():
73		contacts.append(cc)
74		if contact in contacts:
75		contacts.remove(contact)
76		ccemails = []
77		for cc in contacts:
78		ccemails.append("%s <<a href='mailto:%s'>%s</a>>"
79		% (cc.Title(), cc.getEmailAddress(), cc.getEmailAddress()))
80		# CC Emails become mailto links
81		emails = self.context.getCCEmails()
82		if isinstance(emails, str):
83		emails = emails and [emails, ] or []
84		cc_emails = []
85		cc_hrefs = []
86		for cc in emails:
87		cc_emails.append(cc)
88		cc_hrefs.append("<a href='mailto:%s'>%s</a>" % (cc, cc))
89		# render header table
90		self.header_table = HeaderTableView(self.context, self.request)()
91		# Create Partitions View for this ARs sample
92		p = SamplePartitionsView(self.context.getSample(), self.request)
93		p.show_column_toggles = False
94		self.parts = p.contents_table()
95		# Create Field and Lab Analyses tables
96		self.tables = {}
97		for poc in POINTS_OF_CAPTURE:
98		if self.context.getAnalyses(getPointOfCapture=poc):
99		t = self.createAnalysesView(ar,
100		self.request,
101		getPointOfCapture=poc,
102		show_categories=self.context.bika_setup.getCategoriseAnalysisServices(),
103		getRequestUID=self.context.UID())
104		t.allow_edit = True
105		t.form_id = "%s_analyses" % poc
106		t.review_states[0]['transitions'] = [{'id': 'submit'},
107		{'id': 'retract'},
108		{'id': 'verify'}]
109		t.show_workflow_action_buttons = True
110		t.show_select_column = True
111		if getSecurityManager().checkPermission(EditFieldResults, self.context) \
112		and poc == 'field':
113		t.review_states[0]['columns'].remove('DueDate')
114		self.tables[POINTS_OF_CAPTURE.getValue(poc)] = t.contents_table()
115		# Create QC Analyses View for this AR
116		show_cats = self.context.bika_setup.getCategoriseAnalysisServices()
117		qcview = self.createQCAnalyesView(ar,
118		self.request,
119		show_categories=show_cats)
120		qcview.allow_edit = False
121		qcview.show_select_column = False
122		qcview.show_workflow_action_buttons = False
123		qcview.form_id = "%s_qcanalyses"
124		qcview.review_states[0]['transitions'] = [{'id': 'submit'},
125		{'id': 'retract'},
126		{'id': 'verify'}]
127		self.qctable = qcview.contents_table()
128
129		# Create the ResultsInterpretation by department view
130		from resultsinterpretation import ARResultsInterpretationView
131		self.riview = ARResultsInterpretationView(ar, self.request)
132
133		# If a general retracted is done, rise a waring
134		if workflow.getInfoFor(ar, 'review_state') == 'sample_received':
135		allstatus = list()
136		for analysis in ar.getAnalyses():
137		status = workflow.getInfoFor(analysis.getObject(), 'review_state')
138		if status not in ['retracted','to_be_verified','verified']:
139		allstatus = []
140		break
141		else:
142		allstatus.append(status)
143		if len(allstatus) > 0:
144		message = "General Retract Done. Submit this AR manually."
145		self.addMessage(message, 'warning')
146
147		# If is a retracted AR, show the link to child AR and show a warn msg
148		if workflow.getInfoFor(ar, 'review_state') == 'invalid':
149		childar = ar.getRetest() or None
150		message = _('These results have been withdrawn and are '
151		'listed here for trace-ability purposes. Please follow '
152		'the link to the retest')
153		if childar:
154		message = (message + " %s.") % childar.getId()
155		else:
156		message = message + "."
157		self.addMessage(message, 'warning')
158		# If is an AR automatically generated due to a Retraction, show it's
159		# parent AR information
160		invalidated = ar.getInvalidated()
161		if invalidated:
162		message = _('This Analysis Request has been '
163		'generated automatically due to '
164		'the retraction of the Analysis '
165		'Request ${retracted_request_id}.',
166		mapping={'retracted_request_id': invalidated.getId()})
167		self.addMessage(message, 'info')
168		self.renderMessages()
169		return self.template()
170
171		def getAttachments(self):
172		attachments = []
173		ar_atts = self.context.getAttachment()
174		analyses = self.context.getAnalyses(full_objects=True)
175		for att in ar_atts:
176		fsize = 0
177		file = att.getAttachmentFile()
178		if file:
179		fsize = file.get_size()
180		if fsize < 1024:
181		fsize = '%s b' % fsize
182		else:
183		fsize = '%s Kb' % (fsize / 1024)
184		attachments.append({
185		'keywords': att.getAttachmentKeys(),
186		'analysis': '',
187		'size': fsize,
188		'name': file.filename,
189		'Icon': file.icon,
190		'type': att.getAttachmentType().UID() if att.getAttachmentType() else '',
191		'absolute_url': att.absolute_url(),
192		'UID': att.UID(),
193		'report_option': att.getReportOption(),
194		})
195
196		for analysis in analyses:
197		an_atts = analysis.getAttachment()
198		for att in an_atts:
199		file = att.getAttachmentFile()
200		fsize = file.get_size() if file else 0
201		if fsize < 1024:
202		fsize = '%s b' % fsize
203		else:
204		fsize = '%s Kb' % (fsize / 1024)
205		attachments.append({
206		'keywords': att.getAttachmentKeys(),
207		'analysis': analysis.Title(),
208		'size': fsize,
209		'name': file.filename,
210		'Icon': file.icon,
211		'type': att.getAttachmentType().UID() if att.getAttachmentType() else '',
212		'absolute_url': att.absolute_url(),
213		'UID': att.UID(),
214		'report_option': att.getReportOption(),
215		})
216		return attachments
217
218		def addMessage(self, message, msgtype='info'):
219		self.messages.append({'message': message, 'msgtype': msgtype})
220
221		def renderMessages(self):
222		for message in self.messages:
223		self.context.plone_utils.addPortalMessage(
224		message['message'], message['msgtype'])
225
226		def createAnalysesView(self, context, request, **kwargs):
227		return AnalysesView(context, request, **kwargs)
228
229		def createQCAnalyesView(self, context, request, **kwargs):
230		return QCAnalysesView(context, request, **kwargs)
231
232		def tabindex(self):
233		i = 0
234		while True:
235		i += 1
236		yield i
237
238		def now(self):
239		return DateTime()
240
241		def getMemberDiscountApplies(self):
242		client = self.context.portal_type == 'Client' and self.context or self.context.aq_parent
243		return client and client.portal_type == 'Client' and client.getMemberDiscountApplies() or False
244
245		def analysisprofiles(self):
246		""" Return applicable client and Lab AnalysisProfile records
247		"""
248		res = []
249		profiles = []
250		client = self.context.portal_type == 'AnalysisRequest' \
251		and self.context.aq_parent or self.context
252		for profile in client.objectValues("AnalysisProfile"):
253		if isActive(profile):
254		profiles.append((profile.Title(), profile))
255		profiles.sort(lambda x, y: cmp(x[0], y[0]))
256		res += profiles
257		profiles = []
258		for profile in self.context.bika_setup.bika_analysisprofiles.objectValues("AnalysisProfile"):
259		if isActive(profile):
260		lab = t(_('Lab'))
261		title = to_utf8(profile.Title())
262		profiles.append(("%s: %s" % (lab, title), profile))
263		profiles.sort(lambda x, y: cmp(x[0], y[0]))
264		res += profiles
265		return res
266
267		def artemplates(self):
268		""" Return applicable client and Lab ARTemplate records
269		"""
270		res = []
271		templates = []
272		client = self.context.portal_type == 'AnalysisRequest' \
273		and self.context.aq_parent or self.context
274		for template in client.objectValues("ARTemplate"):
275		if isActive(template):
276		templates.append((template.Title(), template))
277		templates.sort(lambda x, y: cmp(x[0], y[0]))
278		res += templates
279		templates = []
280		for template in self.context.bika_setup.bika_artemplates.objectValues("ARTemplate"):
281		if isActive(template):
282		lab = t(_('Lab'))
283		title = to_utf8(template.Title())
284		templates.append(("%s: %s" % (lab, title), template))
285		templates.sort(lambda x, y: cmp(x[0], y[0]))
286		res += templates
287		return res
288
289		def samplingdeviations(self):
290		""" SamplingDeviation vocabulary for AR Add
291		"""
292		bsc = getToolByName(self.context, 'bika_setup_catalog')
293		res = [(sd.getObject().Title(), sd.getObject())
294		for sd in bsc(portal_type='SamplingDeviation',
295		inactive_state='active')]
296		res.sort(lambda x, y: cmp(x[0], y[0]))
297		return res
298
299		def sampleconditions(self):
300		""" SampleConditions vocabulary for AR Add
301		"""
302		bsc = getToolByName(self.context, 'bika_setup_catalog')
303		res = [(sd.getObject().Title(), sd.getObject())
304		for sd in bsc(portal_type='SampleConditions',
305		inactive_state='active')]
306		res.sort(lambda x, y: cmp(x[0], y[0]))
307		return res
308
309		def containertypes(self):
310		""" DefaultContainerType vocabulary for AR Add
311		"""
312		bsc = getToolByName(self.context, 'bika_setup_catalog')
313		res = [(o.getObject().Title(), o.getObject())
314		for o in bsc(portal_type='ContainerType')]
315		res.sort(lambda x, y: cmp(x[0], y[0]))
316		return res
317
318		def SelectedServices(self):
319		""" return information about services currently selected in the
320		context AR.
321		[[PointOfCapture, category uid, service uid],
322		[PointOfCapture, category uid, service uid], ...]
323		"""
324		bac = getToolByName(self.context, 'bika_analysis_catalog')
325		res = []
326		for analysis in bac(portal_type="Analysis",
327		getRequestUID=self.context.UID()):
328		analysis = analysis.getObject()
329		res.append([analysis.getPointOfCapture(),
330		analysis.getCategoryUID(),
331		analysis.getServiceUID()])
332		return res
333
334		def getRestrictedCategories(self):
335		# we are in portal_factory AR context right now
336		parent = self.context.aq_parent
337		if hasattr(parent, "getRestrictedCategories"):
338		return parent.getRestrictedCategories()
339		return []
340
341		def Categories(self):
342		""" Dictionary keys: poc
343		Dictionary values: (Category UID,category Title)
344		"""
345		bsc = getToolByName(self.context, 'bika_setup_catalog')
346		cats = {}
347		restricted = [u.UID() for u in self.getRestrictedCategories()]
348		for service in bsc(portal_type="AnalysisService",
349		inactive_state='active'):
350		cat = (service.getCategoryUID, service.getCategoryTitle)
351		if restricted and cat[0] not in restricted:
352		continue
353		poc = service.getPointOfCapture
354		if poc not in cats:
355		cats[poc] = []
356		if cat not in cats[poc]:
357		cats[poc].append(cat)
358		return cats
359
360		def getDefaultCategories(self):
361		# we are in portal_factory AR context right now
362		parent = self.context.aq_parent
363		if hasattr(parent, "getDefaultCategories"):
364		return parent.getDefaultCategories()
365		return []
366
367		def DefaultCategories(self):
368		""" Used in AR add context, to return list of UIDs for
369		automatically-expanded categories.
370		"""
371		cats = self.getDefaultCategories()
372		return [cat.UID() for cat in cats]
373
374	View Code Duplication	def getDefaultSpec(self):
		0 ignored issues – show Duplication introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
375		""" Returns 'lab' or 'client' to set the initial value of the
376		specification radios
377		"""
378		mt = getToolByName(self.context, 'portal_membership')
379		pg = getToolByName(self.context, 'portal_groups')
380		member = mt.getAuthenticatedMember()
381		member_groups = [pg.getGroupById(group.id).getGroupName()
382		for group in pg.getGroupsByUserId(member.id)]
383		default_spec = ('Clients' in member_groups) and 'client' or 'lab'
384		return default_spec
385
386		def getAnalysisProfileTitle(self):
387		"""Grab the context's current AnalysisProfile Title if any
388		"""
389		return self.context.getProfile() and \
390		self.context.getProfile().Title() or ''
391
392		def getARTemplateTitle(self):
393		"""Grab the context's current ARTemplate Title if any
394		"""
395		return self.context.getTemplate() and \
396		self.context.getTemplate().Title() or ''
397
398		def get_requested_analyses(self):
399		#
400		# title=Get requested analyses
401		#
402		result = []
403		cats = {}
404		for analysis in self.context.getAnalyses(full_objects=True):
405		if analysis.review_state == 'not_requested':
406		continue
407		category_name = analysis.getCategoryTitle()
408		if not category_name in cats:
409		cats[category_name] = {}
410		cats[category_name][analysis.Title()] = analysis
411		cat_keys = cats.keys()
412		cat_keys.sort(lambda x, y: cmp(x.lower(), y.lower()))
413		for cat_key in cat_keys:
414		analyses = cats[cat_key]
415		analysis_keys = analyses.keys()
416		analysis_keys.sort(lambda x, y: cmp(x.lower(), y.lower()))
417		for analysis_key in analysis_keys:
418		result.append(analyses[analysis_key])
419		return result
420
421		def get_analyses_not_requested(self):
422		#
423		# title=Get analyses which have not been requested by the client
424		#
425
426		result = []
427		for analysis in self.context.getAnalyses():
428		if analysis.review_state == 'not_requested':
429		result.append(analysis)
430		return result
431
432		def get_analysisrequest_verifier(self, analysisrequest):
433		"""Get the name of the member who last verified this AR
434		"""
435		wtool = getToolByName(self.context, 'portal_workflow')
436		mtool = getToolByName(self.context, 'portal_membership')
437		verifier = None
438		try:
439		review_history = wtool.getInfoFor(analysisrequest, 'review_history')
440		except:
441		return 'access denied'
442		review_history = [review for review in review_history if review.get('action', '')]
443		if not review_history:
444		return 'no history'
445		for items in review_history:
446		action = items.get('action')
447		if action != 'verify':
448		continue
449		actor = items.get('actor')
450		member = mtool.getMemberById(actor)
451		if not member:
452		verifier = actor
453		continue
454		verifier = member.getProperty('fullname')
455		if verifier is None or verifier == '':
456		verifier = actor
457		return verifier
458
459		def get_custom_fields(self):
460		""" Returns a dictionary with custom fields to be rendered after
461		header_table with this structure:
462		{<fieldid>:{title:<title>, value:<html>}
463		"""
464		custom = {}
465		ar = self.context
466		workflow = getToolByName(self.context, 'portal_workflow')
467		# If is a retracted AR, show the link to child AR and show a warn msg
468		if workflow.getInfoFor(ar, 'review_state') == 'invalid':
469		childar = ar.getRetest() or None
470		anchor = childar and ("<a href='%s'>%s</a>" % (childar.absolute_url(), childar.getId())) or None
471		if anchor:
472		custom['ChildAR'] = {
473		'title': t(_("AR for retested results")),
474		'value': anchor
475		}
476		# If is an AR automatically generated due to a Retraction, show it's
477		# parent AR information
478		invalidated = ar.getInvalidated()
479		if invalidated:
480		anchor = "<a href='%s'>%s</a>" % (invalidated.absolute_url(),
481		invalidated.getId())
482		custom['ParentAR'] = {
483		'title': t(_("Invalid AR retested")),
484		'value': anchor
485		}
486		return custom
487

senaite / senaite.core

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

AnalysisRequestViewView.get_custom_fields() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like