Coverage for oc_meta / run / merge / closure.py: 100%
95 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5"""Definition of the set of entities a merge touches.
7Merging bibliographic resources cascades onto their ``frbr:partOf`` containers
8(issue, volume, journal). When a container is merged it is deleted and every
9reference to it must be redirected, otherwise entities left outside the loaded
10graph keep pointing at a deleted resource. The closure therefore contains:
12- the entities being merged (the seeds);
13- their full ``frbr:partOf`` ancestor chain;
14- the one-hop neighbourhood (both directions, plus responsible-agent role
15 context) of that set, which brings in the siblings and children that refer to
16 any container that could be deleted;
17- every agent role attached to the bibliographic resources in the closure,
18 together with the responsible agent and its identifiers.
20The merge (:mod:`oc_meta.run.merge.entities`) uses it to decide what a batch
21must import so that every entity it mutates is in memory.
22"""
24from __future__ import annotations
26import logging
27from typing import Iterable, List, Set
29from SPARQLWrapper import POST
31from oc_meta.lib.sparql import execute_sparql
33LOGGER = logging.getLogger(__name__)
35FRBR_PART_OF = "http://purl.org/vocab/frbr/core#partOf"
36PRO_IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy"
37PRO_IS_DOCUMENT_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor"
38PRO_WITH_ROLE = "http://purl.org/spar/pro/withRole"
39PRO_PUBLISHER = "http://purl.org/spar/pro/publisher"
40DATACITE_HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier"
41IDENTIFIER_PROGRESS_INTERVAL = 100_000
44def _batched(items: List[str], batch_size: int) -> Iterable[List[str]]:
45 for start in range(0, len(items), batch_size):
46 yield items[start : start + batch_size]
49def _partof_ancestors(endpoint: str, seeds: Set[str], batch_size: int) -> Set[str]:
50 ancestors: Set[str] = set()
51 frontier = list(seeds)
52 while frontier:
53 parents: Set[str] = set()
54 for batch in _batched(frontier, batch_size):
55 clauses = " UNION ".join(
56 f"{{<{uri}> <{FRBR_PART_OF}> ?ancestor}}" for uri in batch
57 )
58 query = f"SELECT DISTINCT ?ancestor WHERE {{ {clauses} }}"
59 results = execute_sparql(endpoint, query, max_retries=5, backoff_factor=0.3)
60 for result in results["results"]["bindings"]:
61 if result["ancestor"]["type"] == "uri":
62 parents.add(result["ancestor"]["value"])
63 frontier = list(parents - ancestors - seeds)
64 ancestors |= parents
65 return ancestors - seeds
68def _one_hop_neighbours(endpoint: str, entities: Set[str], batch_size: int) -> Set[str]:
69 neighbours: Set[str] = set()
70 ordered = list(entities)
71 for batch in _batched(ordered, batch_size):
72 subject_clauses = []
73 object_clauses = []
74 agent_context_clauses = []
75 for uri in batch:
76 subject_clauses.append(f"{{?entity ?p <{uri}>}}")
77 object_clauses.append(f"{{<{uri}> ?p ?entity}}")
78 agent_context_clauses.append(f"{{?entity pro:isHeldBy <{uri}>}}")
79 agent_context_clauses.append(
80 f"{{?agent_role pro:isHeldBy <{uri}> . ?entity pro:isDocumentContextFor ?agent_role}}"
81 )
83 query = f"""
84 PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
85 PREFIX datacite: <http://purl.org/spar/datacite/>
86 PREFIX pro: <http://purl.org/spar/pro/>
87 SELECT DISTINCT ?entity WHERE {{
88 {{
89 {{
90 {" UNION ".join(subject_clauses + object_clauses)}
91 }}
92 FILTER (?p != rdf:type)
93 FILTER (?p != datacite:usesIdentifierScheme)
94 FILTER (?p != pro:withRole)
95 }}
96 UNION
97 {{
98 {" UNION ".join(agent_context_clauses)}
99 }}
100 ?entity ?p2 ?o2 .
101 }}
102 """
103 results = execute_sparql(endpoint, query, max_retries=5, backoff_factor=0.3)
104 for result in results["results"]["bindings"]:
105 if result["entity"]["type"] == "uri":
106 neighbours.add(result["entity"]["value"])
107 return neighbours
110def _publisher_agents(endpoint: str, entities: Set[str], batch_size: int) -> Set[str]:
111 agents: Set[str] = set()
112 ordered = list(entities)
113 for batch in _batched(ordered, batch_size):
114 role_clauses = " UNION ".join(
115 f"{{<{uri}> <{PRO_WITH_ROLE}> <{PRO_PUBLISHER}> . <{uri}> <{PRO_IS_HELD_BY}> ?agent}}"
116 for uri in batch
117 )
118 query = f"""
119 SELECT DISTINCT ?agent ?identifier WHERE {{
120 {{ {role_clauses} }}
121 OPTIONAL {{ ?agent <{DATACITE_HAS_IDENTIFIER}> ?identifier }}
122 }}
123 """
124 results = execute_sparql(endpoint, query, max_retries=5, backoff_factor=0.3)
125 for result in results["results"]["bindings"]:
126 if result["agent"]["type"] == "uri":
127 agents.add(result["agent"]["value"])
128 if "identifier" in result and result["identifier"]["type"] == "uri":
129 agents.add(result["identifier"]["value"])
130 return agents
133def _br_role_context(endpoint: str, entities: Set[str], batch_size: int) -> Set[str]:
134 context: Set[str] = set()
135 ordered = list(entities)
136 for batch in _batched(ordered, batch_size):
137 role_clauses = " UNION ".join(
138 f"{{<{uri}> <{PRO_IS_DOCUMENT_CONTEXT_FOR}> ?role}}" for uri in batch
139 )
140 query = f"""
141 SELECT DISTINCT ?entity WHERE {{
142 {{
143 {role_clauses}
144 BIND(?role AS ?entity)
145 }}
146 UNION
147 {{
148 {role_clauses}
149 ?role <{PRO_IS_HELD_BY}> ?entity .
150 }}
151 UNION
152 {{
153 {role_clauses}
154 ?role <{PRO_IS_HELD_BY}> ?agent .
155 ?agent <{DATACITE_HAS_IDENTIFIER}> ?entity .
156 }}
157 ?entity ?p ?o .
158 }}
159 """
160 results = execute_sparql(endpoint, query, max_retries=5, backoff_factor=0.3)
161 for result in results["results"]["bindings"]:
162 if result["entity"]["type"] == "uri":
163 context.add(result["entity"]["value"])
164 return context
167def compute_related_closure(
168 endpoint: str, entities: Iterable[str], batch_size: int = 10
169) -> Set[str]:
170 """Return every entity touched by merging ``entities`` (seeds included).
172 The result is the seeds, their ``frbr:partOf`` ancestors, the one-hop
173 neighbourhood of that set, every role attached to loaded bibliographic
174 resources, and the responsible agents (with their identifiers) behind those
175 roles. It bounds the cascade: ancestors are followed only through
176 ``frbr:partOf`` (issue -> volume -> journal), the neighbourhood is expanded
177 a single time, so citation chains are not traversed.
178 """
179 seeds = set(entities)
180 core = seeds | _partof_ancestors(endpoint, seeds, batch_size)
181 related = core | _one_hop_neighbours(endpoint, core, batch_size)
182 role_context = _br_role_context(endpoint, related, batch_size)
183 return (
184 related
185 | role_context
186 | _publisher_agents(endpoint, related | role_context, batch_size)
187 )
190def compute_identifier_merge_closure(
191 endpoint: str,
192 surviving_identifiers: Iterable[str],
193 merged_identifiers: Iterable[str],
194 batch_size: int = 1000,
195) -> Set[str]:
196 closure = set(surviving_identifiers)
197 merged = set(merged_identifiers)
198 closure.update(merged)
199 merged_list = list(merged)
201 for batch_number, batch in enumerate(_batched(merged_list, batch_size), start=1):
202 values = " ".join(f"<{identifier}>" for identifier in batch)
203 query = f"""
204 SELECT DISTINCT ?entity WHERE {{
205 VALUES ?identifier {{ {values} }}
206 ?entity <{DATACITE_HAS_IDENTIFIER}> ?identifier .
207 }}
208 """
209 results = execute_sparql(
210 endpoint,
211 query,
212 max_retries=5,
213 backoff_factor=0.3,
214 method=POST,
215 )
216 for result in results["results"]["bindings"]:
217 if result["entity"]["type"] == "uri":
218 closure.add(result["entity"]["value"])
220 processed_entities = min(batch_number * batch_size, len(merged_list))
221 if (
222 processed_entities % IDENTIFIER_PROGRESS_INTERVAL == 0
223 or processed_entities == len(merged_list)
224 ):
225 LOGGER.info(
226 "Computed identifier references for %s of %s merged identifiers",
227 processed_entities,
228 len(merged_list),
229 )
231 return closure