build.bika.lims.catalog.indexers.auditlog.actor() - Code Metrics - Inspection of "Audit log (#1324)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 804f45...1fa2c0 )

by Jordi

created 2019-04-22 21:47 UTC

build.bika.lims.catalog.indexers.auditlog.actor() A

↳ Parent: build.bika.lims.catalog.indexers.auditlog

Complexity

Conditions

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	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-2019 by it's authors.
# Some rights reserved, see README and LICENSE.

import itertools
import re

from bika.lims import api
from bika.lims.api.snapshot import get_last_snapshot
from bika.lims.api.snapshot import get_snapshot_count
from bika.lims.api.snapshot import get_snapshot_metadata
from bika.lims.api.snapshot import get_snapshots
from bika.lims.api.user import get_user_id
from bika.lims.interfaces import IAuditable
from plone.indexer import indexer
from plone.memoize.ram import DontCache
from plone.memoize.ram import cache

UID_RX = re.compile(r"[a-z0-9]{32}$")
DATE_RX = re.compile(r"\d{4}[-/]\d{2}[-/]\d{2}")


def _uid_to_title_cache_key(func, uid):
    brain = api.get_brain_by_uid(uid, default=None)
    if brain is None:
        raise DontCache
    modified = api.get_modification_date(brain).millis()
    return "{}-{}".format(uid, modified)


@cache(_uid_to_title_cache_key)
def get_title_or_id_from_uid(uid):
    """Returns the title or ID from the given UID
    """
    obj = api.get_object_by_uid(uid, default=None)
    if obj is None:
        return ""
    title_or_id = api.get_title(obj) or api.get_id(obj)
    return title_or_id


def get_meta_value_for(snapshot, key, default=None):
    """Returns the metadata value for the given key
    """
    metadata = get_snapshot_metadata(snapshot)
    return metadata.get(key, default)


def get_actor(snapshot):
    """Get the actor of the snapshot
    """
    actor = get_meta_value_for(snapshot, "actor")
    if not actor:
        return get_user_id()
    return actor


def get_fullname(snapshot):
    """Get the actor's fullname of the snapshot
    """
    actor = get_actor(snapshot)
    properties = api.get_user_properties(actor)
    return properties.get("fullname", actor)


def get_action(snapshot):
    """Get the action of the snapshot
    """
    action = get_meta_value_for(snapshot, "action")
    if not action:
        return "Edit"
    return action


def get_created(snapshot):
    """Get the created date of the snapshot
    """
    created = get_meta_value_for(snapshot, "snapshot_created")
    if not created:
        return ""
    return created


@indexer(IAuditable)
def actor(instance):
    """Last modifiying user
    """
    last_snapshot = get_last_snapshot(instance)
    return get_actor(last_snapshot)


@indexer(IAuditable)
def fullname(instance):
    """Last modifiying user
    """
    last_snapshot = get_last_snapshot(instance)
    return get_fullname(last_snapshot)


@indexer(IAuditable)
def modifiers(instance):
    """Returns a list of all users that modified
    """
    snapshots = get_snapshots(instance)
    return map(get_actor, snapshots)


@indexer(IAuditable)
def action(instance):
    """Returns the last performed action
    """
    last_snapshot = get_last_snapshot(instance)
    return get_action(last_snapshot)


@indexer(IAuditable)
def listing_searchable_text(instance):
    """Fulltext search for the audit metadata
    """
    # get all snapshots
    snapshots = get_snapshots(instance)
    # extract all snapshot values, because we are not interested in the
    # fieldnames (keys)
    values = map(lambda s: s.values(), snapshots)
    # prepare a set of unified catalog data
    catalog_data = set()
    # values to skip
    skip_values = ["None", "true", "True", "false", "False"]
    # internal uid -> title cache
    uid_title_cache = {}

    # helper function to recursively unpack the snapshot values
    def append(value):
        if isinstance(value, (list, tuple)):
            map(append, value)
        elif isinstance(value, (dict)):
            map(append, value.items())
        elif isinstance(value, basestring):
            # convert unicode to UTF8
            if isinstance(value, unicode):
                value = api.safe_unicode(value).encode("utf8")
            # skip single short values
            if len(value) < 2:
                return
            # flush non meaningful values
            if value in skip_values:
                return
            # flush ISO dates
            if re.match(DATE_RX, value):
                return
            # fetch the title
            if re.match(UID_RX, value):
                if value in uid_title_cache:

                    value = uid_title_cache[value]
                else:
                    title_or_id = get_title_or_id_from_uid(value)

                    uid_title_cache[value] = title_or_id

                    value = title_or_id
            catalog_data.add(value)

    # extract all meaningful values
    for value in itertools.chain(values):
        append(value)

    return " ".join(catalog_data)


@indexer(IAuditable)
def snapshot_created(instance):
    """Snapshot created date
    """
    last_snapshot = get_last_snapshot(instance)
    snapshot_created = get_created(last_snapshot)
    return api.to_date(snapshot_created)


