Coverage for oc_meta / run / patches / fix_misplaced_editor_ars.py: 67%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import argparse 

8import json 

9import multiprocessing 

10import os 

11import signal 

12from collections import defaultdict 

13from collections.abc import Callable 

14from concurrent.futures import ProcessPoolExecutor, as_completed 

15from zipfile import ZipFile 

16 

17import orjson 

18import yaml 

19from oc_ocdm.graph import GraphSet 

20from oc_ocdm.graph.graph_entity import GraphEntity 

21from oc_ocdm.support import get_prefix 

22from rich.progress import Progress, TaskID 

23from rich_argparse import RichHelpFormatter 

24 

25from oc_meta.constants import CONTAINER_EDITOR_TYPES 

26from oc_meta.core.editor import MetaEditor 

27from oc_meta.lib.console import console, create_progress 

28from oc_meta.lib.file_manager import collect_files, collect_zip_files, find_rdf_file 

29from oc_meta.lib.finder import ResourceFinder 

30 

31FRBR_PART_OF = "http://purl.org/vocab/frbr/core#partOf" 

32IS_DOC_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor" 

33WITH_ROLE = "http://purl.org/spar/pro/withRole" 

34EDITOR_ROLE = "http://purl.org/spar/pro/editor" 

35IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy" 

36HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier" 

37USES_SCHEME = "http://purl.org/spar/datacite/usesIdentifierScheme" 

38HAS_LITERAL_VALUE = ( 

39 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue" 

40) 

41FOAF_FAMILY_NAME = "http://xmlns.com/foaf/0.1/familyName" 

42FOAF_GIVEN_NAME = "http://xmlns.com/foaf/0.1/givenName" 

43FOAF_NAME = "http://xmlns.com/foaf/0.1/name" 

44 

45CONTAINER_EDITOR_TYPE_IRIS = frozenset( 

46 iri 

47 for iri, label in ResourceFinder._IRI_TO_TYPE.items() 

48 if label in CONTAINER_EDITOR_TYPES 

49) 

50 

51BATCH_SIZE = 100 

52 

53_stop_requested = False 

54 

55 

56def _worker_init() -> None: 

57 signal.signal(signal.SIGINT, signal.SIG_IGN) 

58 signal.signal(signal.SIGTERM, signal.SIG_IGN) 

59 

60 

61def _handle_signal(_signum: int, _frame: object) -> None: 

62 global _stop_requested 

63 _stop_requested = True 

64 console.print("[yellow]Interrupt received, finishing current entity...[/yellow]") 

65 

66 

67def _group_by_file( 

68 uris: set[str], rdf_dir: str, dir_split: int, items_per_file: int, zip_output: bool 

69) -> dict[str, set[str]]: 

70 file_to_uris: dict[str, set[str]] = defaultdict(set) 

71 for uri in uris: 

72 fpath = find_rdf_file(uri, rdf_dir, dir_split, items_per_file, zip_output) 

73 file_to_uris[fpath].add(uri) 

74 return dict(file_to_uris) 

75 

76 

77def _make_targeted_batches( 

78 file_targets: dict[str, set[str]], batch_size: int 

79) -> list[list[tuple[str, frozenset[str]]]]: 

80 items = [(fpath, frozenset(uris)) for fpath, uris in file_targets.items()] 

81 return [items[i : i + batch_size] for i in range(0, len(items), batch_size)] 

82 

83 

84def _read_file(fpath: str, zip_output: bool) -> list: 

85 if zip_output: 

86 with ZipFile(fpath, "r") as zf: 

87 return orjson.loads(zf.read(zf.namelist()[0])) 

88 with open(fpath, "rb") as f: 

89 return orjson.loads(f.read()) 

90 

91 

92def _iter_entities(files: list, zip_output: bool): 

93 for fpath in files: 

94 for graph in _read_file(fpath, zip_output): 

95 for entity in graph.get("@graph", []): 

96 yield entity 

97 

98 

99def _scan_br_content_batch( 

100 files: list[str], zip_output: bool 

101) -> tuple[dict[str, list[str]], dict[str, set[str]]]: 

102 frbr_part_of: dict[str, list[str]] = {} 

103 content_ars: dict[str, set[str]] = {} 

