Coverage for oc_meta / run / count / meta_entities.py: 71%

1#!/usr/bin/python 

2 

3# Copyright 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

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

5# 

6# SPDX-License-Identifier: ISC 

7 

8from __future__ import annotations 

9 

10import argparse 

11import os 

12import sys 

13import multiprocessing 

14from concurrent.futures import ProcessPoolExecutor, as_completed 

15from typing import Dict, Set 

16 

17from rich_argparse import RichHelpFormatter 

18from oc_meta.lib.sparql import execute_sparql 

19 

20from oc_meta.lib.console import create_progress 

21from oc_meta.lib.file_manager import get_csv_data 

22from oc_meta.lib.master_of_regex import split_name_and_ids 

23 

24 

25def _count_venues_in_file(filepath: str) -> Set[str]: 

26 csv_data = get_csv_data(filepath) 

27 venues = set() 

28 for row in csv_data: 

29 if not row["venue"]: 

30 continue 

31 venue_name, venue_ids_str = split_name_and_ids(row["venue"]) 

32 if not venue_ids_str: 

33 continue 

34 venue_ids = set(venue_ids_str.split()) 

35 venue_metaid = next( 

36 identifier 

37 for identifier in venue_ids 

38 if identifier.split(":", maxsplit=1)[0] == "omid" 

39 ) 

40 if not venue_ids.difference({venue_metaid}): 

41 venues.add(venue_name.lower()) 

42 else: 

43 venues.add(venue_metaid) 

44 return venues 

45 

46 

47class OCMetaStatistics: 

48 def __init__( 

49 self, 

50 sparql_endpoint: str, 

51 csv_dump_path: str | None = None, 

52 max_retries: int = 3, 

53 retry_delay: int = 5, 

54 ): 

55 self.sparql_endpoint = sparql_endpoint 

56 self.csv_dump_path = csv_dump_path 

57 self.max_retries = max_retries 

58 self.retry_delay = retry_delay 

59 

60 def _execute_sparql_query(self, query: str) -> Dict: 

61 try: 

62 return execute_sparql( 

63 self.sparql_endpoint, 

64 query, 

65 max_retries=self.max_retries, 

66 backoff_factor=self.retry_delay, 

67 ) 

68 except Exception as e: 

69 print(f"Query failed after {self.max_retries} retries.", file=sys.stderr) 

70 raise Exception("SPARQL query failed after multiple retries.") from e 

71 

72 def __enter__(self): 

73 return self 

74 

75 def __exit__(self, exc_type, exc_val, exc_tb): 

76 return False 

77 

78 def count_expressions(self) -> int: 

79 query = """ 

80 PREFIX fabio: <http://purl.org/spar/fabio/> 

81 

82 SELECT (COUNT(DISTINCT ?expression) AS ?count) 

83 WHERE { 

84 ?expression a fabio:Expression . 

85 } 

86 """ 

87 results = self._execute_sparql_query(query) 

88 return int(results["results"]["bindings"][0]["count"]["value"]) 

89 

90 def count_role_entities(self) -> Dict[str, int]: 

91 query = """ 

92 PREFIX pro: <http://purl.org/spar/pro/> 

93 

94 SELECT ?role (COUNT(DISTINCT ?roleInTime) AS ?count) 

95 WHERE { 

96 ?roleInTime pro:withRole ?role . 

97 FILTER(?role IN (pro:author, pro:publisher, pro:editor)) 

98 } 

99 GROUP BY ?role 

100 """ 

101 results = self._execute_sparql_query(query) 

102 

103 role_counts = {"pro:author": 0, "pro:publisher": 0, "pro:editor": 0} 

104 

105 for binding in results["results"]["bindings"]: 

106 role_uri = binding["role"]["value"] 

107 count = int(binding["count"]["value"]) 

108 

109 if role_uri == "http://purl.org/spar/pro/author": 

110 role_counts["pro:author"] = count 

111 elif role_uri == "http://purl.org/spar/pro/publisher": 

112 role_counts["pro:publisher"] = count 

113 elif role_uri == "http://purl.org/spar/pro/editor": 

114 role_counts["pro:editor"] = count 

115 

116 return role_counts 

117 

118 def count_venues_from_csv(self) -> int: 

119 if not self.csv_dump_path: 

120 raise ValueError("CSV dump path is required to count venues") 

121 

122 filenames = sorted(os.listdir(self.csv_dump_path)) 

123 filepaths = [ 

124 os.path.join(self.csv_dump_path, f) for f in filenames if f.endswith(".csv") 

125 ] 

126 

127 all_venues: Set[str] = set() 

128 

129 with create_progress() as progress: 

130 task = progress.add_task( 

131 "Counting venues from CSV files...", total=len(filepaths) 

132 ) 

133 

134 # Use forkserver to avoid deadlocks when forking in a multi-threaded environment 

135 with ProcessPoolExecutor( 

136 mp_context=multiprocessing.get_context("forkserver") 

137 ) as executor: 

138 futures = { 

139 executor.submit(_count_venues_in_file, fp): fp for fp in filepaths 

140 } 

141 for future in as_completed(futures): 

142 venues = future.result() 

143 all_venues.update(venues) 

144 progress.update(task, advance=1) 

145 

146 return len(all_venues) 

147 

148 def run_selected_analyses( 

149 self, analyze_br: bool, analyze_ar: bool, analyze_venues: bool 

150 ) -> Dict: 

151 print("Starting dataset statistics...") 

