Coverage for oc_meta / run / find / hasnext_anomalies.py: 49%
180 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
5from __future__ import annotations
7import argparse
8import os
9from collections import Counter
10from datetime import datetime, timezone
11import multiprocessing
12from typing import Dict, List, Optional, Tuple
14import orjson
15import yaml
16from rich_argparse import RichHelpFormatter
18from oc_meta.lib.console import create_progress
19from oc_meta.lib.file_manager import collect_zip_files
20from oc_meta.lib.file_manager import find_rdf_file
21from oc_meta.run.meta.generate_csv import load_json_from_file
23ROLE_MAP = {
24 "http://purl.org/spar/pro/author": "author",
25 "http://purl.org/spar/pro/editor": "editor",
26 "http://purl.org/spar/pro/publisher": "publisher",
27}
28HAS_NEXT = "https://w3id.org/oc/ontology/hasNext"
29IS_DOC_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor"
30WITH_ROLE = "http://purl.org/spar/pro/withRole"
31IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy"
33_worker_config: Optional[Tuple[str, int, int]] = None
36def _init_worker(rdf_dir: str, dir_split_number: int, items_per_file: int) -> None:
37 global _worker_config
38 _worker_config = (rdf_dir, dir_split_number, items_per_file)
41def _ar_summary(ar_uri: str, info: dict) -> dict:
42 return {
43 "ar": ar_uri,
44 "ra": info["ra"],
45 "has_next": info["has_next"],
46 }
49def load_ar_data(
50 ar_uri: str, rdf_dir: str, dir_split_number: int, items_per_file: int
51) -> Optional[dict]:
52 ar_file = find_rdf_file(
53 ar_uri, rdf_dir, dir_split_number, items_per_file, zip_output=True
54 )
55 if not os.path.exists(ar_file):
56 return None
57 data = load_json_from_file(ar_file)
58 for graph in data:
59 for entity in graph["@graph"]:
60 if entity["@id"] == ar_uri:
61 role_uri = ""
62 if WITH_ROLE in entity:
63 role_uri = entity[WITH_ROLE][0]["@id"]
64 role_type = ROLE_MAP.get(role_uri, "unknown")
66 ra_uri = None
67 if IS_HELD_BY in entity:
68 ra_uri = entity[IS_HELD_BY][0]["@id"]
70 has_next = []
71 if HAS_NEXT in entity:
72 has_next = [item["@id"] for item in entity[HAS_NEXT]]
74 return {
75 "role_type": role_type,
76 "ra": ra_uri,
77 "has_next": has_next,
78 }
79 return None
82def detect_cycles(ar_data: Dict[str, dict], ar_uris_in_group: set) -> List[List[str]]:
83 adj: Dict[str, List[str]] = {}
84 for ar_uri, info in ar_data.items():
85 targets = [t for t in info["has_next"] if t in ar_uris_in_group and t != ar_uri]
86 if targets:
87 adj[ar_uri] = targets
89 globally_visited: set = set()
90 cycles: List[List[str]] = []
92 for start in ar_uris_in_group:
93 if start in globally_visited:
94 continue
96 path: List[str] = []
97 path_set: set = set()
98 stack: List[Tuple[str, int]] = [(start, -1)]
100 while stack:
101 node, ni = stack[-1]
103 if ni == -1:
104 if node in path_set:
105 cycle_start = path.index(node)
106 cycles.append(list(path[cycle_start:]))
107 stack.pop()
108 continue
109 if node in globally_visited:
110 stack.pop()
111 continue
112 path.append(node)
113 path_set.add(node)
114 stack[-1] = (node, 0)
115 continue
117 neighbors = adj.get(node, [])
118 if ni < len(neighbors):
119 stack[-1] = (node, ni + 1)
120 stack.append((neighbors[ni], -1))
121 else:
122 path.pop()
123 path_set.discard(node)
124 globally_visited.add(node)
125 stack.pop()
127 return cycles
130def find_anomalies(br_uri: str, role_type: str, ar_data: Dict[str, dict]) -> List[dict]:
131 anomalies: List[dict] = []
132 ar_uris_in_group = set(ar_data.keys())
134 for ar_uri, info in ar_data.items():
135 if ar_uri in info["has_next"]:
136 anomalies.append(
137 {
138 "anomaly_type": "self_loop",
139 "br": br_uri,
140 "role_type": role_type,
141 "ars_involved": [_ar_summary(ar_uri, info)],
142 "details": f"AR {ar_uri.split('/')[-1]} hasNext points to itself",
143 }
144 )
146 for ar_uri, info in ar_data.items():
147 if len(info["has_next"]) > 1:
148 anomalies.append(
149 {
150 "anomaly_type": "multiple_has_next",
151 "br": br_uri,
152 "role_type": role_type,
153 "ars_involved": [_ar_summary(ar_uri, info)],
154 "details": (
155 f"AR {ar_uri.split('/')[-1]} has"
156 f" {len(info['has_next'])} hasNext targets"
157 ),
158 }
159 )
161 for ar_uri, info in ar_data.items():
162 for target in info["has_next"]:
163 if target not in ar_uris_in_group:
164 anomalies.append(
165 {
166 "anomaly_type": "dangling_has_next",
167 "br": br_uri,
168 "role_type": role_type,
169 "ars_involved": [_ar_summary(ar_uri, info)],
170 "details": (
171 f"AR {ar_uri.split('/')[-1]} hasNext points to"
172 f" {target.split('/')[-1]} which is not in this"
173 " BR/role group"
174 ),
175 }
176 )
178 referenced_ars = set()
179 for info in ar_data.values():
180 for target in info["has_next"]:
181 if target in ar_uris_in_group:
182 referenced_ars.add(target)
184 start_nodes = [ar for ar in ar_uris_in_group if ar not in referenced_ars]
186 if len(ar_data) > 1:
187 if len(start_nodes) == 0:
188 anomalies.append(
189 {
190 "anomaly_type": "no_start_node",
191 "br": br_uri,
192 "role_type": role_type,
193 "ars_involved": [
194 _ar_summary(ar_uri, ar_data[ar_uri]) for ar_uri in ar_data
195 ],
196 "details": (
197 f"All {len(ar_data)} ARs are targets of hasNext"
198 " (fully circular)"
199 ),
200 }
201 )
202 elif len(start_nodes) > 1:
203 anomalies.append(
204 {
205 "anomaly_type": "multiple_start_nodes",
206 "br": br_uri,
207 "role_type": role_type,
208 "ars_involved": [
209 _ar_summary(ar_uri, ar_data[ar_uri]) for ar_uri in start_nodes
210 ],
211 "details": (
212 f"{len(start_nodes)} ARs have no incoming hasNext"
213 " (disconnected fragments)"
214 ),
215 }
216 )
218 cycles = detect_cycles(ar_data, ar_uris_in_group)
219 for cycle in cycles:
220 cycle_ids = [uri.split("/")[-1] for uri in cycle]
221 anomalies.append(
222 {
223 "anomaly_type": "cycle",
224 "br": br_uri,
225 "role_type": role_type,
226 "ars_involved": [
227 _ar_summary(ar_uri, ar_data[ar_uri]) for ar_uri in cycle
228 ],
229 "details": (
230 f"{len(cycle)}-node cycle:"
231 f" {' -> '.join(cycle_ids)} -> {cycle_ids[0]}"
232 ),
233 }
234 )
236 return anomalies
239def _detect_anomalies_in_file(filepath: str) -> Tuple[str, int, List[dict]]:
240 assert _worker_config is not None
241 rdf_dir, dir_split_number, items_per_file = _worker_config
242 anomalies: List[dict] = []
243 br_count = 0
244 data = load_json_from_file(filepath)
245 for graph in data:
246 for entity in graph["@graph"]:
247 if IS_DOC_CONTEXT_FOR not in entity:
248 continue
249 br_count += 1
250 br_uri = entity["@id"]
252 ar_uris = [ar["@id"] for ar in entity[IS_DOC_CONTEXT_FOR]]
253 ar_data: Dict[str, dict] = {}
254 for ar_uri in ar_uris:
255 info = load_ar_data(ar_uri, rdf_dir, dir_split_number, items_per_file)
256 if info:
257 ar_data[ar_uri] = info
259 role_groups: Dict[str, Dict[str, dict]] = {}
260 for ar_uri, info in ar_data.items():
261 role = info["role_type"]
262 if role not in role_groups:
263 role_groups[role] = {}
264 role_groups[role][ar_uri] = info
266 for role_type, group in role_groups.items():
267 anomalies.extend(find_anomalies(br_uri, role_type, group))
269 return (filepath, br_count, anomalies)
272def main() -> None:
273 parser = argparse.ArgumentParser(
274 description="Detect hasNext chain anomalies in RDF data",
275 formatter_class=RichHelpFormatter,
276 )
277 parser.add_argument(
278 "-c", "--config", required=True, help="Meta config YAML file path"
279 )
280 parser.add_argument("-o", "--output", required=True, help="Output JSON report path")
281 parser.add_argument(
282 "--workers",
283 type=int,
284 default=4,
285 help="Number of parallel workers (default: 4)",
286 )
287 args = parser.parse_args()
289 with open(args.config, encoding="utf-8") as f:
290 settings = yaml.safe_load(f)
292 rdf_dir = os.path.join(settings["output_rdf_dir"], "rdf")
293 dir_split_number = settings["dir_split_number"]
294 items_per_file = settings["items_per_file"]
296 br_dir = os.path.join(rdf_dir, "br")
297 if not os.path.exists(br_dir):
298 print(f"Error: BR directory not found at {br_dir}")
299 return
301 all_files = collect_zip_files(br_dir, only_data=True)
303 if not all_files:
304 print("No BR zip files found")
305 return
307 print(f"Processing {len(all_files)} BR files with {args.workers} workers...")
309 total_brs = 0
310 all_anomalies: List[dict] = []
312 ctx = multiprocessing.get_context("forkserver")
313 with ctx.Pool(
314 args.workers,
315 _init_worker,
316 (rdf_dir, dir_split_number, items_per_file),
317 ) as pool:
318 with create_progress() as progress:
319 task = progress.add_task("Scanning for anomalies", total=len(all_files))
320 for filepath, br_count, anomalies in pool.imap_unordered(
321 _detect_anomalies_in_file, all_files
322 ):
323 total_brs += br_count
324 all_anomalies.extend(anomalies)
325 progress.update(task, advance=1)
327 anomalies_by_type = dict(Counter(a["anomaly_type"] for a in all_anomalies))
329 report = {
330 "config": os.path.abspath(args.config),
331 "rdf_dir": rdf_dir,
332 "timestamp": datetime.now(timezone.utc).isoformat(),
333 "total_brs_analyzed": total_brs,
334 "total_anomalies": len(all_anomalies),
335 "anomalies_by_type": anomalies_by_type,
336 "anomalies": all_anomalies,
337 }
339 output_dir = os.path.dirname(os.path.abspath(args.output))
340 os.makedirs(output_dir, exist_ok=True)
341 with open(args.output, "wb") as f:
342 f.write(orjson.dumps(report, option=orjson.OPT_INDENT_2))
344 print(f"Analyzed {total_brs} BRs, found {len(all_anomalies)} anomalies")
345 for atype, count in sorted(anomalies_by_type.items()):
346 print(f" {atype}: {count}")
347 print(f"Report saved to {args.output}")
350if __name__ == "__main__":
351 main()