104 for entity in _iter_entities(files, zip_output): 

105 eid = entity["@id"] 

106 entity_types = frozenset(entity.get("@type", [])) 

107 if entity_types & CONTAINER_EDITOR_TYPE_IRIS and FRBR_PART_OF in entity: 

108 frbr_part_of[eid] = [p["@id"] for p in entity[FRBR_PART_OF]] 

109 if IS_DOC_CONTEXT_FOR in entity: 

110 content_ars[eid] = {x["@id"] for x in entity[IS_DOC_CONTEXT_FOR]} 

111 return frbr_part_of, content_ars 

112 

113 

114def _scan_ar_editors_batch(files: list[str], zip_output: bool) -> dict[str, str]: 

115 editors: dict[str, str] = {} 

116 for entity in _iter_entities(files, zip_output): 

117 if WITH_ROLE in entity and entity[WITH_ROLE][0]["@id"] == EDITOR_ROLE: 

118 editors[entity["@id"]] = entity[IS_HELD_BY][0]["@id"] 

119 return editors 

120 

121 

122def _scan_container_ars_batch( 

123 file_targets: list[tuple[str, frozenset[str]]], zip_output: bool 

124) -> dict[str, set[str]]: 

125 result: dict[str, set[str]] = {} 

126 for fpath, targets in file_targets: 

127 for graph in _read_file(fpath, zip_output): 

128 for entity in graph.get("@graph", []): 

129 eid = entity["@id"] 

130 if eid in targets and IS_DOC_CONTEXT_FOR in entity: 

131 result[eid] = {x["@id"] for x in entity[IS_DOC_CONTEXT_FOR]} 

132 return result 

133 

134 

135def _scan_ra_info_batch( 

136 file_targets: list[tuple[str, frozenset[str]]], zip_output: bool 

137) -> tuple[dict[str, set[str]], dict[str, str]]: 

138 id_result: dict[str, set[str]] = {} 

139 name_result: dict[str, str] = {} 

140 for fpath, targets in file_targets: 

141 for graph in _read_file(fpath, zip_output): 

142 for entity in graph.get("@graph", []): 

143 eid = entity["@id"] 

144 if eid not in targets: 

145 continue 

146 if HAS_IDENTIFIER in entity: 

147 id_result[eid] = {x["@id"] for x in entity[HAS_IDENTIFIER]} 

148 family = entity.get(FOAF_FAMILY_NAME, [{}])[0].get("@value", "") 

149 given = entity.get(FOAF_GIVEN_NAME, [{}])[0].get("@value", "") 

150 if family: 

151 name = ( 

152 f"{family.lower()}, {given.lower()}" 

153 if given 

154 else family.lower() 

155 ) 

156 else: 

157 full = entity.get(FOAF_NAME, [{}])[0].get("@value", "") 

158 name = full.lower() if full else "" 

159 if name: 

160 name_result[eid] = name 

161 return id_result, name_result 

162 

163 

164def _scan_id_values_batch( 

165 file_targets: list[tuple[str, frozenset[str]]], zip_output: bool 

166) -> dict[str, str]: 

167 result: dict[str, str] = {} 

168 for fpath, targets in file_targets: 

169 for graph in _read_file(fpath, zip_output): 

170 for entity in graph.get("@graph", []): 

171 eid = entity["@id"] 

172 if eid in targets and USES_SCHEME in entity: 

173 scheme_iri = entity[USES_SCHEME][0]["@id"] 

174 scheme_name = scheme_iri.rsplit("/", 1)[-1] 

175 value = entity[HAS_LITERAL_VALUE][0]["@value"] 

176 result[eid] = f"{scheme_name}:{value}" 

177 return result 

178 

179 

180def _run_parallel( 

181 executor: ProcessPoolExecutor, 

182 scan_fn: Callable, 

183 batches: list, 

184 zip_output: bool, 

185 progress: Progress, 

186 task_id: TaskID, 

187) -> list: 

188 futures = {executor.submit(scan_fn, batch, zip_output): batch for batch in batches} 

189 results = [] 

190 try: 

191 for future in as_completed(futures): 

192 results.append(future.result()) 

193 progress.advance(task_id, len(futures[future])) 

194 except KeyboardInterrupt: 

195 for f in futures: 