152 print(f"Connected to endpoint: {self.sparql_endpoint}") 

153 if self.csv_dump_path: 

154 print(f"CSV dump path: {self.csv_dump_path}") 

155 print() 

156 

157 results = {} 

158 

159 if analyze_br: 

160 print("1. Counting fabio:Expression entities...") 

161 try: 

162 expressions_count = self.count_expressions() 

163 results["fabio_expressions"] = expressions_count 

164 print(f" Found {expressions_count:,} fabio:Expression entities") 

165 except Exception as e: 

166 print(f" Error: {e}") 

167 results["fabio_expressions"] = None 

168 print() 

169 

170 if analyze_ar: 

171 print("2. Counting pro:author, pro:publisher and pro:editor roles...") 

172 try: 

173 role_counts = self.count_role_entities() 

174 results["roles"] = role_counts 

175 print(f" Found {role_counts['pro:author']:,} pro:author roles") 

176 print(f" Found {role_counts['pro:publisher']:,} pro:publisher roles") 

177 print(f" Found {role_counts['pro:editor']:,} pro:editor roles") 

178 except Exception as e: 

179 print(f" Error: {e}") 

180 results["roles"] = None 

181 print() 

182 

183 if analyze_venues: 

184 print("3. Counting venues from CSV dump...") 

185 if not self.csv_dump_path: 

186 print(" Error: CSV dump path is required for venue counting") 

187 results["venues"] = None 

188 else: 

189 try: 

190 venues_count = self.count_venues_from_csv() 

191 results["venues"] = venues_count 

192 print(f" Found {venues_count:,} distinct venues") 

193 except Exception as e: 

194 print(f" Error: {e}") 

195 results["venues"] = None 

196 print() 

197 

198 print("Statistics completed!") 

199 return results 

200 

201 def run_all_analyses(self) -> Dict: 

202 return self.run_selected_analyses( 

203 analyze_br=True, analyze_ar=True, analyze_venues=True 

204 ) 

205 

206 

207def main(): 

208 parser = argparse.ArgumentParser( 

209 description="Compute OpenCitations Meta dataset statistics", 

210 formatter_class=RichHelpFormatter, 

211 epilog=""" 

212Examples: 

213 # Run all statistics 

214 python -m oc_meta.run.count.meta_entities http://localhost:8890/sparql --csv /path/to/csv/dump 

215 

216 # Count only bibliographic resources 

217 python -m oc_meta.run.count.meta_entities http://localhost:8890/sparql --br 

218 

219 # Count only roles 

220 python -m oc_meta.run.count.meta_entities http://localhost:8890/sparql --ar 

221 

222 # Count only venues (requires CSV dump) 

223 python -m oc_meta.run.count.meta_entities http://localhost:8890/sparql --venues --csv /path/to/csv/dump 

224 

225Statistics computed: 

226 --br: Count fabio:Expression entities (via SPARQL) 

227 --ar: Count pro:author, pro:publisher and pro:editor roles (via SPARQL) 

228 --venues: Count distinct venues with disambiguation (via CSV dump) 

229 

230If no specific options are provided, all statistics will be computed. 

231 """, 

232 ) 

233 

234 parser.add_argument("sparql_endpoint", help="SPARQL endpoint URL") 

235 

236 parser.add_argument( 

237 "--csv", 

238 dest="csv_dump_path", 

239 help="Path to CSV dump directory (required for venue counting)", 

240 ) 

241 

242 parser.add_argument( 

243 "--br", 

244 action="store_true", 

245 help="Count bibliographic resources (fabio:Expression entities)", 

246 ) 

247 

248 parser.add_argument( 

249 "--ar", 

250 action="store_true", 

251 help="Count roles (pro:author, pro:publisher, pro:editor)", 

252 ) 

253 

254 parser.add_argument( 

255 "--venues", action="store_true", help="Count distinct venues (requires --csv)" 

256 ) 

257 

258 args = parser.parse_args() 

259 

260 analyze_br = args.br or not (args.br or args.ar or args.venues) 

261 analyze_ar = args.ar or not (args.br or args.ar or args.venues) 

262 analyze_venues = args.venues or not (args.br or args.ar or args.venues) 

263 

264 if analyze_venues and not args.csv_dump_path: 

265 print("Error: --csv is required for venue counting", file=sys.stderr) 

266 sys.exit(1) 

267 

268 try: 

269 with OCMetaStatistics(args.sparql_endpoint, args.csv_dump_path) as stats: 

270 results = stats.run_selected_analyses( 

271 analyze_br, analyze_ar, analyze_venues 

272 ) 

273 

274 print("\n" + "=" * 50) 

275 print("SUMMARY") 

276 print("=" * 50) 

277 

278 if results.get("fabio_expressions") is not None: 

279 print(f"fabio:Expression entities: {results['fabio_expressions']:,}") 

280 

281 if results.get("roles"): 

282 print(f"pro:author roles: {results['roles']['pro:author']:,}") 

283 print(f"pro:publisher roles: {results['roles']['pro:publisher']:,}") 

284 print(f"pro:editor roles: {results['roles']['pro:editor']:,}") 

285 

286 if results.get("venues") is not None: 

287 print(f"Distinct venues: {results['venues']:,}") 

288 

289 return results 

290 

291 except Exception as e: 

292 print(f"Statistics failed: {e}", file=sys.stderr) 

293 sys.exit(1) 

294 

295 

296if __name__ == "__main__": 

297 main()