Coverage for src/local_deep_research/journal_quality/data_sources/institutions.py: 82%

1"""OpenAlex Institutions data source. 

2 

3Downloads the **bulk snapshot** of ~120K institutions from the public 

4OpenAlex S3 gateway (``openalex.s3.amazonaws.com``) and writes a 

5compact gzipped JSON snapshot used by the institution-scoring tier of 

6the journal filter. Each institution carries its ROR ID, country, 

7h-index, works count, and 2-year mean citedness. 

8 

9Why bulk snapshot instead of the REST API 

10----------------------------------------- 

11 

12The previous implementation cursor-paginated ``/api/institutions`` at 

13200/page with 100 ms polite sleeps — ~550 requests + ~55 s of sleep + 

14actual transfer. Wall-clock was 5–10 minutes per "Download Data" 

15click, which dominated the user-facing latency. The S3 dump is the 

16documented bulk path, CC0, no auth, no rate limits, and finishes 

17much faster. 

18""" 

19 

20from __future__ import annotations 

21 

22import gzip 

23import json 

24import time 

25from pathlib import Path 

26 

27from loguru import logger 

28 

29from ._openalex_common import ( 

30 OPENALEX_S3_BASE, 

31 iter_partitions, 

32 validate_manifest_entries, 

33) 

34from ..scoring import normalize_name 

35from .base import DataSource, atomic_replace 

36 

37_OPENALEX_INSTITUTIONS_MANIFEST = ( 

38 f"{OPENALEX_S3_BASE}/data/jsonl/institutions/manifest.json" 

39) 

40 

41# Safety floor — OpenAlex has ~120K institutions. Refuse to overwrite 

42# good data if the fetch produced far fewer records (likely schema 

43# change, empty partitions, or broken manifest). 

44_MIN_INSTITUTIONS = 50_000 

45 

46 

47class InstitutionSource(DataSource): 

48 key = "institutions" # gitleaks:allow 

49 name = "OpenAlex Institutions" 

50 url = "https://openalex.org" 

51 dataset_url = ( 

52 "https://docs.openalex.org/download-all-data/openalex-snapshot" 

53 ) 

54 license = "CC0 1.0" 

55 license_url = "https://creativecommons.org/publicdomain/zero/1.0/" 

56 description = ( 

57 "~120K research institutions with ROR ID, country, h-index, " 

58 "works count, and citation counts" 

59 ) 

60 filename = "openalex_institutions.json.gz" 

61 count_label = "institutions" 

62 auto_download = False # large; user opts in via dashboard 

63 required = False 

64 approx_size_mb = 10.0 # final compact output, NOT the raw snapshot 

65 

66 def fetch(self, data_dir: Path, progress_cb=None) -> int: 

67 from ...security.safe_requests import ( 

68 safe_get_with_retries as safe_get, 

69 ) 

70 

71 # 1. Fetch the manifest. 

72 logger.info( 

73 f"Fetching OpenAlex institutions manifest: " 

74 f"{_OPENALEX_INSTITUTIONS_MANIFEST}" 

75 ) 

76 # consume_body=True: small JSON but serial bottleneck — a body 

77 # transient here aborts the whole 10-min institutions pull. 

78 manifest_resp = safe_get( 

79 _OPENALEX_INSTITUTIONS_MANIFEST, 

80 timeout=30, 

81 consume_body=True, 

82 require_https=True, 

83 ) 

84 manifest_resp.raise_for_status() 

85 manifest = manifest_resp.json() 

86 

87 # OpenAlex's 2026-06 "standard-format" snapshot renamed the 

88 # manifest's part list from ``entries`` to ``files`` (see 

89 # ``openalex.py``). Entry shape (url + meta) is unchanged. 

90 entries = manifest.get("files", []) 

91 

92 # Validate every part URL before fetching any part — SSRF 

93 # defense in depth. If any entry points outside the OpenAlex 

94 # bucket we refuse the whole fetch rather than partially trust. 

95 validate_manifest_entries(entries, "Institutions") 

96 

97 total_records = sum( 

98 e.get("meta", {}).get("record_count", 0) for e in entries 

99 ) 

100 total_bytes = sum( 

101 e.get("meta", {}).get("content_length", 0) for e in entries 

102 ) 