196 f.cancel() 

197 raise 

198 finally: 

199 executor.shutdown(wait=False, cancel_futures=True) 

200 return results 

201 

202 

203def _classify_actions( 

204 container_to_contents: dict[str, list[str]], 

205 content_editor_ars: dict[str, set[str]], 

206 editor_ar_to_ra: dict[str, str], 

207 ra_identifiers: dict[str, set[str]], 

208 ra_names: dict[str, str], 

209 container_editor_ars: dict[str, set[str]], 

210) -> list[dict]: 

211 results: list[dict] = [] 

212 for container, contents in container_to_contents.items(): 

213 known_ras: set[str] = set() 

214 known_ids: set[str] = set() 

215 known_names: set[str] = set() 

216 for ar in container_editor_ars.get(container, set()): 

217 ra = editor_ar_to_ra[ar] 

218 known_ras.add(ra) 

219 known_ids.update(ra_identifiers.get(ra, set())) 

220 name = ra_names.get(ra, "") 

221 if name: 

222 known_names.add(name) 

223 

224 for content in sorted(contents): 

225 for ar in sorted(content_editor_ars[content]): 

226 ra = editor_ar_to_ra[ar] 

227 ids = ra_identifiers.get(ra, set()) 

228 name = ra_names.get(ra, "") 

229 

230 if ra in known_ras: 

231 action = "skip_duplicate_ra" 

232 match_reason = ra 

233 elif ids & known_ids: 

234 action = "skip_duplicate_id" 

235 match_reason = next(iter(ids & known_ids)) 

236 elif name and name in known_names: 

237 action = "skip_duplicate_name" 

238 match_reason = name 

239 else: 

240 action = "move" 

241 match_reason = None 

242 known_ras.add(ra) 

243 known_ids.update(ids) 

244 if name: 

245 known_names.add(name) 

246 

247 results.append( 

248 { 

249 "content": content, 

250 "container": container, 

251 "ar": ar, 

252 "ra": ra, 

253 "identifiers": sorted(ids), 

254 "action": action, 

255 "match_reason": match_reason, 

256 } 

257 ) 

258 return results 

259 

260 

261def find_misplaced_editor_ars( 

262 base_dir: str, 

263 zip_output: bool, 

264 dir_split: int, 

265 items_per_file: int, 

266 workers: int = 4, 

267 batch_size: int = BATCH_SIZE, 

268) -> tuple[list[dict], dict[str, set[str]]]: 

269 br_dir = os.path.join(base_dir, "br") 

270 ar_dir = os.path.join(base_dir, "ar") 

271 

272 if zip_output: 

273 br_files = collect_zip_files(br_dir, only_data=True) 

274 ar_files = collect_zip_files(ar_dir, only_data=True) 

275 else: 

276 br_files = collect_files(br_dir, "*.json", lambda p: "prov" not in p) 

277 ar_files = collect_files(ar_dir, "*.json", lambda p: "prov" not in p) 

278 

279 br_batches = [ 

280 br_files[i : i + batch_size] for i in range(0, len(br_files), batch_size) 

281 ] 

282 ar_batches = [ 

283 ar_files[i : i + batch_size] for i in range(0, len(ar_files), batch_size) 

284 ] 

285 

286 frbr_part_of: dict[str, list[str]] = {} 

287 content_ars: dict[str, set[str]] = {} 

288 

289 ctx = multiprocessing.get_context("forkserver") 

290 

291 with create_progress() as progress: 

292 br_task = progress.add_task("Scanning BR files", total=len(br_files)) 

293 executor = ProcessPoolExecutor( 

294 max_workers=workers, initializer=_worker_init, mp_context=ctx 

295 ) 

296 for partial_frbr, partial_content_ars in _run_parallel( 

297 executor, _scan_br_content_batch, br_batches, zip_output, progress, br_task 

298 ): 

299 frbr_part_of.update(partial_frbr) 

300 content_ars.update(partial_content_ars) 

301 

302 console.print( 

303 f"BR scan complete: [cyan]{len(frbr_part_of)}[/cyan] content entities, " 

304 f"[cyan]{sum(len(v) for v in content_ars.values())}[/cyan] content ARs" 

305 ) 

306 

307 ar_task = progress.add_task("Scanning AR files", total=len(ar_files)) 

