Coverage for src/local_deep_research/web_search_engines/engines/search_engine_pubchem.py: 90%
261 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""PubChem search engine for chemical compound information."""
3from typing import Any, Dict, List, Optional
4from urllib.parse import quote
6from langchain_core.language_models import BaseLLM
8from ...constants import USER_AGENT
9from ...security.safe_requests import safe_get
10from ...security.secure_logging import logger
11from ..rate_limiting import RateLimitError
12from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
15class PubChemSearchEngine(BaseSearchEngine):
16 """
17 PubChem search engine for chemical compound information.
19 Provides access to chemical structures, properties, and bioactivity data.
20 No authentication required.
21 """
23 is_public = True
24 egress_sensitivity = Sensitivity.NON_SENSITIVE
25 egress_exposure = Exposure.EXPOSING
26 is_generic = False
27 is_scientific = True
28 is_code = False
29 is_lexical = True
30 needs_llm_relevance_filter = True
32 def __init__(
33 self,
34 max_results: int = 10,
35 include_synonyms: bool = True,
36 llm: Optional[BaseLLM] = None,
37 max_filtered_results: Optional[int] = None,
38 settings_snapshot: Optional[Dict[str, Any]] = None,
39 **kwargs,
40 ):
41 """
42 Initialize the PubChem search engine.
44 Args:
45 max_results: Maximum number of search results
46 include_synonyms: Whether to include compound synonyms
47 llm: Language model for relevance filtering
48 max_filtered_results: Maximum results after filtering
49 settings_snapshot: Settings snapshot for thread context
50 """
51 super().__init__(
52 llm=llm,
53 max_filtered_results=max_filtered_results,
54 max_results=max_results,
55 settings_snapshot=settings_snapshot,
56 **kwargs,
57 )
59 self.include_synonyms = include_synonyms
60 self.base_url = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"
61 self.autocomplete_url = (
62 "https://pubchem.ncbi.nlm.nih.gov/rest/autocomplete"
63 )
65 # User-Agent header for API requests
66 self.headers = {"User-Agent": USER_AGENT}
68 def _search_compounds(self, query: str) -> List[str]:
69 """Search for compound names matching the query."""
70 try:
71 url = (
72 f"{self.autocomplete_url}/compound/{quote(query, safe='')}/json"
73 )
74 params = {"limit": self.max_results * 2} # Get extra for filtering
76 response = safe_get(
77 url, params=params, headers=self.headers, timeout=30
78 )
79 self._raise_if_rate_limit(response.status_code)
80 response.raise_for_status()
81 data = response.json()
83 terms: list[str] = data.get("dictionary_terms", {}).get(
84 "compound", []
85 )
86 return terms
88 except RateLimitError:
89 raise
90 except Exception as e:
91 safe_msg = self._scrub_error(e)
92 logger.exception(
93 f"PubChem autocomplete search failed ({type(e).__name__}): {safe_msg}"
94 )
95 return []
97 def _get_compound_by_name(self, name: str) -> Optional[Dict[str, Any]]:
98 """Get compound information by name."""
99 try:
100 self.rate_tracker.apply_rate_limit(self.engine_type)
101 # Get CID first
102 url = f"{self.base_url}/compound/name/{quote(name, safe='')}/cids/JSON"
103 response = safe_get(url, headers=self.headers, timeout=30)
105 if response.status_code == 404:
106 return None
107 self._raise_if_rate_limit(response.status_code)
109 response.raise_for_status()
110 data = response.json()
111 cids = data.get("IdentifierList", {}).get("CID", [])
113 if not cids:
114 return None
116 cid = cids[0]
118 # Get compound properties
119 properties = self._get_compound_properties(cid)
121 # Get compound description
122 description = self._get_compound_description(cid)
124 return {
125 "cid": cid,
126 "name": name,
127 "properties": properties,
128 "description": description,
129 }
131 except RateLimitError:
132 raise
133 except Exception as e:
134 safe_msg = self._scrub_error(e)
135 logger.exception(
136 f"Error fetching PubChem compound: {name} ({type(e).__name__}): {safe_msg}"
137 )
138 return None
140 def _get_compound_properties(self, cid: int) -> Dict[str, Any]:
141 """Get properties for a compound by CID."""
142 try:
143 self.rate_tracker.apply_rate_limit(self.engine_type)
144 properties_list = [
145 "MolecularFormula",
146 "MolecularWeight",
147 "IUPACName",
148 "CanonicalSMILES",
149 "IsomericSMILES",
150 "InChI",
151 "InChIKey",
152 "XLogP",
153 "TPSA",
154 "Complexity",
155 "Charge",
156 "HBondDonorCount",
157 "HBondAcceptorCount",
158 "RotatableBondCount",
159 "HeavyAtomCount",
160 ]
162 url = f"{self.base_url}/compound/cid/{cid}/property/{','.join(properties_list)}/JSON"
163 response = safe_get(url, headers=self.headers, timeout=30)
164 self._raise_if_rate_limit(response.status_code)
165 response.raise_for_status()
166 data = response.json()
168 props = data.get("PropertyTable", {}).get("Properties", [])
169 return props[0] if props else {}
171 except RateLimitError:
172 raise
173 except Exception as e:
174 safe_msg = self._scrub_error(e)
175 logger.exception(
176 f"Error fetching PubChem properties for CID {cid} ({type(e).__name__}): {safe_msg}"
177 )
178 return {}
180 def _get_compound_description(self, cid: int) -> str:
181 """Get description for a compound by CID."""
182 try:
183 self.rate_tracker.apply_rate_limit(self.engine_type)
184 url = f"{self.base_url}/compound/cid/{cid}/description/JSON"
185 response = safe_get(url, headers=self.headers, timeout=30)
187 if response.status_code == 404:
188 return ""
189 self._raise_if_rate_limit(response.status_code)
191 response.raise_for_status()
192 data = response.json()
194 descriptions = data.get("InformationList", {}).get(
195 "Information", []
196 )
197 for desc in descriptions:
198 if desc.get("Description"):
199 return desc.get("Description", "") # type: ignore[no-any-return]
201 return ""
203 except RateLimitError:
204 raise
205 except Exception as e:
206 safe_msg = self._scrub_error(e)
207 logger.exception(
208 f"Error fetching PubChem description for CID {cid} ({type(e).__name__}): {safe_msg}"
209 )
210 return ""
212 def _get_compound_synonyms(self, cid: int, limit: int = 10) -> List[str]:
213 """Get synonyms for a compound by CID."""
214 try:
215 self.rate_tracker.apply_rate_limit(self.engine_type)
216 url = f"{self.base_url}/compound/cid/{cid}/synonyms/JSON"
217 response = safe_get(url, headers=self.headers, timeout=30)
219 if response.status_code == 404:
220 return []
221 self._raise_if_rate_limit(response.status_code)
223 response.raise_for_status()
224 data = response.json()
226 info = data.get("InformationList", {}).get("Information", [])
227 if info:
228 synonyms = info[0].get("Synonym", [])
229 return synonyms[:limit] # type: ignore[no-any-return]
230 return []
232 except RateLimitError:
233 raise
234 except Exception as e:
235 safe_msg = self._scrub_error(e)
236 logger.exception(
237 f"Error fetching PubChem synonyms for CID {cid} ({type(e).__name__}): {safe_msg}"
238 )
239 return []
241 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
242 """
243 Get preview information for PubChem compounds.
245 Args:
246 query: The search query (compound name)
248 Returns:
249 List of preview dictionaries
250 """
251 logger.info(f"Getting PubChem previews for query: {query}")
253 # Apply rate limiting
254 self._last_wait_time = self.rate_tracker.apply_rate_limit(
255 self.engine_type
256 )
258 # Search for matching compound names
259 compound_names = self._search_compounds(query)
261 if not compound_names:
262 # Try direct lookup
263 compound = self._get_compound_by_name(query)
264 if compound:
265 compound_names = [query]
266 else:
267 logger.info("No PubChem compounds found")
268 return []
270 logger.info(f"Found {len(compound_names)} potential compounds")
272 previews: list[dict[str, Any]] = []
273 seen_cids = set()
274 for name in compound_names:
275 if len(previews) >= self.max_results:
276 break
278 try:
279 compound = self._get_compound_by_name(name)
280 if not compound:
281 continue
283 cid = compound["cid"]
285 # Deduplicate by CID (autocomplete may return
286 # case variants like "Caffeine" and "caffeine")
287 if cid in seen_cids:
288 continue
289 seen_cids.add(cid)
290 properties = compound.get("properties", {})
291 description = compound.get("description", "")
293 # Build compound URL
294 compound_url = (
295 f"https://pubchem.ncbi.nlm.nih.gov/compound/{cid}"
296 )
298 # Get key properties
299 molecular_formula = properties.get("MolecularFormula", "")
300 molecular_weight = properties.get("MolecularWeight", "")
301 iupac_name = properties.get("IUPACName", "")
302 smiles = (
303 properties.get("CanonicalSMILES", "")
304 or properties.get("SMILES", "")
305 or properties.get("IsomericSMILES", "")
306 or properties.get("ConnectivitySMILES", "")
307 )
309 # Get drug-relevant properties
310 xlogp = properties.get("XLogP")
311 hbond_donors = properties.get("HBondDonorCount")
312 hbond_acceptors = properties.get("HBondAcceptorCount")
314 # Build snippet
315 snippet_parts = []
316 if molecular_formula:
317 snippet_parts.append(f"Formula: {molecular_formula}")
318 if molecular_weight: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 snippet_parts.append(f"MW: {molecular_weight}")
320 if xlogp is not None: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 snippet_parts.append(f"XLogP: {xlogp}")
322 if hbond_donors is not None or hbond_acceptors is not None: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 hbond_info = []
324 if hbond_donors is not None:
325 hbond_info.append(f"H-Donors: {hbond_donors}")
326 if hbond_acceptors is not None:
327 hbond_info.append(f"H-Acceptors: {hbond_acceptors}")
328 snippet_parts.append(", ".join(hbond_info))
329 if iupac_name:
330 snippet_parts.append(f"IUPAC: {iupac_name}")
331 if description:
332 snippet_parts.append(description[:200])
333 snippet = ". ".join(snippet_parts)
335 preview = {
336 "id": str(cid),
337 "cid": cid,
338 "title": name,
339 "link": compound_url,
340 "snippet": snippet,
341 "molecular_formula": molecular_formula,
342 "molecular_weight": molecular_weight,
343 "iupac_name": iupac_name,
344 "smiles": smiles,
345 "inchi_key": properties.get("InChIKey", ""),
346 "description": description,
347 "source": "PubChem",
348 "_raw": {
349 "properties": properties,
350 "description": description,
351 },
352 }
354 previews.append(preview)
356 except RateLimitError:
357 raise
358 except Exception as e:
359 safe_msg = self._scrub_error(e)
360 logger.exception(
361 f"Error processing PubChem compound: {name} ({type(e).__name__}): {safe_msg}"
362 )
363 continue
365 return previews
367 def _get_full_content(
368 self, relevant_items: List[Dict[str, Any]]
369 ) -> List[Dict[str, Any]]:
370 """
371 Get full content for the relevant PubChem compounds.
373 Args:
374 relevant_items: List of relevant preview dictionaries
376 Returns:
377 List of result dictionaries with full content
378 """
379 logger.info(
380 f"Getting full content for {len(relevant_items)} PubChem compounds"
381 )
383 results = []
384 for item in relevant_items:
385 result = item.copy()
387 cid = item.get("cid")
388 if cid and self.include_synonyms: 388 ↛ 390line 388 didn't jump to line 390 because the condition on line 388 was never true
389 # Get synonyms
390 synonyms = self._get_compound_synonyms(cid)
391 result["synonyms"] = synonyms
393 raw = item.get("_raw", {})
394 if raw: 394 ↛ 446line 394 didn't jump to line 446 because the condition on line 394 was always true
395 properties = raw.get("properties", {})
396 description = raw.get("description", "")
398 # Build content summary
399 content_parts = []
400 content_parts.append(
401 f"Compound: {result.get('title', 'Unknown')}"
402 )
403 if cid is not None: 403 ↛ 406line 403 didn't jump to line 406 because the condition on line 403 was always true
404 content_parts.append(f"CID: {cid}")
406 if result.get("molecular_formula"): 406 ↛ 410line 406 didn't jump to line 410 because the condition on line 406 was always true
407 content_parts.append(
408 f"Molecular Formula: {result['molecular_formula']}"
409 )
410 if result.get("molecular_weight"): 410 ↛ 414line 410 didn't jump to line 414 because the condition on line 410 was always true
411 content_parts.append(
412 f"Molecular Weight: {result['molecular_weight']} g/mol"
413 )
414 if result.get("iupac_name"): 414 ↛ 416line 414 didn't jump to line 416 because the condition on line 414 was always true
415 content_parts.append(f"IUPAC Name: {result['iupac_name']}")
416 if result.get("smiles"): 416 ↛ 418line 416 didn't jump to line 418 because the condition on line 416 was always true
417 content_parts.append(f"SMILES: {result['smiles']}")
418 if result.get("inchi_key"): 418 ↛ 422line 418 didn't jump to line 422 because the condition on line 418 was always true
419 content_parts.append(f"InChIKey: {result['inchi_key']}")
421 # Additional properties
422 if properties.get("XLogP") is not None: 422 ↛ 424line 422 didn't jump to line 424 because the condition on line 422 was always true
423 content_parts.append(f"XLogP: {properties['XLogP']}")
424 if properties.get("TPSA") is not None: 424 ↛ 426line 424 didn't jump to line 426 because the condition on line 424 was always true
425 content_parts.append(f"TPSA: {properties['TPSA']} Ų")
426 if properties.get("HBondDonorCount") is not None: 426 ↛ 430line 426 didn't jump to line 430 because the condition on line 426 was always true
427 content_parts.append(
428 f"H-Bond Donors: {properties['HBondDonorCount']}"
429 )
430 if properties.get("HBondAcceptorCount") is not None: 430 ↛ 435line 430 didn't jump to line 435 because the condition on line 430 was always true
431 content_parts.append(
432 f"H-Bond Acceptors: {properties['HBondAcceptorCount']}"
433 )
435 if result.get("synonyms"): 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true
436 content_parts.append(
437 f"\nSynonyms: {', '.join(result['synonyms'][:5])}"
438 )
440 if description: 440 ↛ 443line 440 didn't jump to line 443 because the condition on line 440 was always true
441 content_parts.append(f"\nDescription: {description}")
443 result["content"] = "\n".join(content_parts)
445 # Clean up internal fields
446 if "_raw" in result: 446 ↛ 449line 446 didn't jump to line 449 because the condition on line 446 was always true
447 del result["_raw"]
449 results.append(result)
451 return results
453 def get_compound(self, cid: int) -> Optional[Dict[str, Any]]:
454 """
455 Get a specific compound by CID.
457 Args:
458 cid: The PubChem compound ID
460 Returns:
461 Compound dictionary or None
462 """
463 try:
464 properties = self._get_compound_properties(cid)
465 description = self._get_compound_description(cid)
466 synonyms = self._get_compound_synonyms(cid)
468 return {
469 "cid": cid,
470 "properties": properties,
471 "description": description,
472 "synonyms": synonyms,
473 }
474 except RateLimitError:
475 raise
476 except Exception as e:
477 safe_msg = self._scrub_error(e)
478 logger.exception(
479 f"Error fetching PubChem compound {cid} ({type(e).__name__}): {safe_msg}"
480 )
481 return None
483 def search_by_formula(self, formula: str) -> List[Dict[str, Any]]:
484 """
485 Search compounds by molecular formula.
487 Args:
488 formula: Molecular formula (e.g., "C6H12O6")
490 Returns:
491 List of matching compounds
492 """
493 try:
494 url = f"{self.base_url}/compound/fastformula/{quote(formula, safe='')}/cids/JSON"
495 response = safe_get(url, headers=self.headers, timeout=30)
497 if response.status_code == 404:
498 return []
499 self._raise_if_rate_limit(response.status_code)
501 response.raise_for_status()
502 data = response.json()
503 cids = data.get("IdentifierList", {}).get("CID", [])
505 results = []
506 for cid in cids[: self.max_results]:
507 compound = self.get_compound(cid)
508 if compound:
509 results.append(compound)
511 return results
513 except RateLimitError:
514 raise
515 except Exception as e:
516 safe_msg = self._scrub_error(e)
517 logger.exception(
518 f"Error searching by formula: {formula} ({type(e).__name__}): {safe_msg}"
519 )
520 return []