Coverage for changes_metadata_manager / generate_provenance.py: 98%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5""" 

6Module to generate provenance snapshots from RDF data. 

7Loads RDF data in various formats from all files in a directory, 

8extracts all subjects, and creates provenance snapshots as named graphs 

9with type prov:Entity. 

10""" 

11 

12import os 

13import datetime 

14from rdflib import Dataset, URIRef, Namespace, Literal 

15from rdflib.namespace import RDF, XSD, DCTERMS 

16 

17CC0 = URIRef("https://creativecommons.org/publicdomain/zero/1.0/") 

18 

19 

20def generate_provenance_snapshots( 

21 input_directory: str, 

22 output_file: str, 

23 input_format: str | None = None, 

24 output_format: str = "json-ld", 

25 agent_orcid: str = "", 

26 primary_source: str = "", 

27): 

28 """ 

29 Generate provenance snapshots from RDF data. 

30 

31 Args: 

32 input_directory: Path to directory containing RDF files 

33 output_file: Path to output file with provenance snapshots (N-Quads format) 

34 input_format: Optional format to use for all input files (overrides auto-detection) 

35 output_format: Format to use for output file (default: nquads) 

36 agent_orcid: ORCID of the responsible agent 

37 primary_source: URI of the primary source for the data 

38 """ 

39 

40 input_graph = Dataset() 

41 default_graph = input_graph.graph() 

42 

43 file_count = 0 

44 

45 rdf_extensions = { 

46 ".ttl": "turtle", 

47 ".nt": "nt", 

48 ".n3": "n3", 

49 ".xml": "xml", 

50 ".rdf": "xml", 

51 ".jsonld": "json-ld", 

52 ".nq": "nquads", 

53 ".trig": "trig", 

54 } 

55 

56 for filename in os.listdir(input_directory): 

57 file_path = os.path.join(input_directory, filename) 

58 

59 if input_format: 

60 format_name = input_format 

61 else: 

62 _, ext = os.path.splitext(filename.lower()) 

63 if ext not in rdf_extensions: 

64 continue 

65 format_name = rdf_extensions[ext] 

66 

67 default_graph.parse(file_path, format=format_name) 

68 file_count += 1 

69 

70 if file_count == 0: 

71 return 

72 

73 dataset = Dataset() 

74 

75 PROV = Namespace("http://www.w3.org/ns/prov#") 

76 dataset.namespace_manager.bind("prov", PROV) 

77 dataset.namespace_manager.bind("dcterms", DCTERMS) 

78 

79 dataset.default_graph.add((URIRef(""), DCTERMS.license, CC0)) 

80 

81 for prefix, namespace in input_graph.namespace_manager.namespaces(): 

82 dataset.namespace_manager.bind(prefix, namespace) 

83 

84 license_subjects = {s for s, p, _ in default_graph if p == DCTERMS.license} 

85 subjects = set() 

86 for s, _, _ in default_graph: 

87 if isinstance(s, URIRef) and s not in license_subjects: 

88 subjects.add(s) 

89 

90 generation_time = ( 

91 datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat() 

92 ) 

93 

94 responsible_agent = URIRef(agent_orcid) 

95 primary_source_uri = URIRef(primary_source) 

96 

97 for subject in subjects: 

98 prov_graph_uri = URIRef(f"{subject}/prov/") 

99 

100 snapshot_uri = URIRef(f"{subject}/prov/se/1") 

101 

102 prov_graph = dataset.graph(identifier=prov_graph_uri) 

103 

104 prov_graph.add((snapshot_uri, RDF.type, PROV.Entity)) 

105 prov_graph.add((snapshot_uri, PROV.specializationOf, subject)) 

106 

107 prov_graph.add( 

108 ( 

109 snapshot_uri, 

110 PROV.generatedAtTime, 

111 Literal(generation_time, datatype=XSD.dateTime), 

112 ) 

113 ) 

114 

115 prov_graph.add((snapshot_uri, PROV.wasAttributedTo, responsible_agent)) 

116 

117 prov_graph.add((snapshot_uri, PROV.hadPrimarySource, primary_source_uri)) 

118 

119 description = f"Entity <{str(subject)}> was created" 

120 prov_graph.add( 

121 (snapshot_uri, DCTERMS.description, Literal(description, lang="en")) 

122 ) 

123 

124 dataset.serialize(destination=output_file, format=output_format)