103 logger.info( 

104 f"OpenAlex institutions snapshot: {len(entries)} parts, " 

105 f"{total_records:,} records, " 

106 f"{total_bytes / 1024 / 1024:.0f} MB compressed" 

107 ) 

108 

109 # 2. Stream-process each part. The ``iter_partitions`` helper 

110 # in ``_openalex_common`` owns the tmp-file lifecycle and 

111 # first-10 malformed-JSON suppression; we focus on the 

112 # compact-format + secondary-index extraction. Compact 

113 # format matches the journal sources snapshot; ROR and 

114 # name indexes keep the runtime lookup path O(1). 

115 institutions: dict = {} 

116 ror_index: dict = {} 

117 name_index: dict = {} 

118 start = time.time() 

119 

120 for idx, total_parts, records in iter_partitions( 

121 entries, 

122 data_dir, 

123 file_prefix="openalex_institutions", 

124 label="Institutions", 

125 safe_get=safe_get, 

126 ): 

127 for inst in records: 

128 inst_id = (inst.get("id") or "").split("/")[-1] 

129 if not inst_id: 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 continue 

131 

132 stats = inst.get("summary_stats") or {} 

133 ror = (inst.get("ror") or "").rstrip("/").split("/")[-1] 

134 compact = { 

135 "n": inst.get("display_name", ""), 

136 "c": inst.get("country_code", ""), 

137 "t": inst.get("type", ""), 

138 "h": stats.get("h_index"), 

139 "if": stats.get("2yr_mean_citedness"), 

140 "w": inst.get("works_count"), 

141 "cb": inst.get("cited_by_count"), 

142 "r": ror or None, 

143 } 

144 institutions[inst_id] = compact 

145 

146 if ror: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 ror_index[ror] = inst_id 

148 

149 primary = normalize_name(inst.get("display_name", "") or "") 

150 if primary: 150 ↛ 152line 150 didn't jump to line 152 because the condition on line 150 was always true

151 name_index[primary] = inst_id 

152 for alt in inst.get("display_name_alternatives") or []: 152 ↛ 153line 152 didn't jump to line 153 because the loop on line 152 never started

153 alt_lower = normalize_name(alt or "") 

154 if alt_lower and alt_lower not in name_index: 

155 name_index[alt_lower] = inst_id 

156 

157 if (idx + 1) % 5 == 0 or idx == total_parts - 1: 157 ↛ 166line 157 didn't jump to line 166 because the condition on line 157 was always true

158 elapsed = time.time() - start 

159 logger.info( 

160 f"OpenAlex institutions: processed " 

161 f"{idx + 1}/{total_parts} parts " 

162 f"({len(institutions):,} records, {elapsed:.0f}s)" 

163 ) 

164 # Per-partition UI ping so the dashboard bar moves 

165 # frequently enough to feel live (vs every 5th partition). 

166 if progress_cb is not None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true

167 try: 

168 progress_cb( 

169 idx + 1, 

170 total_parts, 

171 f"{len(institutions):,} records", 

172 ) 

173 except Exception: 

174 logger.debug( 

175 "institutions progress_cb raised; continuing", 

176 exc_info=True, 

177 ) 

178 

179 if len(institutions) < _MIN_INSTITUTIONS: 

180 raise RuntimeError( 

181 f"OpenAlex institutions: suspiciously few records " 

182 f"({len(institutions):,} < {_MIN_INSTITUTIONS:,}); " 

183 "refusing to overwrite existing data" 

184 ) 

185 

186 payload = { 

187 "i": institutions, # institution ID → compact record 

188 "r": ror_index, # ROR ID → institution ID 

189 "nm": name_index, # lower-cased name → institution ID 

190 } 

191 

192 output = data_dir / self.filename 

193 tmp = data_dir / f"{self.filename}.tmp" 

194 with gzip.open(tmp, "wt", encoding="utf-8") as f: 

195 json.dump(payload, f) 

196 atomic_replace(tmp, output) 

197 

198 elapsed = time.time() - start 

199 logger.info( 

200 f"OpenAlex institutions: saved {len(institutions):,} " 

201 f"institutions in {elapsed:.0f}s" 

202 ) 

203 return len(institutions)