308 executor = ProcessPoolExecutor( 

309 max_workers=workers, initializer=_worker_init, mp_context=ctx 

310 ) 

311 editor_ar_to_ra: dict[str, str] = {} 

312 for partial in _run_parallel( 

313 executor, _scan_ar_editors_batch, ar_batches, zip_output, progress, ar_task 

314 ): 

315 editor_ar_to_ra.update(partial) 

316 

317 console.print( 

318 f"AR scan complete: [cyan]{len(editor_ar_to_ra)}[/cyan] editor ARs found" 

319 ) 

320 

321 # Identify misplaced editor ARs per content entity 

322 content_editor_ars: dict[str, set[str]] = {} 

323 for content, ars in content_ars.items(): 

324 editors = ars & editor_ar_to_ra.keys() 

325 if editors: 

326 content_editor_ars[content] = editors 

327 

328 container_to_contents: dict[str, list[str]] = defaultdict(list) 

329 for content in content_editor_ars: 

330 for container in frbr_part_of[content]: 

331 container_to_contents[container].append(content) 

332 

333 container_uris = set(container_to_contents.keys()) 

334 console.print( 

335 f"Identified [cyan]{sum(len(v) for v in content_editor_ars.values())}[/cyan] " 

336 f"misplaced editor ARs across [cyan]{len(content_editor_ars)}[/cyan] content " 

337 f"entities in [cyan]{len(container_uris)}[/cyan] containers" 

338 ) 

339 

340 container_file_targets = _group_by_file( 

341 container_uris, base_dir, dir_split, items_per_file, zip_output 

342 ) 

343 container_batches = _make_targeted_batches(container_file_targets, batch_size) 

344 

345 container_ars: dict[str, set[str]] = {} 

346 container_editor_ars: dict[str, set[str]] = {} 

347 

348 with create_progress() as progress: 

349 ct_task = progress.add_task( 

350 "Scanning container ARs", total=len(container_file_targets) 

351 ) 

352 executor = ProcessPoolExecutor( 

353 max_workers=workers, initializer=_worker_init, mp_context=ctx 

354 ) 

355 for partial in _run_parallel( 

356 executor, 

357 _scan_container_ars_batch, 

358 container_batches, 

359 zip_output, 

360 progress, 

361 ct_task, 

362 ): 

363 container_ars.update(partial) 

364 

365 for container, ars in container_ars.items(): 

366 editors = ars & editor_ar_to_ra.keys() 

367 if editors: 

368 container_editor_ars[container] = editors 

369 

370 console.print( 

371 f"Container scan: [cyan]{len(container_editor_ars)}[/cyan] containers " 

372 f"already have editor ARs" 

373 ) 

374 

375 # Determine which containers need dedup and collect RA identifiers 

376 containers_needing_dedup: set[str] = set() 

377 for container, contents in container_to_contents.items(): 

378 if ( 

379 container in container_editor_ars 

380 or len(contents) > 1 

381 or any(len(content_editor_ars[c]) > 1 for c in contents) 

382 ): 

383 containers_needing_dedup.add(container) 

384 

385 ras_needing_ids: set[str] = set() 

386 for container in containers_needing_dedup: 

387 for ar in container_editor_ars.get(container, set()): 

388 ras_needing_ids.add(editor_ar_to_ra[ar]) 

389 for content in container_to_contents[container]: 

390 for ar in content_editor_ars[content]: 

391 ras_needing_ids.add(editor_ar_to_ra[ar]) 

392 

393 ra_identifiers: dict[str, set[str]] = {} 

394 ra_names: dict[str, str] = {} 

395 

396 if ras_needing_ids: 

397 console.print( 

398 f"Collecting info for [cyan]{len(ras_needing_ids)}[/cyan] RAs " 

399 f"across [cyan]{len(containers_needing_dedup)}[/cyan] containers needing dedup" 

400 ) 

401 ra_file_targets = _group_by_file( 

402 ras_needing_ids, base_dir, dir_split, items_per_file, zip_output 

403 ) 

404 ra_batches = _make_targeted_batches(ra_file_targets, batch_size) 

405 

406 ra_to_id_uris: dict[str, set[str]] = {} 

407 with create_progress() as progress: 

408 ra_task = progress.add_task( 

409 "Collecting RA info", total=len(ra_file_targets) 

410 ) 

411 executor = ProcessPoolExecutor( 

412 max_workers=workers, initializer=_worker_init, mp_context=ctx 

413 ) 

414 for partial_ids, partial_names in _run_parallel( 

415 executor, _scan_ra_info_batch, ra_batches, zip_output, progress, ra_task 

416 ): 

417 ra_to_id_uris.update(partial_ids) 

418 ra_names.update(partial_names) 

419 

420 all_id_uris: set[str] = set() 

421 for ids in ra_to_id_uris.values(): 

422 all_id_uris.update(ids) 

423 

424 if all_id_uris: 

425 id_file_targets = _group_by_file( 

426 all_id_uris, base_dir, dir_split, items_per_file, zip_output 

427 ) 

428 id_batches = _make_targeted_batches(id_file_targets, batch_size) 

429 

430 id_to_value: dict[str, str] = {} 

431 with create_progress() as progress: 

432 id_task = progress.add_task( 

433 "Collecting ID values", total=len(id_file_targets) 

434 ) 

435 executor = ProcessPoolExecutor( 

436 max_workers=workers, initializer=_worker_init, mp_context=ctx 

437 ) 

438 for partial in _run_parallel( 

439 executor, 

440 _scan_id_values_batch, 

441 id_batches, 

442 zip_output, 

443 progress, 

444 id_task, 

445 ): 

446 id_to_value.update(partial) 

447 

448 for ra, id_uris in ra_to_id_uris.items(): 

449 ids = { 

450 id_to_value[id_uri] for id_uri in id_uris if id_uri in id_to_value 

451 } 

452 if ids: 

453 ra_identifiers[ra] = ids 

454 

455 console.print( 

456 f"RA info complete: [cyan]{len(ra_identifiers)}[/cyan] with identifiers, " 

457 f"[cyan]{len(ra_names)}[/cyan] with names" 

458 ) 

459 

460 return _classify_actions( 

461 dict(container_to_contents), 

462 content_editor_ars, 

463 editor_ar_to_ra, 

464 ra_identifiers, 

465 ra_names, 

466 container_editor_ars, 

467 ), container_editor_ars 

468 

469 

470def fix_container( 

471 editor: MetaEditor, 

472 container_uri: str, 

473 content_actions: list[tuple[str, list[tuple[str, str]], list[tuple[str, str]]]], 

474 existing_ars: set[str], 

475) -> None: 

476 supplier_prefix = get_prefix(container_uri) 

477 g_set = GraphSet( 

478 editor.base_iri, 

479 supplier_prefix=supplier_prefix, 

480 custom_counter_handler=editor.counter_handler, 

481 wanted_label=False, 

482 ) 

483 

484 file_paths: set[str] = set() 

485 all_uris = [container_uri] 

486 for content_uri, move_ars, skip_ars in content_actions: 

487 all_uris.append(content_uri) 

488 for ar_uri, ra_uri in move_ars + skip_ars: 

489 all_uris.append(ar_uri) 

490 all_uris.append(ra_uri) 

491 all_uris.extend(existing_ars) 

492 

493 for uri in all_uris: 

494 fp = find_rdf_file( 

495 uri, 

496 editor.base_dir, 

497 editor.dir_split, 

498 editor.n_file_item, 

499 editor.zip_output_rdf, 

500 ) 

501 file_paths.add(fp) 

502 

503 for fp in file_paths: 

504 imported_graph = editor.reader.load(fp) 

505 if imported_graph is not None: 

506 editor.reader.import_entities_from_graph( 

507 g_set, imported_graph, editor.resp_agent 

508 ) 

509 

510 container_entity = g_set.get_entity(container_uri) 

511 assert container_entity is not None, f"container not found: {container_uri}" 

512 

513 all_move_entities = [] 

514 

515 for content_uri, move_ars, skip_ars in content_actions: 

516 content_entity = g_set.get_entity(content_uri) 

517 assert content_entity is not None, f"content not found: {content_uri}" 

518 

519 contributor_uris = { 

520 o.value 

521 for _, _, o in content_entity.g.triples( 

522 ( 

523 content_entity.res, 

524 GraphEntity.iri_is_document_context_for, 

525 None, 

526 ) 

527 ) 

528 } 

529 ars_on_content = { 

530 ar_uri for ar_uri, _ in move_ars + skip_ars if ar_uri in contributor_uris 

531 } 

532 

533 for ar_uri, ra_uri in move_ars + skip_ars: 

534 if ar_uri in ars_on_content: 

535 ar_entity = g_set.get_entity(ar_uri) 

536 content_entity.remove_contributor(ar_entity) # type: ignore[attr-defined] 

537 ar_entity.remove_next() # type: ignore[attr-defined] 

538 

539 for ar_uri, ra_uri in skip_ars: 

540 if ar_uri in ars_on_content and ar_uri not in existing_ars: 

541 ar_entity = g_set.get_entity(ar_uri) 

542 ar_entity.mark_as_to_be_deleted() # type: ignore[attr-defined] 

543 

544 for ar_uri, ra_uri in move_ars: 

545 if ar_uri in ars_on_content: 

546 ar_entity = g_set.get_entity(ar_uri) 

547 container_entity.has_contributor(ar_entity) # type: ignore[attr-defined] 

548 all_move_entities.append(ar_entity) 

549 else: 

550 ra_entity = g_set.get_entity(ra_uri) 

551 assert ra_entity is not None, f"RA not found: {ra_uri}" 

552 new_ar = g_set.add_ar(editor.resp_agent) 

553 new_ar.create_editor() # type: ignore[attr-defined] 

554 new_ar.is_held_by(ra_entity) # type: ignore[attr-defined] 

555 container_entity.has_contributor(new_ar) # type: ignore[attr-defined] 

556 all_move_entities.append(new_ar) 

557 

558 if existing_ars and all_move_entities: 

559 for ar_uri in sorted(existing_ars): 

560 ar_entity = g_set.get_entity(ar_uri) 

561 if ar_entity is not None and not list( 

562 ar_entity.g.triples((ar_entity.res, GraphEntity.iri_has_next, None)) 

563 ): 

564 ar_entity.has_next(all_move_entities[0]) # type: ignore[attr-defined] 

565 break 

566 

567 for i in range(len(all_move_entities) - 1): 

568 all_move_entities[i].has_next(all_move_entities[i + 1]) # type: ignore[attr-defined] 

569 

570 editor.save(g_set, supplier_prefix) 

571 

572 

573def _load_progress(path: str) -> set[str]: 

574 if not os.path.exists(path): 

575 return set() 

576 with open(path) as f: 

577 return set(json.load(f)) 

578 

579 

580def _save_progress(path: str, completed: set[str]) -> None: 

581 with open(path, "w") as f: 

582 json.dump(list(completed), f) 

583 

584 

585def main() -> None: # pragma: no cover 

586 parser = argparse.ArgumentParser( 

587 description=( 

588 "Fix misplaced editor ARs: move pro:isDocumentContextFor " 

589 "from content entity to its frbr:partOf container" 

590 ), 

591 formatter_class=RichHelpFormatter, 

592 ) 

593 parser.add_argument( 

594 "-c", "--config", required=True, help="Path to meta_config.yaml" 

595 ) 

596 parser.add_argument( 

597 "-r", "--resp-agent", help="Responsible agent URI (required without --dry-run)" 

598 ) 

599 parser.add_argument( 

600 "--dry-run", action="store_true", help="Report cases without modifying" 

601 ) 

602 parser.add_argument( 

603 "-w", 

604 "--workers", 

605 type=int, 

606 default=4, 

607 help="Number of parallel workers for scanning", 

608 ) 

609 parser.add_argument( 

610 "-b", 

611 "--batch-size", 

612 type=int, 

613 default=BATCH_SIZE, 

614 help="Files per batch for scanning", 

615 ) 

616 parser.add_argument( 

617 "--progress-file", 

618 default="fix_misplaced_editor_ars_progress.json", 

619 help=( 

620 "Path to a JSON file used to track completed content entities for resumable execution. " 

621 "Created automatically if it does not exist; deleted on successful completion. " 

622 "Default: fix_misplaced_editor_ars_progress.json in the current working directory." 

623 ), 

624 ) 

625 parser.add_argument( 

626 "--report-file", 

627 default="fix_misplaced_editor_ars_report.json", 

628 ) 

