Coverage for oc_meta / run / infodir / gen.py: 97%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import argparse 

10import os 

11import shutil 

12import tempfile 

13from datetime import datetime, timezone 

14 

15from rich_argparse import RichHelpFormatter 

16 

17from oc_meta.lib.console import console 

18from oc_meta.run.infodir._common import ( 

19 CounterKey, 

20 SourceDataError, 

21 SparseCounterStore, 

22 scan_data, 

23 scan_provenance, 

24 write_json, 

25) 

26 

27DEFAULT_WORKERS = 4 

28DEFAULT_MAX_EXAMPLES = 100 

29 

30 

31def generate_info_dir( 

32 root_path: str, 

33 info_dir: str, 

34 workers: int = DEFAULT_WORKERS, 

35 max_examples: int = DEFAULT_MAX_EXAMPLES, 

36 report_path: str | None = None, 

37) -> dict[str, object]: 

38 root_path = os.path.abspath(root_path) 

39 info_dir = os.path.abspath(info_dir) 

40 if os.path.exists(info_dir): 

41 raise FileExistsError(f"Info directory already exists: {info_dir}") 

42 if max_examples < 0: 

43 raise ValueError("max_examples must be non-negative") 

44 

45 destination_parent = os.path.dirname(info_dir) 

46 os.makedirs(destination_parent, exist_ok=True) 

47 staging_dir = tempfile.mkdtemp( 

48 prefix=f".{os.path.basename(info_dir)}.", suffix=".tmp", dir=destination_parent 

49 ) 

50 scratch_dir = os.path.join(staging_dir, ".scratch") 

51 os.makedirs(scratch_dir) 

52 store = SparseCounterStore(scratch_dir) 

53 published = False 

54 

55 try: 

56 provenance_scan = scan_provenance(root_path, store, workers) 

57 data_scan = scan_data( 

58 root_path, 

59 store, 

60 workers, 

61 max_examples, 

62 ) 

63 if provenance_scan.zip_files == 0 and data_scan.zip_files == 0: 

64 raise SourceDataError(f"No RDF ZIP files found in {root_path}") 

65 

66 all_keys = store.keys() | set(data_scan.maxima) 

67 entity_maxima: dict[CounterKey, int] = {} 

68 for key in all_keys: 

69 entity_maxima[key] = max( 

70 data_scan.maxima[key] if key in data_scan.maxima else 0, 

71 store.maximum(key), 

72 ) 

73 

74 for (prefix, short_name), maximum in sorted(entity_maxima.items()): 

75 prefix_dir = os.path.join(staging_dir, prefix) 

76 os.makedirs(prefix_dir, exist_ok=True) 

77 with open( 

78 os.path.join(prefix_dir, f"info_file_{short_name}.txt"), 

79 "w", 

80 encoding="utf-8", 

81 ) as output_file: 

82 output_file.write(f"{maximum}\n") 

83 

84 for key in sorted(store.keys()): 

85 prefix, short_name = key 

86 store.render( 

87 key, 

88 os.path.join(staging_dir, prefix, f"prov_file_{short_name}.txt"), 

89 ) 

90 

91 store.close() 

92 shutil.rmtree(scratch_dir) 

93 status = ( 

94 "generated_with_warnings" if data_scan.missing_provenance else "generated" 

95 ) 

96 report: dict[str, object] = { 

97 "timestamp": datetime.now(timezone.utc).isoformat(), 

98 "status": status, 

99 "root_path": root_path, 

100 "info_dir": info_dir, 

101 "source": { 

102 "data_zip_files": data_scan.zip_files, 

103 "provenance_zip_files": provenance_scan.zip_files, 

104 "data_entities": data_scan.entities, 

105 "provenance_entities": provenance_scan.entities, 

106 }, 

107 "live_entities_without_provenance": { 

108 "total": data_scan.missing_provenance, 

109 "examples": data_scan.missing_examples, 

110 "truncated": data_scan.missing_provenance 

111 > len(data_scan.missing_examples), 

112 }, 

113 } 

114 

115 os.replace(staging_dir, info_dir) 

116 published = True 

117 finally: 

118 store.close() 

119 if not published and os.path.isdir(staging_dir): 

120 shutil.rmtree(staging_dir) 

121 

122 if report_path is not None: 

123 write_json(report_path, report) 

124 

125 console.print(f"Generated info directory: {info_dir}") 

126 console.print(f"Live entities without provenance: {data_scan.missing_provenance}") 

127 if report_path is not None: 

128 console.print(f"Report saved to {os.path.abspath(report_path)}") 

129 return report 

130 

131 

132def main() -> int: # pragma: no cover 

133 parser = argparse.ArgumentParser( 

134 description="Scan RDF directories and populate filesystem counter files.", 

135 formatter_class=RichHelpFormatter, 

136 ) 

137 parser.add_argument("directory", type=str, help="Path to the RDF directory to scan") 

138 parser.add_argument("info_dir", type=str, help="New counter directory to create") 

139 parser.add_argument( 

140 "-o", 

141 "--output", 

142 help="Generation report path (default: <info_dir>.generation-report.json)", 

143 ) 

144 parser.add_argument( 

145 "--workers", 

146 type=int, 

147 default=DEFAULT_WORKERS, 

148 help=f"Worker processes (default: {DEFAULT_WORKERS})", 

149 ) 

150 parser.add_argument( 

151 "--max-examples", 

152 type=int, 

153 default=DEFAULT_MAX_EXAMPLES, 

154 help=f"Examples retained per warning category (default: {DEFAULT_MAX_EXAMPLES})", 

155 ) 

156 args = parser.parse_args() 

157 report_path = ( 

158 args.output 

159 if args.output is not None 

160 else f"{os.path.abspath(args.info_dir)}.generation-report.json" 

161 ) 

162 generate_info_dir( 

163 args.directory, 

164 args.info_dir, 

165 workers=args.workers, 

166 max_examples=args.max_examples, 

167 report_path=report_path, 

168 ) 

169 return 0 

170 

171 

172if __name__ == "__main__": # pragma: no cover 

173 raise SystemExit(main())