"""The literature, as a source of its own. A web result is a page whose citation has to be reconstructed from its title; a PubMed record is structured — title, journal, year, authors, PMID — so a reference can be exact, and a reference list can carry a number somebody can look up. NCBI's E-utilities need no key. A key raises the rate limit from three requests a second to ten, which matters for indexing and not for one person drafting one article, so the key is optional and its absence is not a misconfiguration. Either way this reaches NCBI rather than a commercial third party. Ported from the same thing in ped-ai, including the two lessons that cost somebody an afternoon there: the query ladder, and the spacing between retries. """ import logging import re import time import urllib.parse import urllib.request from app.services import site_settings logger = logging.getLogger(__name__) BASE = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" MAX_RESULTS = 6 TIMEOUT = 15 #: NCBI asks callers to identify themselves so they can get in touch before #: blocking you. Free, and the polite thing to do. TOOL_NAME = "pedshub" #: PubMed ANDs every term after automatic mapping, so one unrecognised word #: takes the whole query to zero: "febrile seizures" finds six papers, #: "febrile seizures in under-fives" finds none. A topic written by a person is #: full of such words. STOPWORDS = set(( "a an and are as at be but by for from how in into is it its of on or " "that the their there these this to under over with what when where which " "who why show shows showed evidence review overview update current recent " "latest new approach approaches management use using guidance guidelines" ).split()) def settings() -> dict: return { "enabled": site_settings.get_flag("pubmed_enabled", False), "api_key": (site_settings.get_value("pubmed_api_key") or "").strip(), "email": (site_settings.get_value("pubmed_contact_email") or "").strip(), } def _clip(text, limit: int) -> str: return re.sub(r"\s+", " ", str(text or "")).strip()[:limit] def _year(pubdate) -> str: found = re.search(r"\d{4}", str(pubdate or "")) return found.group(0) if found else "" def _url(path: str, params: dict, config: dict) -> str: query = {k: v for k, v in params.items() if v not in (None, "")} query["tool"] = TOOL_NAME if config.get("email"): query["email"] = config["email"] if config.get("api_key"): query["api_key"] = config["api_key"] return f"{BASE}/{path}?{urllib.parse.urlencode(query)}" def _get(url: str) -> str: request = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(request, timeout=TIMEOUT) as response: return response.read().decode("utf-8", "replace") def _content_words(text: str) -> list[str]: # Anything after a colon or a dash is almost always a subtitle, not a subject. head = re.split(r"[:—–]|\s-\s", str(text))[0] cleaned = re.sub(r"[^a-z0-9\s-]", " ", head.lower()) return [word for word in cleaned.split() if len(word) > 2 and word not in STOPWORDS] def _candidates(text: str) -> list[str]: """Progressively shorter versions of the query, longest first. So a topic that finds nothing is retried against something PubMed can actually match, and the most specific query that works is the one used. """ words = _content_words(text) tried = [text.strip()] for count in range(min(len(words), 4), 1, -1): candidate = " ".join(words[:count]) if candidate and candidate not in tried: tried.append(candidate) if len(words) == 1 and words[0] not in tried: tried.append(words[0]) return tried def _abstracts(pmids: list[str], config: dict) -> dict[str, str]: """Fetched separately, because esummary does not carry them. An abstract is what makes a record useful to write from rather than merely cite. Parsed leniently: a missing one is normal — editorials, letters — and must not lose the record. """ if not pmids: return {} try: raw = _get(_url("efetch.fcgi", { "db": "pubmed", "id": ",".join(pmids), "retmode": "xml", "rettype": "abstract", }, config)) except Exception: return {} out: dict[str, str] = {} for chunk in raw.split("")[1:]: found = re.search(r"]*>(\d+)", chunk) if not found: continue parts = [re.sub(r"<[^>]+>", " ", part) for part in re.findall(r"]*>([\s\S]*?)", chunk)] if parts: out[found.group(1)] = _clip(" ".join(parts), 1500) return out def search(query: str, limit: int = MAX_RESULTS) -> dict: """Search PubMed for a topic. Never raises, for the same reason corpus retrieval does not: a lookup failing must not fail the article somebody is writing. """ text = str(query or "").strip()[:400] if not text: return {"results": [], "reason": "empty query"} config = settings() if not config["enabled"]: return {"results": [], "reason": "PubMed is turned off for this site"} try: import json def ids_for(term: str) -> list[str]: found = json.loads(_get(_url("esearch.fcgi", { "db": "pubmed", "term": term, "retmode": "json", "retmax": str(limit), "sort": "relevance", }, config))) return (found.get("esearchresult") or {}).get("idlist", [])[:limit] # Three requests a second without a key, ten with one. The ladder can # make three esearch calls before the esummary and efetch that follow, # which is enough to turn a working search into a 429 — measured, on # this exact path. So retries are spaced; the first attempt, which is # the usual case, waits for nothing. tries = _candidates(text) gap = 0.12 if config["api_key"] else 0.38 used, pmids = tries[0], [] for index, term in enumerate(tries): if index: time.sleep(gap) used = term pmids = ids_for(term) if pmids: break if not pmids: return {"results": [], "reason": "no results", "query": used} summary = json.loads(_get(_url("esummary.fcgi", { "db": "pubmed", "id": ",".join(pmids), "retmode": "json", }, config))).get("result", {}) abstracts = _abstracts(pmids, config) results = [] for pmid in pmids: record = summary.get(pmid) or {} names = [a.get("name") for a in record.get("authors", []) if a.get("name")] results.append({ "pmid": pmid, "title": _clip(record.get("title"), 300) or "Untitled", "journal": _clip(record.get("source"), 150), "year": _year(record.get("pubdate")), # Three and "et al" is how a citation reads; the full list is # noise here. "authors": ", ".join(names[:3]) + (", et al" if len(names) > 3 else ""), "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/", "abstract": abstracts.get(pmid, ""), }) # Saying which query worked matters: the screen reports what was # searched, and reporting the phrase somebody typed when a narrowed one # found the results would be a lie. return {"results": results, "reason": None, "query": used} except Exception as exc: logger.warning("PubMed search failed: %s", exc) return {"results": [], "reason": "PubMed could not be reached"} def for_prompt(results: list[dict]) -> str: """Formatted so a model can cite it exactly, PMID and all.""" blocks = [] for record in results: blocks.append("\n".join(filter(None, [ record["title"], ". ".join(filter(None, [record["authors"], record["journal"], record["year"]])), f"PMID: {record['pmid']} — {record['url']}", record["abstract"] or "(no abstract available)", ]))) return "\n\n---\n\n".join(blocks) def as_references(results: list[dict]) -> list[dict]: """In the shape an article's reference list already uses. Fields, not a sentence. Every other reference on an article is {title, author, pages, ...} and the reader draws those fields — so a flat line came out as a blank row under a References heading. A paper fills the journal, year and PMID instead of the pages, and the PMID is what makes it checkable. """ return [{ "title": record["title"].rstrip("."), "author": record["authors"] or None, "pages": [], "journal": record["journal"] or None, "year": record["year"] or None, "pmid": record["pmid"], } for record in results]