@indexer(IAuditable)
def snapshot_version(instance):
    """Snapshot created date
    """
    return get_snapshot_count(instance)


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-2019 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			import itertools
22			import re
23
24			from bika.lims import api
25			from bika.lims.api.snapshot import get_last_snapshot
26			from bika.lims.api.snapshot import get_snapshot_count
27			from bika.lims.api.snapshot import get_snapshot_metadata
28			from bika.lims.api.snapshot import get_snapshots
29			from bika.lims.api.user import get_user_id
30			from bika.lims.interfaces import IAuditable
31			from plone.indexer import indexer
32			from plone.memoize.ram import DontCache
33			from plone.memoize.ram import cache
34
35			UID_RX = re.compile(r"[a-z0-9]{32}$")
36			DATE_RX = re.compile(r"\d{4}[-/]\d{2}[-/]\d{2}")
37
38
39			def _uid_to_title_cache_key(func, uid):
40			brain = api.get_brain_by_uid(uid, default=None)
41			if brain is None:
42			raise DontCache
43			modified = api.get_modification_date(brain).millis()
44			return "{}-{}".format(uid, modified)
45
46
47			@cache(_uid_to_title_cache_key)
48			def get_title_or_id_from_uid(uid):
49			"""Returns the title or ID from the given UID
50			"""
51			obj = api.get_object_by_uid(uid, default=None)
52			if obj is None:
53			return ""
54			title_or_id = api.get_title(obj) or api.get_id(obj)
55			return title_or_id
56
57
58			def get_meta_value_for(snapshot, key, default=None):
59			"""Returns the metadata value for the given key
60			"""
61			metadata = get_snapshot_metadata(snapshot)
62			return metadata.get(key, default)
63
64
65			def get_actor(snapshot):
66			"""Get the actor of the snapshot
67			"""
68			actor = get_meta_value_for(snapshot, "actor")
69			if not actor:
70			return get_user_id()
71			return actor
72
73
74			def get_fullname(snapshot):
75			"""Get the actor's fullname of the snapshot
76			"""
77			actor = get_actor(snapshot)
78			properties = api.get_user_properties(actor)
79			return properties.get("fullname", actor)
80
81
82			def get_action(snapshot):
83			"""Get the action of the snapshot
84			"""
85			action = get_meta_value_for(snapshot, "action")
86			if not action:
87			return "Edit"
88			return action
89
90
91			def get_created(snapshot):
92			"""Get the created date of the snapshot
93			"""
94			created = get_meta_value_for(snapshot, "snapshot_created")
95			if not created:
96			return ""
97			return created
98
99
100			@indexer(IAuditable)
101			def actor(instance):
102			"""Last modifiying user
103			"""
104			last_snapshot = get_last_snapshot(instance)
105			return get_actor(last_snapshot)
106
107
108			@indexer(IAuditable)
109			def fullname(instance):
110			"""Last modifiying user
111			"""
112			last_snapshot = get_last_snapshot(instance)
113			return get_fullname(last_snapshot)
114
115
116			@indexer(IAuditable)
117			def modifiers(instance):
118			"""Returns a list of all users that modified
119			"""
120			snapshots = get_snapshots(instance)
121			return map(get_actor, snapshots)
122
123
124			@indexer(IAuditable)
125			def action(instance):
126			"""Returns the last performed action
127			"""
128			last_snapshot = get_last_snapshot(instance)
129			return get_action(last_snapshot)
130
131
132			@indexer(IAuditable)
133			def listing_searchable_text(instance):
134			"""Fulltext search for the audit metadata
135			"""
136			# get all snapshots
137			snapshots = get_snapshots(instance)
138			# extract all snapshot values, because we are not interested in the
139			# fieldnames (keys)
140			values = map(lambda s: s.values(), snapshots)
141			# prepare a set of unified catalog data
142			catalog_data = set()
143			# values to skip
144			skip_values = ["None", "true", "True", "false", "False"]
145			# internal uid -> title cache
146			uid_title_cache = {}
147
148			# helper function to recursively unpack the snapshot values
149			def append(value):
150			if isinstance(value, (list, tuple)):
151			map(append, value)
152			elif isinstance(value, (dict)):
153			map(append, value.items())
154			elif isinstance(value, basestring):
155			# convert unicode to UTF8
156			if isinstance(value, unicode):
157			value = api.safe_unicode(value).encode("utf8")
158			# skip single short values
159			if len(value) < 2:
160			return
161			# flush non meaningful values
162			if value in skip_values:
163			return
164			# flush ISO dates
165			if re.match(DATE_RX, value):
166			return
167			# fetch the title
168			if re.match(UID_RX, value):
169			if value in uid_title_cache:
			0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-20 20:50 UTC by Report Bug Copy Issue Report The variable `uid_title_cache` does not seem to be defined. Loading history...
170			value = uid_title_cache[value]
171			else:
172			title_or_id = get_title_or_id_from_uid(value)
			0 ignored issues – show Bug introduced 2019-04-20 20:50 UTC by Report Bug Copy Issue Report The loop variable `value` might not be defined here. Loading history...
173			uid_title_cache[value] = title_or_id
			0 ignored issues – show Bug introduced 2019-04-20 20:50 UTC by Report Bug Copy Issue Report The loop variable `value` might not be defined here. Loading history...
174			value = title_or_id
175			catalog_data.add(value)
176
177			# extract all meaningful values
178			for value in itertools.chain(values):
179			append(value)
180
181			return " ".join(catalog_data)
182
183
184			@indexer(IAuditable)
185			def snapshot_created(instance):
186			"""Snapshot created date
187			"""
188			last_snapshot = get_last_snapshot(instance)
189			snapshot_created = get_created(last_snapshot)
190			return api.to_date(snapshot_created)
191
192
193			@indexer(IAuditable)
194			def snapshot_version(instance):
195			"""Snapshot created date
196			"""
197			return get_snapshot_count(instance)
198

senaite / senaite.core

Push — master ( 804f45...1fa2c0 )

build.bika.lims.catalog.indexers.auditlog.actor() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like