28170365024

Committed 25 Jun 2026 12:29PM UTC coverage: 91.323% (-0.001%) from 91.324%

Build # 28170365024

Build Type

push

github

Committed by

PascalRepond

Commit Message

fix(theme): repair broken CSS files and demo assets

The per-organisation and wiki CSS files are served as raw static files but used
SCSS-style `//` comments, which are invalid in plain CSS.

* Replace `//` comments with valid `/* */` CSS comments.
* Point the organisation logos to the local SVG assets under
  /static/themes/images/ instead of the remote resources.rero.ch CDN,
  as intended when the local assets were added.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-by: Pascal Repond <pascal.repond@rero.ch>

Coverage Stats

24228 of 26530 relevant lines covered (91.32%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.78

/rero_ils/modules/items/api/api.py

# SPDX-FileCopyrightText: Fondation RERO+
# SPDX-FileCopyrightText: UCLouvain
# SPDX-License-Identifier: AGPL-3.0-or-later

"""API for manipulating items."""

import contextlib
from datetime import UTC, datetime
from functools import partial

from elasticsearch.exceptions import NotFoundError
from elasticsearch_dsl import Q
from invenio_search import current_search_client

from rero_ils.modules.api import IlsRecordError, IlsRecordsIndexer, IlsRecordsSearch
from rero_ils.modules.documents.api import DocumentsSearch
from rero_ils.modules.fetchers import id_fetcher
from rero_ils.modules.item_types.api import ItemTypesSearch
from rero_ils.modules.minters import id_minter
from rero_ils.modules.organisations.api import Organisation
from rero_ils.modules.patrons.api import current_librarian
from rero_ils.modules.providers import Provider
from rero_ils.modules.utils import extracted_data_from_ref

from ..models import ItemIdentifier, ItemMetadata, ItemStatus
from .circulation import ItemCirculation
from .issue import ItemIssue

# provider
ItemProvider = type("ItemProvider", (Provider,), {"identifier": ItemIdentifier, "pid_type": "item"})
# minter
item_id_minter = partial(id_minter, provider=ItemProvider)
# fetcher
item_id_fetcher = partial(id_fetcher, provider=ItemProvider)


class ItemsSearch(IlsRecordsSearch):
    """ItemsSearch."""

    class Meta:
        """Search only on item index."""

        index = "items"
        doc_types = None
        fields = ("*",)
        facets = {}

        default_filter = None

    def available_query(self):
        """Base search index query to compute availability.

        :returns: search index query.
        """
        must_not_filters = [
            # should not be masked
            Q("term", _masked=True),
            # should not be in_transit (even without loan)
            Q("term", status=ItemStatus.IN_TRANSIT),
            # if issue the status should be received
            Q("exists", field="issue") & ~Q("term", issue__status="received"),
        ]

        if not_available_item_types := [
            hit.pid for hit in ItemTypesSearch().source("pid").filter("term", negative_availability=True).scan()
        ]:
            # negative availability item type and not temporary item types
            has_items_filters = Q("terms", item_type__pid=not_available_item_types)
            has_items_filters &= ~Q("exists", field="temporary_item_type")
            # temporary item types with negative availability
            has_items_filters |= Q("terms", temporary_item_type__pid=not_available_item_types)
            # add to the must not filters
            must_not_filters.append(has_items_filters)
        return self.filter(Q("bool", must_not=must_not_filters))


class Item(ItemCirculation, ItemIssue):
    """Item class."""

    minter = item_id_minter
    fetcher = item_id_fetcher
    provider = ItemProvider
    model_cls = ItemMetadata
    pids_exist_check = {
        "required": {"loc": "location", "doc": "document", "itty": "item_type"},
        "not_required": {
            "org": "organisation",
            # We cannot make the holding required because it is created later
            "hold": "holding",
        },
    }

    def delete(self, force=False, dbcommit=False, delindex=False):
        """Delete record."""
        from rero_ils.modules.items.tasks import delete_item_in_collections

        delete_item_in_collections.delay(self.pid, dbcommit=True, reindex=True)
        return super().delete(force, dbcommit, delindex)

    def delete_from_index(self):
        """Delete record from index."""
        with contextlib.suppress(NotFoundError, ValueError):
            ItemsIndexer().delete(self)

    def reasons_not_to_delete(self):
        """Get reasons not to delete record."""
        cannot_delete = {}
        links = self.get_links_to_me()
        # local_fields and collections aren't a reason to block suppression
        links.pop("local_fields", None)
        links.pop("collections", None)
        if links:
            cannot_delete["links"] = links
        return cannot_delete

    def in_collection(self, **kwargs):
        """Get published collection pids for current item."""
        from ...collections.api import CollectionsSearch

        output = []
        search = (
            CollectionsSearch()
            .filter("term", items__pid=self.get("pid"))
            .filter("term", published=True)
            .sort({"title_sort": {"order": "asc"}})
            .params(preserve_order=True)
            .source(["pid", "organisation", "title", "description"])
        )
        orgs = {}
        for hit in search.scan():
            hit = hit.to_dict()
            org_pid = hit["organisation"]["pid"]
            if org_pid not in orgs:
                orgs[org_pid] = Organisation.get_record_by_pid(org_pid)
            collection_data = {
                "pid": hit["pid"],  # required property
                "title": hit["title"],  # required property
                "description": hit.get("description"),  # optional property
                "viewcode": orgs[org_pid].get("code"),
            }
            collection_data = {k: v for k, v in collection_data.items() if v}
            output.append(collection_data)
        return output

    def replace_refs(self):
        """Replace $ref with real data."""
        tmp_itty_end_date = self.get("temporary_item_type", {}).get("end_date")
        tmp_loc_end_date = self.get("temporary_location", {}).get("end_date")
        data = super().replace_refs()
        if tmp_itty_end_date:
            data["temporary_item_type"]["end_date"] = tmp_itty_end_date
        if tmp_loc_end_date:
            data["temporary_location"]["end_date"] = tmp_loc_end_date
        return data

    @classmethod
    def get_item_record_for_ui(cls, **kwargs):
        """Return the item record for ui calls.

        retrieving the item record is possible from an item pid, barcode or
        from a loan.

        :param kwargs: contains at least one of item_pid, item_barcode, or pid.
        :return: the item record.
        """
        from ...loans.api import Loan

        item = None
        item_pid = kwargs.get("item_pid")
        item_barcode = kwargs.pop("item_barcode", None)
        loan_pid = kwargs.get("pid")
        if item_pid:
            item = Item.get_record_by_pid(item_pid)
        elif item_barcode:
            org_pid = kwargs.get("organisation_pid", current_librarian.organisation_pid)
            item = Item.get_item_by_barcode(item_barcode, org_pid)
        elif loan_pid:
            item_pid = Loan.get_record_by_pid(loan_pid).item_pid
            item = Item.get_record_by_pid(item_pid)
        return item

    @classmethod
    def format_end_date(cls, end_date):
        """Return a formatted end date."""
        # (`datetime.now(timezone.utc)` by default)
        if end_date is None:
            end_date = datetime.now(UTC)
        return end_date.strftime("%Y-%m-%d")

    @classmethod
    def get_items_with_obsolete_temporary_item_type_or_location(cls, end_date=None):
        """Get all items with an obsolete temporary item_type or location.

        An end_date could be attached to the item temporary item_type or
        temporary location. If this date is less or equal to sysdate, then the
        temporary_item_type or temorary_location must be considered as obsolete
        and the field must be removed.

        :param end_date: the end_date to check.
        :return A generator of `ItemRecord` object.
        """
        end_date = cls.format_end_date(end_date)
        items_query = ItemsSearch()
        loc_search_query = items_query.filter("range", temporary_location__end_date={"lte": end_date})
        locs = [(hit.meta.id, "loc") for hit in loc_search_query.source("pid").scan()]

        itty_search_query = items_query.filter("range", temporary_item_type__end_date={"lte": end_date})
        itty = [(hit.meta.id, "itty") for hit in itty_search_query.source("pid").scan()]
        hits = itty + locs
        for id, field_type in hits:
            yield Item.get_record(id), field_type


class ItemsIndexer(IlsRecordsIndexer):
    """Indexing items in search index."""

    record_cls = Item

    @classmethod
    def _search_item(cls, record):
        """Get the item from the corresponding index.

        :param record: an item object
        :returns: the search index document or {}
        """
        try:
            search_item = current_search_client.get(ItemsSearch.Meta.index, record.id)
            return search_item["_source"]
        except NotFoundError:
            return {}

    @classmethod
    def _update_status_in_doc(cls, record, search_item):
        """Update the status of a given item in the document index.

        :param record: an item object
        :param search_item: a dict of the search index item
        """
        # retrieve the document in the corresponding es index
        document_pid = extracted_data_from_ref(record.get("document"))
        doc = next(DocumentsSearch().extra(version=True).filter("term", pid=document_pid).scan())
        # update the item status in the document
        data = doc.to_dict()
        for hold in data.get("holdings", []):
            for item in hold.get("items", []):
                if item["pid"] == record.pid:
                    item["status"] = record["status"]
                    break
            else:
                continue
            break
        # reindex the document with the same version
        current_search_client.index(
            index=DocumentsSearch.Meta.index,
            id=doc.meta.id,
            body=data,
            version=doc.meta.version,
            version_type="external_gte",
        )

    def index(self, record):
        """Index an item.

        :param record: an item object
        "returns: the elastiscsearch client result
        """
        from ...holdings.api import Holding

        # get previous indexed version
        search_item = self._search_item(record)

        # call the parent
        return_value = super().index(record)

        # fast document reindex for circulation operations
        if search_item and record.get("status") != search_item.get("status"):
            self._update_status_in_doc(record, search_item)
            return return_value

        # reindex the holding / doc for non circulation operations
        holding_pid = extracted_data_from_ref(record.get("holding"))
        holding = Holding.get_record_by_pid(holding_pid)
        holding.reindex()
        # reindex the old holding
        old_holding_pid = None
        if search_item:
            # reindex old holding ot update hte count
            old_holding_pid = search_item.get("holding", {}).get("pid")
            if old_holding_pid != holding_pid:
                old_holding = Holding.get_record_by_pid(old_holding_pid)
                old_holding.reindex()
        return return_value

    def delete(self, record):
        """Delete a record from index.

        :param record: Record instance.
        """
        from rero_ils.modules.holdings.api import Holding

        return_value = super().delete(record)
        holding_pid = extracted_data_from_ref(record.get("holding"))
        holding = Holding.get_record_by_pid(holding_pid)
        # delete only if a standard item
        deleted = False
        if not holding.is_serial:
            with contextlib.suppress(IlsRecordError.NotDeleted):
                holding.delete(force=False, dbcommit=True, delindex=True)
                deleted = True
        if not deleted:
            # for items count
            holding.reindex()
        return return_value

    def bulk_index(self, record_id_iterator):
        """Bulk index records.

        :param record_id_iterator: Iterator yielding record UUIDs.
        """
        super().bulk_index(record_id_iterator, doc_type="item")

1	# SPDX-FileCopyrightText: Fondation RERO+
2	# SPDX-FileCopyrightText: UCLouvain
3	# SPDX-License-Identifier: AGPL-3.0-or-later
4
5	"""API for manipulating items."""
6
7	import contextlib	1✔
8	from datetime import UTC, datetime	1✔
9	from functools import partial	1✔
10
11	from elasticsearch.exceptions import NotFoundError	1✔
12	from elasticsearch_dsl import Q	1✔
13	from invenio_search import current_search_client	1✔
14
15	from rero_ils.modules.api import IlsRecordError, IlsRecordsIndexer, IlsRecordsSearch	1✔
16	from rero_ils.modules.documents.api import DocumentsSearch	1✔
17	from rero_ils.modules.fetchers import id_fetcher	1✔
18	from rero_ils.modules.item_types.api import ItemTypesSearch	1✔
19	from rero_ils.modules.minters import id_minter	1✔
20	from rero_ils.modules.organisations.api import Organisation	1✔
21	from rero_ils.modules.patrons.api import current_librarian	1✔
22	from rero_ils.modules.providers import Provider	1✔
23	from rero_ils.modules.utils import extracted_data_from_ref	1✔
24
25	from ..models import ItemIdentifier, ItemMetadata, ItemStatus	1✔
26	from .circulation import ItemCirculation	1✔
27	from .issue import ItemIssue	1✔
28
29	# provider
30	ItemProvider = type("ItemProvider", (Provider,), {"identifier": ItemIdentifier, "pid_type": "item"})	1✔
31	# minter
32	item_id_minter = partial(id_minter, provider=ItemProvider)	1✔
33	# fetcher
34	item_id_fetcher = partial(id_fetcher, provider=ItemProvider)	1✔
35
36
37	class ItemsSearch(IlsRecordsSearch):	1✔
38	"""ItemsSearch."""
39
40	class Meta:	1✔
41	"""Search only on item index."""
42
43	index = "items"	1✔
44	doc_types = None	1✔
45	fields = ("*",)	1✔
46	facets = {}	1✔
47
48	default_filter = None	1✔
49
50	def available_query(self):	1✔
51	"""Base search index query to compute availability.
52
53	:returns: search index query.
54	"""
55	must_not_filters = [	1✔
56	# should not be masked
57	Q("term", _masked=True),
58	# should not be in_transit (even without loan)
59	Q("term", status=ItemStatus.IN_TRANSIT),
60	# if issue the status should be received
61	Q("exists", field="issue") & ~Q("term", issue__status="received"),
62	]
63
64	if not_available_item_types := [	1✔
65	hit.pid for hit in ItemTypesSearch().source("pid").filter("term", negative_availability=True).scan()
66	]:
67	# negative availability item type and not temporary item types
68	has_items_filters = Q("terms", item_type__pid=not_available_item_types)	1✔
69	has_items_filters &= ~Q("exists", field="temporary_item_type")	1✔
70	# temporary item types with negative availability
71	has_items_filters \|= Q("terms", temporary_item_type__pid=not_available_item_types)	1✔
72	# add to the must not filters
73	must_not_filters.append(has_items_filters)	1✔
74	return self.filter(Q("bool", must_not=must_not_filters))	1✔
75
76
77	class Item(ItemCirculation, ItemIssue):	1✔
78	"""Item class."""
79
80	minter = item_id_minter	1✔
81	fetcher = item_id_fetcher	1✔
82	provider = ItemProvider	1✔
83	model_cls = ItemMetadata	1✔
84	pids_exist_check = {	1✔
85	"required": {"loc": "location", "doc": "document", "itty": "item_type"},
86	"not_required": {
87	"org": "organisation",
88	# We cannot make the holding required because it is created later
89	"hold": "holding",
90	},
91	}
92
93	def delete(self, force=False, dbcommit=False, delindex=False):	1✔
94	"""Delete record."""
95	from rero_ils.modules.items.tasks import delete_item_in_collections	1✔
96
97	delete_item_in_collections.delay(self.pid, dbcommit=True, reindex=True)	1✔
98	return super().delete(force, dbcommit, delindex)	1✔
99
100	def delete_from_index(self):	1✔
101	"""Delete record from index."""
102	with contextlib.suppress(NotFoundError, ValueError):	1✔
103	ItemsIndexer().delete(self)	1✔
104
105	def reasons_not_to_delete(self):	1✔
106	"""Get reasons not to delete record."""
107	cannot_delete = {}	1✔
108	links = self.get_links_to_me()	1✔
109	# local_fields and collections aren't a reason to block suppression
110	links.pop("local_fields", None)	1✔
111	links.pop("collections", None)	1✔
112	if links:	1✔
113	cannot_delete["links"] = links	1✔
114	return cannot_delete	1✔
115
116	def in_collection(self, **kwargs):	1✔
117	"""Get published collection pids for current item."""
118	from ...collections.api import CollectionsSearch	1✔
119
120	output = []	1✔
121	search = (	1✔
122	CollectionsSearch()
123	.filter("term", items__pid=self.get("pid"))
124	.filter("term", published=True)
125	.sort({"title_sort": {"order": "asc"}})
126	.params(preserve_order=True)
127	.source(["pid", "organisation", "title", "description"])
128	)
129	orgs = {}	1✔
130	for hit in search.scan():	1✔
131	hit = hit.to_dict()	×
132	org_pid = hit["organisation"]["pid"]	×
133	if org_pid not in orgs:	×
134	orgs[org_pid] = Organisation.get_record_by_pid(org_pid)	×
135	collection_data = {	×
136	"pid": hit["pid"], # required property
137	"title": hit["title"], # required property
138	"description": hit.get("description"), # optional property
139	"viewcode": orgs[org_pid].get("code"),
140	}
141	collection_data = {k: v for k, v in collection_data.items() if v}	×
142	output.append(collection_data)	×
143	return output	1✔
144
145	def replace_refs(self):	1✔
146	"""Replace $ref with real data."""
147	tmp_itty_end_date = self.get("temporary_item_type", {}).get("end_date")	1✔
148	tmp_loc_end_date = self.get("temporary_location", {}).get("end_date")	1✔
149	data = super().replace_refs()	1✔
150	if tmp_itty_end_date:	1✔
151	data["temporary_item_type"]["end_date"] = tmp_itty_end_date	1✔
152	if tmp_loc_end_date:	1✔
153	data["temporary_location"]["end_date"] = tmp_loc_end_date	1✔
154	return data	1✔
155
156	@classmethod	1✔
157	def get_item_record_for_ui(cls, **kwargs):	1✔
158	"""Return the item record for ui calls.
159
160	retrieving the item record is possible from an item pid, barcode or
161	from a loan.
162
163	:param kwargs: contains at least one of item_pid, item_barcode, or pid.
164	:return: the item record.
165	"""
166	from ...loans.api import Loan	1✔
167
168	item = None	1✔
169	item_pid = kwargs.get("item_pid")	1✔
170	item_barcode = kwargs.pop("item_barcode", None)	1✔
171	loan_pid = kwargs.get("pid")	1✔
172	if item_pid:	1✔
173	item = Item.get_record_by_pid(item_pid)	1✔
174	elif item_barcode:	1✔
175	org_pid = kwargs.get("organisation_pid", current_librarian.organisation_pid)	1✔
176	item = Item.get_item_by_barcode(item_barcode, org_pid)	1✔
177	elif loan_pid:	1✔
178	item_pid = Loan.get_record_by_pid(loan_pid).item_pid	1✔
179	item = Item.get_record_by_pid(item_pid)	1✔
180	return item	1✔
181
182	@classmethod	1✔
183	def format_end_date(cls, end_date):	1✔
184	"""Return a formatted end date."""
185	# (`datetime.now(timezone.utc)` by default)
186	if end_date is None:	1✔
187	end_date = datetime.now(UTC)	1✔
188	return end_date.strftime("%Y-%m-%d")	1✔
189
190	@classmethod	1✔
191	def get_items_with_obsolete_temporary_item_type_or_location(cls, end_date=None):	1✔
192	"""Get all items with an obsolete temporary item_type or location.
193
194	An end_date could be attached to the item temporary item_type or
195	temporary location. If this date is less or equal to sysdate, then the
196	temporary_item_type or temorary_location must be considered as obsolete
197	and the field must be removed.
198
199	:param end_date: the end_date to check.
200	:return A generator of `ItemRecord` object.
201	"""
202	end_date = cls.format_end_date(end_date)	1✔
203	items_query = ItemsSearch()	1✔
204	loc_search_query = items_query.filter("range", temporary_location__end_date={"lte": end_date})	1✔
205	locs = [(hit.meta.id, "loc") for hit in loc_search_query.source("pid").scan()]	1✔
206
207	itty_search_query = items_query.filter("range", temporary_item_type__end_date={"lte": end_date})	1✔
208	itty = [(hit.meta.id, "itty") for hit in itty_search_query.source("pid").scan()]	1✔
209	hits = itty + locs	1✔
210	for id, field_type in hits:	1✔
211	yield Item.get_record(id), field_type	1✔
212
213
214	class ItemsIndexer(IlsRecordsIndexer):	1✔
215	"""Indexing items in search index."""
216
217	record_cls = Item	1✔
218
219	@classmethod	1✔
220	def _search_item(cls, record):	1✔
221	"""Get the item from the corresponding index.
222
223	:param record: an item object
224	:returns: the search index document or {}
225	"""
226	try:	1✔
227	search_item = current_search_client.get(ItemsSearch.Meta.index, record.id)	1✔
228	return search_item["_source"]	1✔
229	except NotFoundError:	1✔
230	return {}	1✔
231
232	@classmethod	1✔
233	def _update_status_in_doc(cls, record, search_item):	1✔
234	"""Update the status of a given item in the document index.
235
236	:param record: an item object
237	:param search_item: a dict of the search index item
238	"""
239	# retrieve the document in the corresponding es index
240	document_pid = extracted_data_from_ref(record.get("document"))	1✔
241	doc = next(DocumentsSearch().extra(version=True).filter("term", pid=document_pid).scan())	1✔
242	# update the item status in the document
243	data = doc.to_dict()	1✔
244	for hold in data.get("holdings", []):	1✔
245	for item in hold.get("items", []):	1✔
246	if item["pid"] == record.pid:	1✔
247	item["status"] = record["status"]	1✔
248	break	1✔
249	else:
250	continue	1✔
251	break	1✔
252	# reindex the document with the same version
253	current_search_client.index(	1✔
254	index=DocumentsSearch.Meta.index,
255	id=doc.meta.id,
256	body=data,
257	version=doc.meta.version,
258	version_type="external_gte",
259	)
260
261	def index(self, record):	1✔
262	"""Index an item.
263
264	:param record: an item object
265	"returns: the elastiscsearch client result
266	"""
267	from ...holdings.api import Holding	1✔
268
269	# get previous indexed version
270	search_item = self._search_item(record)	1✔
271
272	# call the parent
273	return_value = super().index(record)	1✔
274
275	# fast document reindex for circulation operations
276	if search_item and record.get("status") != search_item.get("status"):	1✔
277	self._update_status_in_doc(record, search_item)	1✔
278	return return_value	1✔
279
280	# reindex the holding / doc for non circulation operations
281	holding_pid = extracted_data_from_ref(record.get("holding"))	1✔
282	holding = Holding.get_record_by_pid(holding_pid)	1✔
283	holding.reindex()	1✔
284	# reindex the old holding
285	old_holding_pid = None	1✔
286	if search_item:	1✔
287	# reindex old holding ot update hte count
288	old_holding_pid = search_item.get("holding", {}).get("pid")	1✔
289	if old_holding_pid != holding_pid:	1✔
290	old_holding = Holding.get_record_by_pid(old_holding_pid)	1✔
291	old_holding.reindex()	1✔
292	return return_value	1✔
293
294	def delete(self, record):	1✔
295	"""Delete a record from index.
296
297	:param record: Record instance.
298	"""
299	from rero_ils.modules.holdings.api import Holding	1✔
300
301	return_value = super().delete(record)	1✔
302	holding_pid = extracted_data_from_ref(record.get("holding"))	1✔
303	holding = Holding.get_record_by_pid(holding_pid)	1✔
304	# delete only if a standard item
305	deleted = False	1✔
306	if not holding.is_serial:	1✔
307	with contextlib.suppress(IlsRecordError.NotDeleted):	1✔
308	holding.delete(force=False, dbcommit=True, delindex=True)	1✔
309	deleted = True	1✔
310	if not deleted:	1✔
311	# for items count
312	holding.reindex()	1✔
313	return return_value	1✔
314
315	def bulk_index(self, record_id_iterator):	1✔
316	"""Bulk index records.
317
318	:param record_id_iterator: Iterator yielding record UUIDs.
319	"""
320	super().bulk_index(record_id_iterator, doc_type="item")	1✔

rero / rero-ils / 28170365024

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous