1
|
|
|
# -*- coding: utf-8 -*- |
2
|
|
|
# |
3
|
|
|
# This file is part of SENAITE.CORE. |
4
|
|
|
# |
5
|
|
|
# SENAITE.CORE is free software: you can redistribute it and/or modify it under |
6
|
|
|
# the terms of the GNU General Public License as published by the Free Software |
7
|
|
|
# Foundation, version 2. |
8
|
|
|
# |
9
|
|
|
# This program is distributed in the hope that it will be useful, but WITHOUT |
10
|
|
|
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
11
|
|
|
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more |
12
|
|
|
# details. |
13
|
|
|
# |
14
|
|
|
# You should have received a copy of the GNU General Public License along with |
15
|
|
|
# this program; if not, write to the Free Software Foundation, Inc., 51 |
16
|
|
|
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. |
17
|
|
|
# |
18
|
|
|
# Copyright 2018-2025 by it's authors. |
19
|
|
|
# Some rights reserved, see README and LICENSE. |
20
|
|
|
|
21
|
|
|
import mimetypes |
22
|
|
|
import os |
23
|
|
|
import re |
24
|
|
|
import tempfile |
25
|
|
|
from email import Encoders |
26
|
|
|
from email.MIMEBase import MIMEBase |
27
|
|
|
from time import time |
28
|
|
|
|
29
|
|
|
import six |
30
|
|
|
from AccessControl import ModuleSecurityInfo |
31
|
|
|
from AccessControl import allow_module |
32
|
|
|
from AccessControl import getSecurityManager |
33
|
|
|
from Acquisition import aq_inner |
34
|
|
|
from Acquisition import aq_parent |
35
|
|
|
from bika.lims import api |
36
|
|
|
from bika.lims import logger |
37
|
|
|
from bika.lims.browser import BrowserView |
38
|
|
|
from bika.lims.interfaces import IClient |
39
|
|
|
from bika.lims.interfaces import IClientAwareMixin |
40
|
|
|
from DateTime import DateTime |
41
|
|
|
from plone.protect.utils import addTokenToUrl |
42
|
|
|
from plone.registry.interfaces import IRegistry |
43
|
|
|
from plone.subrequest import subrequest |
44
|
|
|
from Products.Archetypes.interfaces.field import IComputedField |
45
|
|
|
from Products.Archetypes.public import DisplayList |
46
|
|
|
from Products.CMFCore.utils import getToolByName |
47
|
|
|
from Products.CMFCore.WorkflowCore import WorkflowException |
48
|
|
|
from Products.CMFPlone.utils import safe_unicode |
49
|
|
|
from Products.DCWorkflow.events import AfterTransitionEvent |
50
|
|
|
from Products.DCWorkflow.events import BeforeTransitionEvent |
51
|
|
|
from senaite.core.p3compat import cmp |
52
|
|
|
from six.moves.urllib.request import urlopen |
53
|
|
|
from weasyprint import CSS |
54
|
|
|
from weasyprint import HTML |
55
|
|
|
from weasyprint import default_url_fetcher |
56
|
|
|
from zope.component import queryUtility |
57
|
|
|
from zope.event import notify |
58
|
|
|
from zope.i18n.locales import locales |
59
|
|
|
|
60
|
|
|
ModuleSecurityInfo('email.Utils').declarePublic('formataddr') |
61
|
|
|
allow_module('csv') |
62
|
|
|
|
63
|
|
|
|
64
|
|
|
def to_utf8(text): |
65
|
|
|
if text is None: |
66
|
|
|
text = '' |
67
|
|
|
unicode_obj = safe_unicode(text) |
68
|
|
|
# If it receives a dictionary or list, it will not work |
69
|
|
|
if isinstance(unicode_obj, unicode): |
|
|
|
|
70
|
|
|
return unicode_obj.encode('utf-8') |
71
|
|
|
return unicode_obj |
72
|
|
|
|
73
|
|
|
|
74
|
|
|
def to_unicode(text): |
75
|
|
|
if text is None: |
76
|
|
|
text = '' |
77
|
|
|
return safe_unicode(text) |
78
|
|
|
|
79
|
|
|
|
80
|
|
|
def t(i18n_msg): |
81
|
|
|
"""Safely translate and convert to UTF8, any zope i18n msgid returned from |
82
|
|
|
a bikaMessageFactory _ |
83
|
|
|
""" |
84
|
|
|
# cannot use bika.lims.deprecated (circular dependencies) |
85
|
|
|
import warnings |
86
|
|
|
warnings.simplefilter("always", DeprecationWarning) |
87
|
|
|
warn = "Deprecated: use senaite.core.i18n.translate instead" |
88
|
|
|
warnings.warn(warn, category=DeprecationWarning, stacklevel=2) |
89
|
|
|
warnings.simplefilter("default", DeprecationWarning) |
90
|
|
|
|
91
|
|
|
# prevent circular dependencies |
92
|
|
|
from senaite.core.i18n import translate |
93
|
|
|
return translate(i18n_msg) |
94
|
|
|
|
95
|
|
|
|
96
|
|
|
# Wrapper for PortalTransport's sendmail - don't know why there sendmail |
97
|
|
|
# method is marked private |
98
|
|
|
ModuleSecurityInfo('Products.bika.utils').declarePublic('sendmail') |
99
|
|
|
# Protected( Publish, 'sendmail') |
100
|
|
|
|
101
|
|
|
|
102
|
|
|
def sendmail(portal, from_addr, to_addrs, msg): |
103
|
|
|
mailspool = portal.portal_mailspool |
104
|
|
|
mailspool.sendmail(from_addr, to_addrs, msg) |
105
|
|
|
|
106
|
|
|
|
107
|
|
|
class js_log(BrowserView): |
108
|
|
|
|
109
|
|
|
def __call__(self, message): |
110
|
|
|
"""Javascript sends a string for us to place into the log. |
111
|
|
|
""" |
112
|
|
|
self.logger.info(message) |
113
|
|
|
|
114
|
|
|
|
115
|
|
|
class js_err(BrowserView): |
116
|
|
|
|
117
|
|
|
def __call__(self, message): |
118
|
|
|
"""Javascript sends a string for us to place into the error log |
119
|
|
|
""" |
120
|
|
|
self.logger.error(message) |
121
|
|
|
|
122
|
|
|
|
123
|
|
|
class js_warn(BrowserView): |
124
|
|
|
|
125
|
|
|
def __call__(self, message): |
126
|
|
|
"""Javascript sends a string for us to place into the warn log |
127
|
|
|
""" |
128
|
|
|
self.logger.warning(message) |
129
|
|
|
|
130
|
|
|
|
131
|
|
|
def getUsers(context, roles, allow_empty=True): |
132
|
|
|
""" Present a DisplayList containing users in the specified |
133
|
|
|
list of roles |
134
|
|
|
""" |
135
|
|
|
mtool = getToolByName(context, 'portal_membership') |
136
|
|
|
pairs = allow_empty and [['', '']] or [] |
137
|
|
|
users = mtool.searchForMembers(roles=roles) |
138
|
|
|
for user in users: |
139
|
|
|
uid = user.getId() |
140
|
|
|
fullname = user.getProperty('fullname') |
141
|
|
|
if not fullname: |
142
|
|
|
fullname = uid |
143
|
|
|
pairs.append((uid, fullname)) |
144
|
|
|
pairs.sort(lambda x, y: cmp(x[1], y[1])) |
145
|
|
|
return DisplayList(pairs) |
146
|
|
|
|
147
|
|
|
|
148
|
|
|
def formatDateQuery(context, date_id): |
149
|
|
|
""" Obtain and reformat the from and to dates |
150
|
|
|
into a date query construct |
151
|
|
|
""" |
152
|
|
|
from_date = context.REQUEST.get('%s_fromdate' % date_id, None) |
153
|
|
|
if from_date: |
154
|
|
|
from_date = from_date + ' 00:00' |
155
|
|
|
to_date = context.REQUEST.get('%s_todate' % date_id, None) |
156
|
|
|
if to_date: |
157
|
|
|
to_date = to_date + ' 23:59' |
158
|
|
|
|
159
|
|
|
date_query = {} |
160
|
|
|
if from_date and to_date: |
161
|
|
|
date_query = {'query': [from_date, to_date], |
162
|
|
|
'range': 'min:max'} |
163
|
|
|
elif from_date or to_date: |
164
|
|
|
date_query = {'query': from_date or to_date, |
165
|
|
|
'range': from_date and 'min' or 'max'} |
166
|
|
|
|
167
|
|
|
return date_query |
168
|
|
|
|
169
|
|
|
|
170
|
|
|
def formatDateParms(context, date_id): |
171
|
|
|
""" Obtain and reformat the from and to dates |
172
|
|
|
into a printable date parameter construct |
173
|
|
|
""" |
174
|
|
|
from_date = context.REQUEST.get('%s_fromdate' % date_id, None) |
175
|
|
|
to_date = context.REQUEST.get('%s_todate' % date_id, None) |
176
|
|
|
|
177
|
|
|
date_parms = {} |
178
|
|
|
if from_date and to_date: |
179
|
|
|
date_parms = 'from %s to %s' % (from_date, to_date) |
180
|
|
|
elif from_date: |
181
|
|
|
date_parms = 'from %s' % (from_date) |
182
|
|
|
elif to_date: |
183
|
|
|
date_parms = 'to %s' % (to_date) |
184
|
|
|
|
185
|
|
|
return date_parms |
186
|
|
|
|
187
|
|
|
|
188
|
|
|
def formatDecimalMark(value, decimalmark='.'): |
189
|
|
|
""" |
190
|
|
|
Dummy method to replace decimal mark from an input string. |
191
|
|
|
Assumes that 'value' uses '.' as decimal mark and ',' as |
192
|
|
|
thousand mark. |
193
|
|
|
::value:: is a string |
194
|
|
|
::returns:: is a string with the decimal mark if needed |
195
|
|
|
""" |
196
|
|
|
# We have to consider the possibility of working with decimals such as |
197
|
|
|
# X.000 where those decimals are important because of the precission |
198
|
|
|
# and significant digits matters |
199
|
|
|
# Using 'float' the system delete the extre desimals with 0 as a value |
200
|
|
|
# Example: float(2.00) -> 2.0 |
201
|
|
|
# So we have to save the decimal length, this is one reason we are usnig |
202
|
|
|
# strings for results |
203
|
|
|
rawval = str(value) |
204
|
|
|
try: |
205
|
|
|
return decimalmark.join(rawval.split('.')) |
206
|
|
|
except Exception: |
207
|
|
|
return rawval |
208
|
|
|
|
209
|
|
|
|
210
|
|
|
# encode_header function copied from roundup's rfc2822 package. |
211
|
|
|
hqre = re.compile(r'^[A-z0-9!"#$%%&\'()*+,-./:;<=>?@\[\]^_`{|}~ ]+$') |
212
|
|
|
|
213
|
|
|
ModuleSecurityInfo('Products.bika.utils').declarePublic('encode_header') |
214
|
|
|
|
215
|
|
|
|
216
|
|
|
def encode_header(header, charset='utf-8'): |
217
|
|
|
""" Will encode in quoted-printable encoding only if header |
218
|
|
|
contains non latin characters |
219
|
|
|
""" |
220
|
|
|
|
221
|
|
|
# Return empty headers unchanged |
222
|
|
|
if not header: |
223
|
|
|
return header |
224
|
|
|
|
225
|
|
|
# return plain header if it does not contain non-ascii characters |
226
|
|
|
if hqre.match(header): |
227
|
|
|
return header |
228
|
|
|
|
229
|
|
|
quoted = '' |
230
|
|
|
# max_encoded = 76 - len(charset) - 7 |
231
|
|
|
for c in header: |
232
|
|
|
# Space may be represented as _ instead of =20 for readability |
233
|
|
|
if c == ' ': |
234
|
|
|
quoted += '_' |
235
|
|
|
# These characters can be included verbatim |
236
|
|
|
elif hqre.match(c): |
237
|
|
|
quoted += c |
238
|
|
|
# Otherwise, replace with hex value like =E2 |
239
|
|
|
else: |
240
|
|
|
quoted += "=%02X" % ord(c) |
241
|
|
|
|
242
|
|
|
return '=?%s?q?%s?=' % (charset, quoted) |
243
|
|
|
|
244
|
|
|
|
245
|
|
|
def zero_fill(matchobj): |
246
|
|
|
return matchobj.group().zfill(8) |
247
|
|
|
|
248
|
|
|
|
249
|
|
|
num_sort_regex = re.compile(r'\d+') |
250
|
|
|
|
251
|
|
|
ModuleSecurityInfo('Products.bika.utils').declarePublic('sortable_title') |
252
|
|
|
|
253
|
|
|
|
254
|
|
|
def sortable_title(portal, title): |
255
|
|
|
"""Convert title to sortable title |
256
|
|
|
""" |
257
|
|
|
if not title: |
258
|
|
|
return '' |
259
|
|
|
|
260
|
|
|
def_charset = portal.plone_utils.getSiteEncoding() |
261
|
|
|
sortabletitle = str(title.lower().strip()) |
262
|
|
|
# Replace numbers with zero filled numbers |
263
|
|
|
sortabletitle = num_sort_regex.sub(zero_fill, sortabletitle) |
264
|
|
|
# Truncate to prevent bloat |
265
|
|
|
for charset in [def_charset, 'latin-1', 'utf-8']: |
266
|
|
|
try: |
267
|
|
|
sortabletitle = safe_unicode(sortabletitle, charset)[:30] |
268
|
|
|
sortabletitle = sortabletitle.encode(def_charset or 'utf-8') |
269
|
|
|
break |
270
|
|
|
except UnicodeError: |
271
|
|
|
pass |
272
|
|
|
except TypeError: |
273
|
|
|
# If we get a TypeError if we already have a unicode string |
274
|
|
|
sortabletitle = sortabletitle[:30] |
275
|
|
|
break |
276
|
|
|
return sortabletitle |
277
|
|
|
|
278
|
|
|
|
279
|
|
|
# TODO Remove this function |
280
|
|
|
def logged_in_client(context, member=None): |
281
|
|
|
return api.get_current_client() |
282
|
|
|
|
283
|
|
|
|
284
|
|
|
def changeWorkflowState(content, wf_id, state_id, trigger_events=False, **kw): |
285
|
|
|
"""Change the workflow state of an object |
286
|
|
|
@param content: Content obj which state will be changed |
287
|
|
|
@param state_id: name of the state to put on content |
288
|
|
|
@param kw: change the values of same name of the state mapping |
289
|
|
|
@return: True if succeed. Otherwise, False |
290
|
|
|
""" |
291
|
|
|
portal_workflow = api.get_tool("portal_workflow") |
292
|
|
|
workflow = portal_workflow.getWorkflowById(wf_id) |
293
|
|
|
if not workflow: |
294
|
|
|
logger.error("%s: Cannot find workflow id %s" % (content, wf_id)) |
295
|
|
|
return False |
296
|
|
|
|
297
|
|
|
action = kw.get("action") |
298
|
|
|
tdef = workflow.transitions.get(action) |
299
|
|
|
actor = kw.get("actor", api.user.get_user_id()) |
300
|
|
|
now = DateTime() |
301
|
|
|
|
302
|
|
|
wf_state = { |
303
|
|
|
"action": action, |
304
|
|
|
"actor": actor, |
305
|
|
|
"comments": "Setting state to %s" % state_id, |
306
|
|
|
"review_state": state_id, |
307
|
|
|
"time": now, |
308
|
|
|
} |
309
|
|
|
|
310
|
|
|
# Get old and new state info |
311
|
|
|
old_state = workflow._getWorkflowStateOf(content) |
312
|
|
|
new_state = workflow.states.get(state_id, None) |
313
|
|
|
if new_state is None: |
314
|
|
|
raise WorkflowException("Destination state undefined: {}" |
315
|
|
|
.format(state_id)) |
316
|
|
|
|
317
|
|
|
if trigger_events: |
318
|
|
|
# Notify the *before* transition event |
319
|
|
|
notify(BeforeTransitionEvent( |
320
|
|
|
content, workflow, old_state, new_state, tdef, wf_state, None)) |
321
|
|
|
|
322
|
|
|
# Change status and update permissions |
323
|
|
|
portal_workflow.setStatusOf(wf_id, content, wf_state) |
324
|
|
|
workflow.updateRoleMappingsFor(content) |
325
|
|
|
|
326
|
|
|
# Map changes to catalog |
327
|
|
|
indexes = ["allowedRolesAndUsers", "review_state", "is_active"] |
328
|
|
|
content.reindexObject(idxs=indexes) |
329
|
|
|
|
330
|
|
|
if trigger_events: |
331
|
|
|
# Notify the *after* transition event |
332
|
|
|
notify(AfterTransitionEvent( |
333
|
|
|
content, workflow, old_state, new_state, tdef, wf_state, None)) |
334
|
|
|
|
335
|
|
|
return True |
336
|
|
|
|
337
|
|
|
|
338
|
|
|
def tmpID(): |
339
|
|
|
import binascii |
340
|
|
|
return binascii.hexlify(os.urandom(16)) |
341
|
|
|
|
342
|
|
|
|
343
|
|
|
def isnumber(s): |
344
|
|
|
return api.is_floatable(s) |
345
|
|
|
|
346
|
|
|
|
347
|
|
|
def senaite_url_fetcher(url): |
348
|
|
|
"""Uses plone.subrequest to fetch an internal image resource. |
349
|
|
|
|
350
|
|
|
If the URL points to an external resource, the URL is handed |
351
|
|
|
to weasyprint.default_url_fetcher. |
352
|
|
|
|
353
|
|
|
Please see these links for details: |
354
|
|
|
|
355
|
|
|
- https://github.com/plone/plone.subrequest |
356
|
|
|
- https://pypi.python.org/pypi/plone.subrequest |
357
|
|
|
- https://github.com/senaite/senaite.core/issues/538 |
358
|
|
|
|
359
|
|
|
:returns: A dict with the following keys: |
360
|
|
|
|
361
|
|
|
* One of ``string`` (a byte string) or ``file_obj`` |
362
|
|
|
(a file-like object) |
363
|
|
|
* Optionally: ``mime_type``, a MIME type extracted e.g. from a |
364
|
|
|
*Content-Type* header. If not provided, the type is guessed from the |
365
|
|
|
file extension in the URL. |
366
|
|
|
* Optionally: ``encoding``, a character encoding extracted e.g. from a |
367
|
|
|
*charset* parameter in a *Content-Type* header |
368
|
|
|
* Optionally: ``redirected_url``, the actual URL of the resource |
369
|
|
|
if there were e.g. HTTP redirects. |
370
|
|
|
* Optionally: ``filename``, the filename of the resource. Usually |
371
|
|
|
derived from the *filename* parameter in a *Content-Disposition* |
372
|
|
|
header |
373
|
|
|
|
374
|
|
|
If a ``file_obj`` key is given, it is the caller’s responsibility |
375
|
|
|
to call ``file_obj.close()``. |
376
|
|
|
""" |
377
|
|
|
|
378
|
|
|
logger.info("Fetching URL '{}' for WeasyPrint".format(url)) |
379
|
|
|
|
380
|
|
|
# get the pyhsical path from the URL |
381
|
|
|
request = api.get_request() |
382
|
|
|
host = request.get_header("HOST") |
383
|
|
|
path = "/".join(request.physicalPathFromURL(url)) |
384
|
|
|
|
385
|
|
|
# fetch the object by sub-request |
386
|
|
|
portal = api.get_portal() |
387
|
|
|
context = portal.restrictedTraverse(path, None) |
388
|
|
|
|
389
|
|
|
# We double check here to avoid an edge case, where we have the same path |
390
|
|
|
# as well in our local site, e.g. we have `/senaite/img/systems/senaite.png`, |
391
|
|
|
# but the user requested http://www.ridingbytes.com/img/systems/senaite.png: |
392
|
|
|
# |
393
|
|
|
# "/".join(request.physicalPathFromURL("http://www.ridingbytes.com/img/systems/senaite.png")) |
394
|
|
|
# '/senaite/img/systems/senaite.png' |
395
|
|
|
if context is None or host not in url: |
396
|
|
|
logger.info("URL is external, passing over to the default URL fetcher...") |
397
|
|
|
return default_url_fetcher(url) |
398
|
|
|
|
399
|
|
|
logger.info("URL is local, fetching data by path '{}' via subrequest".format(path)) |
400
|
|
|
|
401
|
|
|
# get the data via an authenticated subrequest |
402
|
|
|
response = subrequest(path) |
403
|
|
|
|
404
|
|
|
# Prepare the return data as required by WeasyPrint |
405
|
|
|
string = response.getBody() |
406
|
|
|
filename = url.split("/")[-1] |
407
|
|
|
mime_type = mimetypes.guess_type(url)[0] |
408
|
|
|
redirected_url = url |
409
|
|
|
|
410
|
|
|
return { |
411
|
|
|
"string": string, |
412
|
|
|
"filename": filename, |
413
|
|
|
"mime_type": mime_type, |
414
|
|
|
"redirected_url": redirected_url, |
415
|
|
|
} |
416
|
|
|
|
417
|
|
|
|
418
|
|
|
def createPdf(htmlreport, outfile=None, css=None, images={}): |
419
|
|
|
"""create a PDF from some HTML. |
420
|
|
|
htmlreport: rendered html |
421
|
|
|
outfile: pdf filename; if supplied, caller is responsible for creating |
422
|
|
|
and removing it. |
423
|
|
|
css: remote URL of css file to download |
424
|
|
|
images: A dictionary containing possible URLs (keys) and local filenames |
425
|
|
|
(values) with which they may to be replaced during rendering. |
426
|
|
|
# WeasyPrint will attempt to retrieve images directly from the URL |
427
|
|
|
# referenced in the HTML report, which may refer back to a single-threaded |
428
|
|
|
# (and currently occupied) zeoclient, hanging it. All image source |
429
|
|
|
# URL's referenced in htmlreport should be local files. |
430
|
|
|
""" |
431
|
|
|
# A list of files that should be removed after PDF is written |
432
|
|
|
cleanup = [] |
433
|
|
|
css_def = '' |
434
|
|
|
if css: |
435
|
|
|
if css.startswith("http://") or css.startswith("https://"): |
436
|
|
|
# Download css file in temp dir |
437
|
|
|
u = urlopen(css) |
438
|
|
|
_cssfile = tempfile.mktemp(suffix='.css') |
439
|
|
|
localFile = open(_cssfile, 'w') |
440
|
|
|
localFile.write(u.read()) |
441
|
|
|
localFile.close() |
442
|
|
|
cleanup.append(_cssfile) |
443
|
|
|
else: |
444
|
|
|
_cssfile = css |
445
|
|
|
cssfile = open(_cssfile, 'r') |
446
|
|
|
css_def = cssfile.read() |
447
|
|
|
|
448
|
|
|
htmlreport = to_utf8(htmlreport) |
449
|
|
|
|
450
|
|
|
for (key, val) in images.items(): |
451
|
|
|
htmlreport = htmlreport.replace(key, val) |
452
|
|
|
|
453
|
|
|
# render |
454
|
|
|
htmlreport = to_utf8(htmlreport) |
455
|
|
|
renderer = HTML(string=htmlreport, url_fetcher=senaite_url_fetcher, encoding='utf-8') |
456
|
|
|
pdf_fn = outfile if outfile else tempfile.mktemp(suffix=".pdf") |
457
|
|
|
if css: |
458
|
|
|
renderer.write_pdf(pdf_fn, stylesheets=[CSS(string=css_def)]) |
459
|
|
|
else: |
460
|
|
|
renderer.write_pdf(pdf_fn) |
461
|
|
|
# return file data |
462
|
|
|
pdf_data = open(pdf_fn, "rb").read() |
463
|
|
|
if outfile is None: |
464
|
|
|
os.remove(pdf_fn) |
465
|
|
|
for fn in cleanup: |
466
|
|
|
os.remove(fn) |
467
|
|
|
return pdf_data |
468
|
|
|
|
469
|
|
|
|
470
|
|
|
def attachPdf(mimemultipart, pdfreport, filename=None): |
471
|
|
|
part = MIMEBase('application', "pdf") |
472
|
|
|
part.add_header('Content-Disposition', |
473
|
|
|
'attachment; filename="%s.pdf"' % (filename or tmpID())) |
474
|
|
|
part.set_payload(pdfreport) |
475
|
|
|
Encoders.encode_base64(part) |
476
|
|
|
mimemultipart.attach(part) |
477
|
|
|
|
478
|
|
|
|
479
|
|
|
def currency_format(context, locale): |
480
|
|
|
locale = locales.getLocale(locale) |
481
|
|
|
currency = context.bika_setup.getCurrency() |
482
|
|
|
symbol = locale.numbers.currencies[currency].symbol |
483
|
|
|
|
484
|
|
|
def format(val): |
485
|
|
|
return '%s %0.2f' % (symbol, val) |
486
|
|
|
return format |
487
|
|
|
|
488
|
|
|
|
489
|
|
|
def getHiddenAttributesForClass(classname): |
490
|
|
|
try: |
491
|
|
|
registry = queryUtility(IRegistry) |
492
|
|
|
hiddenattributes = registry.get('bika.lims.hiddenattributes', ()) |
493
|
|
|
if hiddenattributes is not None: |
494
|
|
|
for alist in hiddenattributes: |
495
|
|
|
if alist[0] == classname: |
496
|
|
|
return alist[1:] |
497
|
|
|
except Exception: |
498
|
|
|
logger.warning( |
499
|
|
|
'Probem accessing optionally hidden attributes in registry') |
500
|
|
|
|
501
|
|
|
return [] |
502
|
|
|
|
503
|
|
|
|
504
|
|
|
def dicts_to_dict(dictionaries, key_subfieldname): |
505
|
|
|
"""Convert a list of dictionaries into a dictionary of dictionaries. |
506
|
|
|
|
507
|
|
|
key_subfieldname must exist in each Record's subfields and have a value, |
508
|
|
|
which will be used as the key for the new dictionary. If a key is duplicated, |
509
|
|
|
the earlier value will be overwritten. |
510
|
|
|
""" |
511
|
|
|
result = {} |
512
|
|
|
for d in dictionaries: |
513
|
|
|
result[d[key_subfieldname]] = d |
514
|
|
|
return result |
515
|
|
|
|
516
|
|
|
|
517
|
|
|
def format_supsub(text): |
518
|
|
|
""" |
519
|
|
|
Mainly used for Analysis Service's unit. Transform the text adding |
520
|
|
|
sub and super html scripts: |
521
|
|
|
For super-scripts, use ^ char |
522
|
|
|
For sub-scripts, use _ char |
523
|
|
|
The expression "cm^2" will be translated to "cm²" and the |
524
|
|
|
expression "b_(n-1)" will be translated to "b n-1". |
525
|
|
|
The expression "n_(fibras)/cm^3" will be translated as |
526
|
|
|
"n fibras / cm³" |
527
|
|
|
:param text: text to be formatted |
528
|
|
|
""" |
529
|
|
|
out = [] |
530
|
|
|
subsup = [] |
531
|
|
|
clauses = [] |
532
|
|
|
insubsup = True |
533
|
|
|
for c in str(text).strip(): |
534
|
|
|
if c == '(': |
535
|
|
|
if insubsup is False: |
536
|
|
|
out.append(c) |
537
|
|
|
clauses.append(')') |
538
|
|
|
else: |
539
|
|
|
clauses.append('') |
540
|
|
|
|
541
|
|
|
elif c == ')': |
542
|
|
|
if len(clauses) > 0: |
543
|
|
|
out.append(clauses.pop()) |
544
|
|
|
if len(subsup) > 0: |
545
|
|
|
out.append(subsup.pop()) |
546
|
|
|
|
547
|
|
|
elif c == '^': |
548
|
|
|
subsup.append('</sup>') |
549
|
|
|
out.append('<sup>') |
550
|
|
|
insubsup = True |
551
|
|
|
continue |
552
|
|
|
|
553
|
|
|
elif c == '_': |
554
|
|
|
subsup.append('</sub>') |
555
|
|
|
out.append('<sub>') |
556
|
|
|
insubsup = True |
557
|
|
|
continue |
558
|
|
|
|
559
|
|
|
elif c == ' ': |
560
|
|
|
if insubsup is True: |
561
|
|
|
out.append(subsup.pop()) |
562
|
|
|
else: |
563
|
|
|
out.append(c) |
564
|
|
|
elif c in ['+', '-']: |
565
|
|
|
if len(clauses) == 0 and len(subsup) > 0: |
566
|
|
|
out.append(subsup.pop()) |
567
|
|
|
out.append(c) |
568
|
|
|
else: |
569
|
|
|
out.append(c) |
570
|
|
|
|
571
|
|
|
insubsup = False |
572
|
|
|
|
573
|
|
|
while True: |
574
|
|
|
if len(subsup) == 0: |
575
|
|
|
break |
576
|
|
|
out.append(subsup.pop()) |
577
|
|
|
|
578
|
|
|
return ''.join(out) |
579
|
|
|
|
580
|
|
|
|
581
|
|
|
def checkPermissions(permissions=[], obj=None): |
582
|
|
|
""" |
583
|
|
|
Checks if a user has permissions for a given object. |
584
|
|
|
|
585
|
|
|
Args: |
586
|
|
|
permissions: The permissions the current user must be compliant with |
587
|
|
|
obj: The object for which the permissions apply |
588
|
|
|
|
589
|
|
|
Returns: |
590
|
|
|
1 if the user complies with all the permissions for the given object. |
591
|
|
|
Otherwise, it returns empty. |
592
|
|
|
""" |
593
|
|
|
if not obj: |
594
|
|
|
return False |
595
|
|
|
sm = getSecurityManager() |
596
|
|
|
for perm in permissions: |
597
|
|
|
if not sm.checkPermission(perm, obj): |
598
|
|
|
return '' |
599
|
|
|
return True |
600
|
|
|
|
601
|
|
|
|
602
|
|
|
def getFromString(obj, string, default=None): |
603
|
|
|
attr_obj = obj |
604
|
|
|
attrs = string.split('.') |
605
|
|
|
for attr in attrs: |
606
|
|
|
attr_obj = api.safe_getattr(attr_obj, attr, default=None) |
607
|
|
|
if not attr_obj: |
608
|
|
|
break |
609
|
|
|
return attr_obj or default |
610
|
|
|
|
611
|
|
|
|
612
|
|
|
def measure_time(func_to_measure): |
613
|
|
|
""" |
614
|
|
|
This decorator allows to measure the execution time |
615
|
|
|
of a function and prints it to the console. |
616
|
|
|
:param func_to_measure: function to be decorated |
617
|
|
|
""" |
618
|
|
|
def wrap(*args, **kwargs): |
619
|
|
|
start_time = time() |
620
|
|
|
return_value = func_to_measure(*args, **kwargs) |
621
|
|
|
finish_time = time() |
622
|
|
|
log = "%s took %0.4f seconds. start_time = %0.4f - finish_time = %0.4f\n" % (func_to_measure.func_name, |
623
|
|
|
finish_time - start_time, |
624
|
|
|
start_time, |
625
|
|
|
finish_time) |
626
|
|
|
print(log) |
627
|
|
|
return return_value |
628
|
|
|
return wrap |
629
|
|
|
|
630
|
|
|
|
631
|
|
|
def copy_field_values(src, dst, ignore_fieldnames=None, ignore_fieldtypes=None): |
632
|
|
|
ignore_fields = ignore_fieldnames if ignore_fieldnames else [] |
633
|
|
|
ignore_types = ignore_fieldtypes if ignore_fieldtypes else [] |
634
|
|
|
if 'id' not in ignore_fields: |
635
|
|
|
ignore_fields.append('id') |
636
|
|
|
|
637
|
|
|
src_schema = src.Schema() |
638
|
|
|
dst_schema = dst.Schema() |
639
|
|
|
|
640
|
|
|
for field in src_schema.fields(): |
641
|
|
|
if IComputedField.providedBy(field): |
642
|
|
|
continue |
643
|
|
|
fieldname = field.getName() |
644
|
|
|
if fieldname in ignore_fields \ |
645
|
|
|
or field.type in ignore_types \ |
646
|
|
|
or fieldname not in dst_schema: |
647
|
|
|
continue |
648
|
|
|
value = field.get(src) |
649
|
|
|
if value: |
650
|
|
|
dst_schema[fieldname].set(dst, value) |
651
|
|
|
|
652
|
|
|
|
653
|
|
|
def get_link(href, value=None, csrf=True, **kwargs): |
654
|
|
|
""" |
655
|
|
|
Returns a well-formed link. If href is None/empty, returns an empty string |
656
|
|
|
:param href: value to be set for attribute href |
657
|
|
|
:param value: the text to be displayed. If None, the href itself is used |
658
|
|
|
:param value: if True, the CSRF token is added in the href |
659
|
|
|
:param kwargs: additional attributes and values |
660
|
|
|
:return: a well-formed html anchor |
661
|
|
|
""" |
662
|
|
|
if not href: |
663
|
|
|
return "" |
664
|
|
|
anchor_value = value and value or href |
665
|
|
|
attr = render_html_attributes(**kwargs) |
666
|
|
|
# Add a CSRF token |
667
|
|
|
if csrf and href.startswith("http"): |
668
|
|
|
href = addTokenToUrl(href) |
669
|
|
|
return '<a href="{}" {}>{}</a>'.format(href, attr, anchor_value) |
670
|
|
|
|
671
|
|
|
|
672
|
|
|
def get_link_for(obj, **kwargs): |
673
|
|
|
"""Returns a well-formed html anchor to the object |
674
|
|
|
""" |
675
|
|
|
if not obj: |
676
|
|
|
return "" |
677
|
|
|
href = api.get_url(obj) |
678
|
|
|
value = api.get_title(obj) |
679
|
|
|
return get_link(href=href, value=value, **kwargs) |
680
|
|
|
|
681
|
|
|
|
682
|
|
|
def get_email_link(email, value=None): |
683
|
|
|
""" |
684
|
|
|
Returns a well-formed link to an email address. If email is None/empty, |
685
|
|
|
returns an empty string |
686
|
|
|
:param email: email address |
687
|
|
|
:param link_text: text to be displayed. If None, the email itself is used |
688
|
|
|
:return: a well-formatted html anchor |
689
|
|
|
""" |
690
|
|
|
if not email: |
691
|
|
|
return "" |
692
|
|
|
mailto = 'mailto:{}'.format(email) |
693
|
|
|
link_value = value and value or email |
694
|
|
|
return get_link(mailto, link_value) |
695
|
|
|
|
696
|
|
|
|
697
|
|
|
def get_phone_link(phone, value=None): |
698
|
|
|
""" |
699
|
|
|
Returns a well-formed link to a phone number. If phone is None/empty, |
700
|
|
|
returns an empty string |
701
|
|
|
:param phone: phone number |
702
|
|
|
:param value: text to be displayed. If None, the phone itself is used |
703
|
|
|
:return: a well-formatted html anchor |
704
|
|
|
""" |
705
|
|
|
if not phone: |
706
|
|
|
return "" |
707
|
|
|
number = re.sub(r"[)(\s-]", "", phone) |
708
|
|
|
tel = "tel:{}".format(number) |
709
|
|
|
link_value = value and value or phone |
710
|
|
|
return get_link(tel, link_value) |
711
|
|
|
|
712
|
|
|
|
713
|
|
|
def get_image(name, **kwargs): |
714
|
|
|
"""Returns a well-formed image |
715
|
|
|
:param name: file name of the image |
716
|
|
|
:param kwargs: additional attributes and values |
717
|
|
|
:return: a well-formed html img |
718
|
|
|
""" |
719
|
|
|
if not name: |
720
|
|
|
return "" |
721
|
|
|
portal = api.get_portal() |
722
|
|
|
theme = portal.restrictedTraverse("@@senaite_theme") |
723
|
|
|
basename, ext = os.path.splitext(name) |
724
|
|
|
if basename in theme.icons(): |
725
|
|
|
if "width" not in kwargs: |
726
|
|
|
kwargs["width"] = "16" |
727
|
|
|
return theme.icon_tag(basename, **kwargs) |
728
|
|
|
portal_url = api.get_url(portal) |
729
|
|
|
attr = render_html_attributes(**kwargs) |
730
|
|
|
html = '<img src="{}/++resource++bika.lims.images/{}" {}/>' |
731
|
|
|
return html.format(portal_url, name, attr) |
732
|
|
|
|
733
|
|
|
|
734
|
|
|
def get_progress_bar_html(percentage): |
735
|
|
|
"""Returns an html that represents a progress bar |
736
|
|
|
""" |
737
|
|
|
return '<div class="progress md-progress">' \ |
738
|
|
|
'<div class="progress-bar" style="width: {0}%">{0}%</div>' \ |
739
|
|
|
'</div>'.format(percentage or 0) |
740
|
|
|
|
741
|
|
|
|
742
|
|
|
def render_html_attributes(**kwargs): |
743
|
|
|
"""Returns a string representation of attributes for html entities |
744
|
|
|
:param kwargs: attributes and values |
745
|
|
|
:return: a well-formed string representation of attributes""" |
746
|
|
|
attr = list() |
747
|
|
|
if kwargs: |
748
|
|
|
attr = ['{}="{}"'.format(key, val) for key, val in kwargs.items()] |
749
|
|
|
return " ".join(attr).replace("css_class", "class") |
750
|
|
|
|
751
|
|
|
|
752
|
|
|
def get_registry_value(key, default=None): |
753
|
|
|
""" |
754
|
|
|
Gets the utility for IRegistry and returns the value for the key passed in. |
755
|
|
|
If there is no value for the key passed in, returns default value |
756
|
|
|
:param key: the key in the registry to look for |
757
|
|
|
:param default: default value if the key is not registered |
758
|
|
|
:return: value in the registry for the key passed in |
759
|
|
|
""" |
760
|
|
|
# cannot use bika.lims.deprecated (circular dependencies) |
761
|
|
|
import warnings |
762
|
|
|
warnings.simplefilter("always", DeprecationWarning) |
763
|
|
|
warn = "Deprecated: use senaite.core.api.get_registry_record instead" |
764
|
|
|
warnings.warn(warn, category=DeprecationWarning, stacklevel=2) |
765
|
|
|
warnings.simplefilter("default", DeprecationWarning) |
766
|
|
|
return api.get_registry_record(key, default=default) |
767
|
|
|
|
768
|
|
|
|
769
|
|
|
def check_permission(permission, obj): |
770
|
|
|
""" |
771
|
|
|
Returns if the current user has rights for the permission passed in against |
772
|
|
|
the obj passed in |
773
|
|
|
:param permission: name of the permission |
774
|
|
|
:param obj: the object to check the permission against for the current user |
775
|
|
|
:return: 1 if the user has rights for this permission for the passed in obj |
776
|
|
|
""" |
777
|
|
|
mtool = api.get_tool('portal_membership') |
778
|
|
|
object = api.get_object(obj) |
779
|
|
|
return mtool.checkPermission(permission, object) |
780
|
|
|
|
781
|
|
|
|
782
|
|
|
def to_int(value, default=0): |
783
|
|
|
""" |
784
|
|
|
Tries to convert the value passed in as an int. If no success, returns the |
785
|
|
|
default value passed in |
786
|
|
|
:param value: the string to convert to integer |
787
|
|
|
:param default: the default fallback |
788
|
|
|
:return: int representation of the value passed in |
789
|
|
|
""" |
790
|
|
|
try: |
791
|
|
|
return int(value) |
792
|
|
|
except (TypeError, ValueError): |
793
|
|
|
return to_int(default, default=0) |
794
|
|
|
|
795
|
|
|
|
796
|
|
|
def get_strings(data): |
797
|
|
|
""" |
798
|
|
|
Convert unicode values to strings even if they belong to lists or dicts. |
799
|
|
|
:param data: an object. |
800
|
|
|
:return: The object with all unicode values converted to string. |
801
|
|
|
""" |
802
|
|
|
# if this is a unicode string, return its string representation |
803
|
|
|
if isinstance(data, unicode): |
|
|
|
|
804
|
|
|
return data.encode('utf-8') |
805
|
|
|
|
806
|
|
|
# if this is a list of values, return list of string values |
807
|
|
|
if isinstance(data, list): |
808
|
|
|
return map(get_strings, data) |
809
|
|
|
|
810
|
|
|
# if this is a dictionary, return dictionary of string keys and values |
811
|
|
|
if isinstance(data, dict): |
812
|
|
|
|
813
|
|
|
return { |
814
|
|
|
get_strings(key): get_strings(value) |
815
|
|
|
for key, value in six.iteritems(data) |
816
|
|
|
} |
817
|
|
|
# if it's anything else, return it in its original form |
818
|
|
|
return data |
819
|
|
|
|
820
|
|
|
|
821
|
|
|
def get_unicode(data): |
822
|
|
|
""" |
823
|
|
|
Convert string values to unicode even if they belong to lists or dicts. |
824
|
|
|
:param data: an object. |
825
|
|
|
:return: The object with all string values converted to unicode. |
826
|
|
|
""" |
827
|
|
|
# if this is a common string, return its unicode representation |
828
|
|
|
if isinstance(data, str): |
829
|
|
|
return safe_unicode(data) |
830
|
|
|
|
831
|
|
|
# if this is a list of values, return list of unicode values |
832
|
|
|
if isinstance(data, list): |
833
|
|
|
return [get_unicode(item) for item in data] |
834
|
|
|
|
835
|
|
|
# if this is a dictionary, return dictionary of unicode keys and values |
836
|
|
|
if isinstance(data, dict): |
837
|
|
|
return { |
838
|
|
|
get_unicode(key): get_unicode(value) |
839
|
|
|
for key, value in six.iteritems(data) |
840
|
|
|
} |
841
|
|
|
# if it's anything else, return it in its original form |
842
|
|
|
return data |
843
|
|
|
|
844
|
|
|
|
845
|
|
|
def is_bika_installed(): |
846
|
|
|
"""Check if Bika LIMS is installed in the Portal |
847
|
|
|
""" |
848
|
|
|
qi = api.portal.get_tool("portal_quickinstaller") |
849
|
|
|
return qi.isProductInstalled("bika.lims") |
850
|
|
|
|
851
|
|
|
|
852
|
|
|
def get_display_list(brains_or_objects=None, none_item=False): |
853
|
|
|
""" |
854
|
|
|
Returns a DisplayList with the items sorted by Title |
855
|
|
|
:param brains_or_objects: list of brains or objects |
856
|
|
|
:param none_item: adds an item with empty uid and text "Select.." in pos 0 |
857
|
|
|
:return: DisplayList (uid, title) sorted by title ascending |
858
|
|
|
:rtype: DisplayList |
859
|
|
|
""" |
860
|
|
|
if brains_or_objects is None: |
861
|
|
|
return get_display_list(list(), none_item) |
862
|
|
|
|
863
|
|
|
items = list() |
864
|
|
|
for brain in brains_or_objects: |
865
|
|
|
uid = api.get_uid(brain) |
866
|
|
|
if not uid: |
867
|
|
|
continue |
868
|
|
|
title = api.get_title(brain) |
869
|
|
|
items.append((uid, title)) |
870
|
|
|
|
871
|
|
|
# Sort items by title ascending |
872
|
|
|
items.sort(lambda x, y: cmp(x[1], y[1])) |
873
|
|
|
|
874
|
|
|
# Add the first item? |
875
|
|
|
if none_item: |
876
|
|
|
items.insert(0, ('', t('Select...'))) |
877
|
|
|
|
878
|
|
|
return DisplayList(items) |
879
|
|
|
|
880
|
|
|
|
881
|
|
|
def to_choices(display_list): |
882
|
|
|
"""Converts a display list to a choices list |
883
|
|
|
""" |
884
|
|
|
if not display_list: |
885
|
|
|
return [] |
886
|
|
|
|
887
|
|
|
return map( |
888
|
|
|
lambda item: { |
889
|
|
|
"ResultValue": item[0], |
890
|
|
|
"ResultText": item[1]}, |
891
|
|
|
display_list.items()) |
892
|
|
|
|
893
|
|
|
|
894
|
|
|
def chain(obj): |
895
|
|
|
"""Generator to walk the acquistion chain of object, considering that it |
896
|
|
|
could be a function. |
897
|
|
|
|
898
|
|
|
If the thing we are accessing is actually a bound method on an instance, |
899
|
|
|
then after we've checked the method itself, get the instance it's bound to |
900
|
|
|
using im_self, so that we can continue to walk up the acquistion chain from |
901
|
|
|
it (incidentally, this is why we can't juse use aq_chain()). |
902
|
|
|
""" |
903
|
|
|
context = aq_inner(obj) |
904
|
|
|
|
905
|
|
|
while context is not None: |
906
|
|
|
yield context |
907
|
|
|
|
908
|
|
|
func_object = getattr(context, "im_self", None) |
909
|
|
|
if func_object is not None: |
910
|
|
|
context = aq_inner(func_object) |
911
|
|
|
else: |
912
|
|
|
# Don't use aq_inner() since portal_factory (and probably other) |
913
|
|
|
# things, depends on being able to wrap itself in a fake context. |
914
|
|
|
context = aq_parent(context) |
915
|
|
|
|
916
|
|
|
|
917
|
|
|
def get_client(obj): |
918
|
|
|
"""Returns the client the object passed-in belongs to, if any |
919
|
|
|
|
920
|
|
|
This walks the acquisition chain up until we find something which provides |
921
|
|
|
either IClient or IClientAwareMixin |
922
|
|
|
""" |
923
|
|
|
for obj in chain(obj): |
924
|
|
|
if IClient.providedBy(obj): |
925
|
|
|
return obj |
926
|
|
|
elif IClientAwareMixin.providedBy(obj): |
927
|
|
|
# ClientAwareMixin can return a Client, even if there is no client |
928
|
|
|
# in the acquisition chain |
929
|
|
|
return obj.getClient() |
930
|
|
|
|
931
|
|
|
return None |
932
|
|
|
|
933
|
|
|
|
934
|
|
|
def get_fas_ico(icon_id, **attrs): |
935
|
|
|
"""Returns a well-formed fontawesome (fas) icon |
936
|
|
|
""" |
937
|
|
|
css_class = attrs.pop("css_class", "") |
938
|
|
|
if not icon_id.startswith("fa-"): |
939
|
|
|
icon_id = "fa-%s" % icon_id |
940
|
|
|
attrs["css_class"] = " ".join([css_class, "fas", icon_id]).strip() |
941
|
|
|
return "<i %s/>" % render_html_attributes(**attrs) |
942
|
|
|
|