629 args = parser.parse_args() 

630 

631 if not args.dry_run and not args.resp_agent: 

632 parser.error("--resp-agent is required when not using --dry-run") 

633 

634 with open(args.config) as f: 

635 config = yaml.safe_load(f) 

636 

637 rdf_dir = os.path.join(config["base_output_dir"], "rdf") 

638 zip_output = config.get("zip_output_rdf", False) 

639 dir_split = config["dir_split_number"] 

640 items_per_file = config["items_per_file"] 

641 

642 console.print("Scanning RDF files for misplaced editor ARs...") 

643 cases, container_editor_ars = find_misplaced_editor_ars( 

644 rdf_dir, 

645 zip_output, 

646 dir_split, 

647 items_per_file, 

648 args.workers, 

649 args.batch_size, 

650 ) 

651 

652 content_groups: dict[tuple[str, str], list[dict]] = defaultdict(list) 

653 for case in cases: 

654 content_groups[(case["content"], case["container"])].append(case) 

655 

656 container_groups: dict[str, list[dict]] = defaultdict(list) 

657 for case in cases: 

658 container_groups[case["container"]].append(case) 

659 

660 total_ars = len(cases) 

661 move_count = sum(1 for c in cases if c["action"] == "move") 

662 skip_ra_count = sum(1 for c in cases if c["action"] == "skip_duplicate_ra") 

663 skip_id_count = sum(1 for c in cases if c["action"] == "skip_duplicate_id") 

664 skip_name_count = sum(1 for c in cases if c["action"] == "skip_duplicate_name") 

665 

666 console.print( 

667 f"\n[bold]Found [green]{total_ars}[/green] misplaced editor ARs " 

668 f"across [green]{len(content_groups)}[/green] content entities " 

669 f"in [green]{len(container_groups)}[/green] containers[/bold]\n" 

670 f" [green]{move_count}[/green] to move, " 

671 f"[yellow]{skip_ra_count}[/yellow] skip (same RA), " 

672 f"[yellow]{skip_id_count}[/yellow] skip (same identifier), " 

673 f"[yellow]{skip_name_count}[/yellow] skip (same name)" 

674 ) 

675 

676 if args.dry_run: 

677 report = { 

678 "summary": { 

679 "total_misplaced_ars": total_ars, 

680 "affected_content_entities": len(content_groups), 

681 "unique_containers": len(container_groups), 

682 "ars_to_move": move_count, 

683 "ars_skipped_duplicate_ra": skip_ra_count, 

684 "ars_skipped_duplicate_id": skip_id_count, 

685 "ars_skipped_duplicate_name": skip_name_count, 

686 }, 

687 "cases": [ 

688 { 

689 "content": content, 

690 "container": container, 

691 "editor_ars": [ 

692 { 

693 "ar": a["ar"], 

694 "ra": a["ra"], 

695 "identifiers": a["identifiers"], 

696 "action": a["action"], 

697 "match_reason": a["match_reason"], 

698 } 

699 for a in actions 

700 ], 

701 } 

702 for (content, container), actions in sorted(content_groups.items()) 

703 ], 

704 } 

705 with open(args.report_file, "w") as f: 

706 json.dump(report, f, indent=2) 

707 console.print(f"\n[bold]Dry run report written to {args.report_file}[/bold]") 

708 return 

709 

710 completed = _load_progress(args.progress_file) 

711 if completed: 

712 console.print(f"Resuming: {len(completed)} containers already processed") 

713 

714 signal.signal(signal.SIGINT, _handle_signal) 

715 signal.signal(signal.SIGTERM, _handle_signal) 

716 

717 editor = MetaEditor(args.config, args.resp_agent) 

718 succeeded = failed = skipped = 0 

719 

720 with create_progress() as progress: 

721 task = progress.add_task( 

722 "Fixing misplaced editor ARs", total=len(container_groups) 

723 ) 

724 for container, ar_actions in container_groups.items(): 

725 if _stop_requested: 

726 break 

727 if container in completed: 

728 skipped += 1 

729 progress.advance(task) 

730 continue 

731 try: 

732 per_content: dict[ 

733 str, tuple[list[tuple[str, str]], list[tuple[str, str]]]