""" title: Smart PubMed Research Assistant author: Research Assistant version: 6.1.0 date: 2025-01-01 license: MIT description: Intelligent PubMed research assistant. Uses PubMed's own Automatic Term Mapping. Fetches abstracts for AI synthesis. Vancouver-style numbered references with proper AI citation instructions. RIS export for Zotero. Works with or without NCBI API key. """ import requests import re import xml.etree.ElementTree as ET from typing import List, Dict, Optional, Tuple from pydantic import BaseModel, Field class Tools: class Valves(BaseModel): ncbi_api_key: str = Field( default="", description="Optional: NCBI API key for faster searches (get free at https://www.ncbi.nlm.nih.gov/account/settings/). Leave empty to work without it." ) def __init__(self): self.base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" self.valves = self.Valves() self._cache = {} self._last_results = [] self._last_query = "" def _api_params(self, params: dict) -> dict: """Add API key only if configured""" if self.valves.ncbi_api_key and self.valves.ncbi_api_key.strip(): params["api_key"] = self.valves.ncbi_api_key.strip() return params # ================================================================ # MAIN SEARCH # ================================================================ def search_pubmed( self, query: str = Field( ..., description="Your research question in plain English. Examples: 'current guidelines for management of gastric reflux in children', 'low back pain treatment', 'ECMO outcomes in neonates'", ), max_results: int = Field( 10, description="How many articles (1-200). More = slower but more comprehensive.", ), include_abstracts: bool = Field( True, description="Include abstracts for AI synthesis. False = faster metadata-only search.", ), ) -> str: """ Intelligent PubMed search with abstract retrieval. Returns Vancouver-style numbered references. AI can read abstracts and cite using [1], [2], etc. """ try: max_results = self._safe_int(max_results, 10, 1, 200) query = str(query).strip() if not query: return "Please ask me a research question." if isinstance(include_abstracts, str): include_abstracts = include_abstracts.lower() not in ("false", "no", "0") elif not isinstance(include_abstracts, bool): include_abstracts = True # PHASE 1: Let PubMed understand the query analysis = self._analyze_via_pubmed(query) # PHASE 2: Detect query type query_type = self._detect_query_type(query) # PHASE 3: Search iteratively all_articles, search_log = self._iterative_search( query, analysis, query_type, max_results ) # PHASE 4: Fetch abstracts if include_abstracts and all_articles: pmids = [a["pmid"] for a in all_articles] abstracts = self._fetch_abstracts(pmids) for article in all_articles: article["abstract"] = abstracts.get(article["pmid"], "") # PHASE 5: Score and rank scored = self._score_relevance(all_articles, query, query_type) top = scored[:max_results] # PHASE 6: Assign reference numbers for i, article in enumerate(top): article["ref_number"] = i + 1 # PHASE 7: Store self._last_results = top self._last_query = query # PHASE 8: Format if not top: return self._format_no_results(query, analysis, search_log) return self._format_results( query, analysis, query_type, search_log, top, len(all_articles), include_abstracts ) except Exception as e: return self._error_msg(str(e)) # ================================================================ # GET RESULTS IN DIFFERENT FORMATS # ================================================================ def get_results( self, format: str = Field( "list", description="'list' (Vancouver references), 'ris' (Zotero export), 'summary' (AI synthesis), 'abstracts' (all abstracts), 'detailed' (full metadata)", ), ) -> str: """ Get last search results in different formats. References use Vancouver numbered style throughout. """ try: fmt = str(format).strip().lower() if format else "list" if not self._last_results: return "No stored results. Run `search_pubmed` first." if fmt == "ris": return self._export_ris() elif fmt == "summary": return self._synthesize() elif fmt == "abstracts": return self._format_abstracts_only() elif fmt == "detailed": return self._format_detailed() else: return self._format_vancouver_list() except Exception as e: return self._error_msg(str(e)) # ================================================================ # PICO SEARCH # ================================================================ def pico_search( self, population: str = Field(..., description="Who? e.g. 'children under 5 in Africa'"), intervention: str = Field("", description="What? e.g. 'proton pump inhibitors'"), comparison: str = Field("", description="Versus? e.g. 'lifestyle modification'"), outcome: str = Field("", description="Result? e.g. 'symptom resolution'"), max_results: int = Field(15, description="How many articles (1-200)"), ) -> str: """PICO framework search with abstracts and Vancouver references.""" try: max_results = self._safe_int(max_results, 15, 1, 200) pico = {} for label, val in [("Population", population), ("Intervention", intervention), ("Comparison", comparison), ("Outcome", outcome)]: val = str(val).strip() if val else "" if val: pico[label] = val if not pico: return "Please provide at least a Population." pico_analysis = {} for comp, text in pico.items(): pico_analysis[comp] = self._analyze_via_pubmed(text) all_articles, search_log = self._pico_iterative_search( pico, pico_analysis, max_results ) # Fetch abstracts if all_articles: pmids = [a["pmid"] for a in all_articles] abstracts = self._fetch_abstracts(pmids) for a in all_articles: a["abstract"] = abstracts.get(a["pmid"], "") combined = " ".join(pico.values()) scored = self._score_relevance( all_articles, combined, self._detect_query_type(combined) ) top = scored[:max_results] # Assign reference numbers for i, a in enumerate(top): a["ref_number"] = i + 1 self._last_results = top self._last_query = "PICO: " + "; ".join( k + "=" + v for k, v in pico.items() ) # Format md = "# ๐Ÿ”ฌ PICO Search Results\n\n" md += "## Framework\n\n" md += "| Component | Input | PubMed Mapped To |\n" md += "|-----------|-------|------------------|\n" for comp, text in pico.items(): a = pico_analysis[comp] mapped = ", ".join(a.get("mesh_found", [])[:3]) or text md += "| **" + comp + "** | " + text + " | " + mapped + " |\n" md += "\n" for s in search_log: icon = "โœ…" if s["found"] > 0 else "โญ•" md += icon + " **" + s["name"] + "** โ†’ " + str(s["found"]) + " \n" md += "\n" if top: with_abs = sum(1 for a in top if a.get("abstract")) md += "## Results (" + str(len(top)) + " articles" if with_abs: md += ", " + str(with_abs) + " with abstracts" md += ")\n\n" md += self._format_article_list(top, show_abstracts=True) md += "\n## References (Vancouver Style)\n\n" md += self._build_vancouver_list(top) # Citation instructions for PICO too if with_abs: md += self._build_citation_instructions(top, combined) else: md += "**No results.** Try broader terms.\n" md += self._format_next_steps() return md except Exception as e: return self._error_msg(str(e)) # ================================================================ # MESH FINDER # ================================================================ def find_mesh( self, topic: str = Field(..., description="Any medical topic"), ) -> str: """Find MeSH terms via PubMed's own term mapping.""" try: analysis = self._analyze_via_pubmed(str(topic).strip()) md = "# ๐Ÿท๏ธ MeSH: " + topic + "\n\n" if analysis["mesh_found"]: md += "| MeSH Term | Syntax |\n|---|---|\n" for t in analysis["mesh_found"]: md += "| " + t + " | `\"" + t + "\"[MeSH]` |\n" md += "\n" if analysis["query_translation"]: md += "**PubMed translation:**\n```\n" + analysis["query_translation"] + "\n```\n" return md except Exception as e: return self._error_msg(str(e)) # ================================================================ # CORE: PUBMED QUERY ANALYSIS # ================================================================ def _analyze_via_pubmed(self, query: str) -> Dict: """Let PubMed itself understand the query โ€” no word splitting""" cache_key = query.lower().strip() if cache_key in self._cache: return self._cache[cache_key] result = { "original": query, "query_translation": "", "mesh_found": [], "result_count": 0, } try: resp = requests.get( self.base_url + "/esearch.fcgi", params=self._api_params({ "db": "pubmed", "term": query, "retmode": "json", "retmax": "0", }), timeout=15, ) if resp.status_code == 200: es = resp.json().get("esearchresult", {}) result["result_count"] = int(es.get("count", 0)) result["query_translation"] = es.get("querytranslation", "") if result["query_translation"]: mesh = re.findall( r'"([^"]+)"\[MeSH Terms\]', result["query_translation"] ) result["mesh_found"] = list(dict.fromkeys(mesh)) # Also try full phrase as MeSH phrase_query = '"' + query + '"[MeSH Terms]' resp2 = requests.get( self.base_url + "/esearch.fcgi", params=self._api_params({ "db": "pubmed", "term": phrase_query, "retmode": "json", "retmax": "0", }), timeout=10, ) if resp2.status_code == 200: es2 = resp2.json().get("esearchresult", {}) trans2 = es2.get("querytranslation", "") if int(es2.get("count", 0)) > 0 and trans2: for t in re.findall(r'"([^"]+)"\[MeSH Terms\]', trans2): if t not in result["mesh_found"]: result["mesh_found"].append(t) except Exception: pass self._cache[cache_key] = result return result # ================================================================ # QUERY TYPE DETECTION # ================================================================ def _detect_query_type(self, query: str) -> str: q = query.lower() type_map = { "guidelines": [ "guideline", "guidelines", "protocol", "recommendation", "consensus", "management of" ], "systematic_review": ["systematic review", "meta-analysis"], "outcomes": ["outcome", "outcomes", "effectiveness", "efficacy"], "epidemiology": ["prevalence", "incidence", "epidemiology"], "diagnosis": ["diagnosis", "diagnostic", "screening"], "treatment": ["treatment", "therapy", "drug", "medication"], "risk_factors": ["risk factor", "cause", "etiology"], } for qtype, words in type_map.items(): if any(w in q for w in words): return qtype return "general" def _get_type_filter(self, qt: str) -> str: filters = { "guidelines": " AND (\"Practice Guideline\"[PT] OR \"Guideline\"[PT] OR guideline[ti])", "systematic_review": " AND (\"Systematic Review\"[PT] OR \"Meta-Analysis\"[PT])", "outcomes": " AND (\"Clinical Trial\"[PT] OR \"Comparative Study\"[PT])", "diagnosis": " AND (diagnosis[ti] OR diagnostic[ti])", "epidemiology": " AND (prevalence[ti] OR incidence[ti] OR epidemiology[sh])", "treatment": " AND (\"Clinical Trial\"[PT] OR \"Randomized Controlled Trial\"[PT])", } return filters.get(qt, "") # ================================================================ # SEARCH STRATEGIES # ================================================================ def _build_strategies(self, query, analysis, query_type): strategies = [] mesh = analysis.get("mesh_found", []) count = analysis.get("result_count", 0) tf = self._get_type_filter(query_type) # Strategy 1: PubMed auto-map if count > 0: strategies.append(("PubMed Auto-Mapping", query)) # Strategy 2: MeSH + type filter if mesh and tf: mesh_terms = ['"' + t + '"[MeSH]' for t in mesh[:4]] mq = " AND ".join(mesh_terms) strategies.append(( "MeSH + " + query_type + " filter", mq + tf )) # Strategy 3: MeSH combined if mesh: mesh_terms = ['"' + t + '"[MeSH]' for t in mesh[:4]] strategies.append(("MeSH Combined", " AND ".join(mesh_terms))) # Strategy 4: Core MeSH (top 2) if len(mesh) >= 2: q = '"' + mesh[0] + '"[MeSH] AND "' + mesh[1] + '"[MeSH]' strategies.append(("Core MeSH", q)) # Strategy 5: Primary MeSH + filter if mesh and tf: q = '"' + mesh[0] + '"[MeSH]' + tf strategies.append(("Primary MeSH + " + query_type, q)) # Strategy 6: Title/Abstract strategies.append(("Title/Abstract", "(" + query + ")[tiab]")) # Strategy 7: All fields strategies.append(("All Fields", query)) return strategies def _iterative_search(self, query, analysis, query_type, max_results): strategies = self._build_strategies(query, analysis, query_type) return self._run_strategies(strategies, max_results) def _pico_iterative_search(self, pico, pico_analysis, max_results): strategies = [] # Build per-component queries comp_parts = [] for comp, analysis in pico_analysis.items(): mesh = analysis.get("mesh_found", []) if mesh: if len(mesh) > 1: mesh_terms = ['"' + t + '"[MeSH]' for t in mesh[:3]] joined = " OR ".join(mesh_terms) comp_parts.append("(" + joined + ")") else: comp_parts.append('"' + mesh[0] + '"[MeSH]') else: comp_parts.append("(" + pico[comp] + ")") # Strategy 1: Full PICO if len(comp_parts) >= 2: strategies.append(("Full PICO", " AND ".join(comp_parts))) # Strategy 2: Natural language combined = " ".join(pico.values()) strategies.append(("Natural Language", combined)) # Strategy 3: Component pairs for pair_name, keys in [("P+I", ["Population", "Intervention"]), ("P+O", ["Population", "Outcome"])]: parts = [] for k in keys: if k in pico_analysis: mesh = pico_analysis[k].get("mesh_found", []) if mesh: parts.append('"' + mesh[0] + '"[MeSH]') elif k in pico: parts.append(pico[k]) if len(parts) == 2: strategies.append((pair_name, " AND ".join(parts))) # Strategy 4: Broad strategies.append(("Broad", combined)) return self._run_strategies(strategies, max_results) def _run_strategies(self, strategies, max_results): all_articles = [] seen = set() log = [] for name, sq in strategies: if not sq: continue fetch_count = min(max_results * 2, 200) results = self._run_search(sq, fetch_count) log.append({"name": name, "query": sq, "found": len(results)}) for a in results: if a["pmid"] not in seen: seen.add(a["pmid"]) a["found_via"] = name all_articles.append(a) if len(all_articles) >= max_results * 3 and len(log) >= 3: break return all_articles, log # ================================================================ # ABSTRACT FETCHING # ================================================================ def _fetch_abstracts(self, pmids: List[str]) -> Dict[str, str]: """Fetch full abstracts. Batched. Works with or without API key.""" abstracts = {} if not pmids: return abstracts batch_size = 25 for i in range(0, len(pmids), batch_size): batch = pmids[i:i + batch_size] try: resp = requests.get( self.base_url + "/efetch.fcgi", params=self._api_params({ "db": "pubmed", "id": ",".join(batch), "rettype": "xml", "retmode": "xml", }), timeout=30, ) if resp.status_code != 200: continue try: root = ET.fromstring(resp.content) except ET.ParseError: self._regex_extract(batch, abstracts, resp.text) continue for art_elem in root.findall(".//PubmedArticle"): pmid_elem = art_elem.find(".//PMID") if pmid_elem is None: continue pmid = pmid_elem.text # Abstract parts = [] abs_elem = art_elem.find(".//Abstract") if abs_elem is not None: for txt in abs_elem.findall("AbstractText"): label = txt.get("Label", "") text = self._elem_text(txt) if text: if label: parts.append("**" + label + ":** " + text) else: parts.append(text) if parts: full = "\n\n".join(parts) # MeSH headings mesh_list = [] for m in art_elem.findall(".//MeshHeading/DescriptorName"): if m.text: mesh_list.append(m.text) if mesh_list: full += "\n\n**MeSH:** " + ", ".join(mesh_list[:10]) # Keywords kw_list = [] for k in art_elem.findall(".//Keyword"): if k.text: kw_list.append(k.text) if kw_list: full += "\n**Keywords:** " + ", ".join(kw_list[:10]) abstracts[pmid] = full except Exception: continue return abstracts def _elem_text(self, elem) -> str: """Get all text from XML element including children""" parts = [] if elem.text: parts.append(elem.text) for child in elem: if child.text: parts.append(child.text) if child.tail: parts.append(child.tail) return " ".join(parts).strip() def _regex_extract(self, pmids, abstracts, xml_text): """Fallback regex abstract extraction""" for pmid in pmids: if pmid in abstracts: continue pattern = ( r"]*>" + re.escape(pmid) + r"" r".*?(.*?)" ) match = re.search(pattern, xml_text, re.DOTALL) if match: text = re.sub(r"<[^>]+>", " ", match.group(1)) text = re.sub(r"\s+", " ", text).strip() if text: abstracts[pmid] = text # ================================================================ # RELEVANCE SCORING # ================================================================ def _score_relevance(self, articles, query, query_type): query_words = set(re.findall(r"[a-z]{3,}", query.lower())) query_words -= { "the", "and", "for", "with", "from", "that", "this", "are", "was", "were", "been", "have", "has", "how", "what", "which", "current", "recent", "new", "using", "based" } type_words = { "guidelines": ["guideline", "guidelines", "recommendation", "consensus", "management"], "systematic_review": ["systematic", "review", "meta-analysis"], "outcomes": ["outcome", "outcomes", "effectiveness", "efficacy"], "diagnosis": ["diagnosis", "diagnostic", "screening"], "treatment": ["treatment", "therapy", "therapeutic"], "epidemiology": ["prevalence", "incidence", "epidemiology"], } for a in articles: score = 0 tl = a.get("title", "").lower() ab = a.get("abstract", "").lower() title_words = set(re.findall(r"[a-z]{3,}", tl)) score += len(query_words & title_words) * 5 if ab: ab_words = set(re.findall(r"[a-z]{3,}", ab)) score += min(15, len(query_words & ab_words) * 2) score += 3 for w in type_words.get(query_type, []): if w in tl: score += 10 if w in ab: score += 3 year = self._extract_year(a.get("pubdate", "")) if year: if year >= 2023: score += 8 elif year >= 2020: score += 5 elif year >= 2015: score += 2 jl = a.get("journal", "").lower() if any(j in jl for j in [ "lancet", "bmj", "jama", "new england", "cochrane", "pediatrics", "annals", "nature", "plos" ]): score += 5 a["relevance_score"] = score articles.sort(key=lambda x: x.get("relevance_score", 0), reverse=True) return articles # ================================================================ # PUBMED SEARCH API # ================================================================ def _run_search(self, query, max_results): max_results = self._safe_int(max_results, 10, 1, 200) try: resp = requests.get( self.base_url + "/esearch.fcgi", params=self._api_params({ "db": "pubmed", "term": str(query), "retmode": "json", "retmax": str(max_results), "sort": "relevance", }), timeout=20, ) resp.raise_for_status() es = resp.json().get("esearchresult", {}) if "ERROR" in es: return [] ids = es.get("idlist", es.get("IdList", [])) if not ids: return [] resp = requests.get( self.base_url + "/esummary.fcgi", params=self._api_params({ "db": "pubmed", "id": ",".join(ids), "retmode": "json" }), timeout=20, ) resp.raise_for_status() sums = resp.json().get("result", {}) articles = [] for aid in ids: if aid not in sums or not isinstance(sums[aid], dict): continue art = sums[aid] if "title" not in art: continue articles.append({ "title": art.get("title", "Untitled"), "authors": ", ".join( a["name"] for a in art.get("authors", []) if isinstance(a, dict) and a.get("name") ), "pubdate": art.get("pubdate", ""), "journal": art.get("fulljournalname", ""), "volume": art.get("volume", ""), "issue": art.get("issue", ""), "pages": art.get("pages", ""), "doi": next( (x["value"] for x in art.get("articleids", []) if isinstance(x, dict) and x.get("idtype") == "doi"), "" ), "pmid": aid, "url": "https://pubmed.ncbi.nlm.nih.gov/" + aid + "/", "abstract": "", "ref_number": 0, }) return articles except Exception: return [] # ================================================================ # VANCOUVER STYLE REFERENCES # ================================================================ def _vancouver_ref(self, article: Dict) -> str: """Format one article as Vancouver reference""" ref_num = article.get("ref_number", 0) parts = [] # Authors authors = article.get("authors", "") if authors: auth_list = [a.strip() for a in authors.split(",") if a.strip()] if len(auth_list) > 6: auth_str = ", ".join(auth_list[:6]) + ", et al" else: auth_str = ", ".join(auth_list) parts.append(auth_str + ".") else: parts.append("[No authors listed].") # Title title = article.get("title", "Untitled").rstrip(".") parts.append(title + ".") # Journal journal = article.get("journal", "") if journal: parts.append(journal + ".") # Year;Volume(Issue):Pages year = self._extract_year(article.get("pubdate", "")) pub_detail = "" if year: pub_detail = str(year) vol = article.get("volume", "") if vol: if pub_detail: pub_detail += ";" pub_detail += vol issue = article.get("issue", "") if issue: pub_detail += "(" + issue + ")" pages = article.get("pages", "") if pages: pub_detail += ":" + pages if pub_detail: parts.append(pub_detail + ".") # DOI doi = article.get("doi", "") if doi: parts.append("doi:" + doi + ".") # PMID pmid = article.get("pmid", "") if pmid: parts.append("PMID: " + pmid + ".") ref_text = " ".join(parts) return "[" + str(ref_num) + "] " + ref_text def _build_vancouver_list(self, articles: List[Dict]) -> str: """Build complete Vancouver reference list""" md = "" for a in articles: md += self._vancouver_ref(a) + "\n\n" return md # ================================================================ # AI CITATION INSTRUCTIONS # ================================================================ def _build_citation_instructions(self, articles: List[Dict], query: str) -> str: """ Build explicit instructions that tell the AI model how to cite using reference numbers. The AI reads this and learns to use [1], [2], etc. instead of 'undefined'. """ md = "## ๐Ÿค– AI Analysis Ready\n\n" md += "Abstracts are loaded. You can ask me to analyze, " md += "summarize, or compare these findings.\n\n" # Reference map md += "### Reference Map for Citation\n\n" md += "When discussing findings from these articles, " md += "ALWAYS cite using the numbered references below:\n\n" for a in articles: ref = a.get("ref_number", 0) auth_list = a.get("authors", "").split(", ") first_author = auth_list[0] if auth_list and auth_list[0] else "Unknown" yr = self._extract_year(a.get("pubdate", "")) or "n.d." short_title = a.get("title", "") if len(short_title) > 80: short_title = short_title[:77] + "..." md += ("- **[" + str(ref) + "]** = " + first_author + " et al. (" + str(yr) + ") โ€” " + short_title + "\n") md += "\n" md += "**CITATION INSTRUCTIONS FOR AI:**\n\n" md += "1. When you mention ANY finding, conclusion, or recommendation " md += "from the articles above, you MUST cite the source using " md += "the reference number in square brackets.\n" md += "2. Use the format: [1], [2], [3] or [1,3,5] for multiple sources.\n" md += "3. Example: 'Current guidelines recommend a stepwise approach " md += "to management [1]. PPIs should be reserved for confirmed " md += "erosive disease [2,4]. Conservative measures are first-line [1,3].'\n" md += "4. NEVER write 'undefined', 'source', or leave citations blank.\n" md += "5. EVERY factual claim must have at least one [number] citation.\n" md += "6. If unsure which reference supports a claim, " md += "use the most likely one based on the title and abstract.\n\n" # Suggested prompts md += "### Ask me:\n\n" md += "> Summarize the key findings from these articles\n\n" md += "> What is the current evidence on " + query + "?\n\n" md += "> Compare the conclusions across these studies\n\n" md += "> What are the main recommendations?\n\n" md += "> Are there any conflicting findings?\n\n" return md # ================================================================ # MAIN RESULTS FORMATTER # ================================================================ def _format_results(self, query, analysis, query_type, search_log, articles, total, show_abs): md = "# ๐Ÿ“š PubMed Search Results\n\n" md += "**Question:** " + query + "\n\n" # Query understanding md += "## ๐Ÿง  Query Understanding\n\n" md += "**Type:** " + query_type.replace("_", " ").title() + "\n" if analysis["mesh_found"]: md += "**MeSH:** " + ", ".join(analysis["mesh_found"]) + "\n" if analysis["query_translation"]: md += "\n```\n" + analysis["query_translation"] + "\n```\n" md += "\n" # Search process md += "## ๐Ÿ”ง Search (" md += str(len(search_log)) + " strategies, " md += str(total) + " candidates)\n\n" for s in search_log: icon = "โœ…" if s["found"] > 0 else "โญ•" md += icon + " **" + s["name"] + "** โ†’ " + str(s["found"]) + " \n" md += "\n" # Results count with_abs = sum(1 for a in articles if a.get("abstract")) md += "## ๐Ÿ“„ Top " + str(len(articles)) + " Results" if with_abs: md += " (" + str(with_abs) + " with abstracts)" md += "\n\n" # Article list with abstracts md += self._format_article_list(articles, show_abs) # Vancouver reference list md += "## ๐Ÿ“ References (Vancouver Style)\n\n" md += self._build_vancouver_list(articles) # AI citation instructions if with_abs: md += self._build_citation_instructions(articles, query) md += self._format_next_steps() return md def _format_article_list(self, articles, show_abstracts=True): md = "" for a in articles: ref = a.get("ref_number", 0) score = a.get("relevance_score", 0) stars = min(5, max(1, score // 5)) md += "### [" + str(ref) + "] " + a.get("title", "Untitled") + "\n\n" if a.get("authors"): auth_list = a["authors"].split(", ") if len(auth_list) > 3: auth_str = ", ".join(auth_list[:3]) + ", et al." else: auth_str = a["authors"] md += "**Authors:** " + auth_str + "\n\n" info = [] if a.get("journal"): info.append("*" + a["journal"] + "*") if a.get("pubdate"): info.append(a["pubdate"]) v = a.get("volume", "") if v: if a.get("issue"): v += "(" + a["issue"] + ")" if a.get("pages"): v += ":" + a["pages"] info.append(v) if info: md += " | ".join(info) + "\n\n" links = "" if a.get("doi"): links += "[DOI](https://doi.org/" + a["doi"] + ") ยท " links += "[PMID " + a["pmid"] + "](" + a["url"] + ")" links += " ยท " + "โญ" * stars md += links + "\n\n" if show_abstracts and a.get("abstract"): md += "
\n๐Ÿ“‹ Abstract [" + str(ref) + "]\n\n" md += a["abstract"] + "\n\n
\n\n" md += "---\n\n" return md def _format_next_steps(self): return ( "\n## ๐Ÿ’ก Next Steps\n\n" "| Say | Get |\n|-----|-----|\n" "| `get results as list` | Vancouver reference list |\n" "| `get results as ris` | RIS file for Zotero |\n" "| `get results as summary` | AI synthesis with citations |\n" "| `get results as abstracts` | All abstracts for reading |\n" "| `get results as detailed` | Full metadata |\n\n" ) def _format_no_results(self, query, analysis, search_log): md = "# No Results\n\n**Query:** " + query + "\n\n" if analysis["query_translation"]: md += "```\n" + analysis["query_translation"] + "\n```\n\n" for s in search_log: md += "โŒ " + s["name"] + ": `" + s["query"] + "`\n\n" md += "Try simpler terms or `find_mesh`.\n" return md # ================================================================ # OUTPUT FORMATS # ================================================================ def _format_vancouver_list(self): """Numbered Vancouver reference list""" md = "# ๐Ÿ“‹ References (" + str(len(self._last_results)) + ")\n\n" md += "**Search:** " + self._last_query + "\n\n" md += self._build_vancouver_list(self._last_results) md += "\n> Say `get results as ris` for Zotero export\n" return md def _export_ris(self): ris = "" for a in self._last_results: ris += self._to_ris(a) return ( "# ๐Ÿ“ฅ RIS Export (" + str(len(self._last_results)) + " refs)\n\n" "1. Copy the code block\n" "2. Save as `references.ris`\n" "3. Zotero โ†’ File โ†’ Import\n\n" "```ris\n" + ris + "```\n" ) def _format_abstracts_only(self): md = "# ๐Ÿ“‹ Abstracts (" + str(len(self._last_results)) + ")\n\n" md += "**Search:** " + self._last_query + "\n\n---\n\n" for a in self._last_results: ref = a.get("ref_number", 0) yr = self._extract_year(a.get("pubdate", "")) or "n.d." auth_list = a.get("authors", "").split(", ") first = auth_list[0] if auth_list and auth_list[0] else "Unknown" md += "## [" + str(ref) + "] " + a.get("title", "") + "\n" md += "*" + first + " et al. (" + str(yr) + ") โ€” " md += a.get("journal", "") + "*\n\n" if a.get("abstract"): md += a["abstract"] + "\n\n" else: md += "*No abstract available.*\n\n" md += "---\n\n" return md def _synthesize(self): articles = self._last_results md = "# ๐Ÿ“Š Research Summary\n\n" md += "**Question:** " + self._last_query + "\n" md += "**Articles:** " + str(len(articles)) + "\n\n" years = [self._extract_year(a.get("pubdate", "")) for a in articles] years = [y for y in years if y] if years: md += "**Range:** " + str(min(years)) + "โ€“" + str(max(years)) + "\n\n" with_abs = sum(1 for a in articles if a.get("abstract")) md += "**Abstracts available:** " + str(with_abs) + "/" + str(len(articles)) + "\n\n" # Journals journals = {} for a in articles: j = a.get("journal", "Unknown") journals[j] = journals.get(j, 0) + 1 md += "## Sources\n\n" for j, c in sorted(journals.items(), key=lambda x: -x[1])[:8]: md += "- " + j + " (" + str(c) + ")\n" md += "\n" # Themes from abstracts + titles all_text = " ".join( a.get("title", "") + " " + a.get("abstract", "") for a in articles ) wf = {} stops = { "the", "and", "for", "with", "from", "that", "this", "was", "were", "been", "have", "has", "study", "review", "patients", "results", "methods", "conclusion", "background", "objective", "clinical", "using", "based", "among", "between", "group", "data", "included", "also", "more", "than", "which", "these", "other" } for w in re.findall(r"[a-z]{4,}", all_text.lower()): if w not in stops: wf[w] = wf.get(w, 0) + 1 md += "## Key Themes\n\n" for w, c in sorted(wf.items(), key=lambda x: -x[1])[:15]: if c >= 3: md += "- **" + w + "** (" + str(c) + "ร—)\n" md += "\n" # Articles with citations md += "## Articles\n\n" for a in articles[:20]: ref = a.get("ref_number", 0) yr = self._extract_year(a.get("pubdate", "")) or "n.d." auth_list = a.get("authors", "").split(", ") first = auth_list[0] if auth_list and auth_list[0] else "Unknown" md += "**[" + str(ref) + "]** " + first + " (" + str(yr) + "). " md += a.get("title", "") + " *" + a.get("journal", "") + "*\n" if a.get("abstract"): snippet = a["abstract"][:200] if len(a["abstract"]) > 200: snippet += "..." md += " " + snippet + "\n" md += "\n" md += "---\n\n" # Citation instructions for synthesis md += "### Reference Map\n\n" for a in articles: ref = a.get("ref_number", 0) auth_list = a.get("authors", "").split(", ") first = auth_list[0] if auth_list and auth_list[0] else "Unknown" yr = self._extract_year(a.get("pubdate", "")) or "n.d." md += "**[" + str(ref) + "]** = " + first + " et al. (" + str(yr) + ")\n" md += "\n**Use [number] citations when discussing these findings.**\n" md += "**NEVER write 'undefined'. Always use [1], [2], etc.**\n" return md def _format_detailed(self): md = "# ๐Ÿ“‘ Detailed (" + str(len(self._last_results)) + ")\n\n" for a in self._last_results: ref = a.get("ref_number", 0) md += "## [" + str(ref) + "] " + a.get("title", "") + "\n\n" md += "- **Authors:** " + a.get("authors", "Unknown") + "\n" md += "- **Journal:** " + a.get("journal", "Unknown") + "\n" md += "- **Date:** " + a.get("pubdate", "Unknown") + "\n" if a.get("doi"): md += "- **DOI:** [" + a["doi"] + "](https://doi.org/" + a["doi"] + ")\n" md += "- **PMID:** [" + a["pmid"] + "](" + a["url"] + ")\n" md += "- **Relevance:** " + str(a.get("relevance_score", 0)) md += " ยท via " + a.get("found_via", "?") + "\n" if a.get("abstract"): md += "\n**Abstract:**\n\n" + a["abstract"] + "\n" md += "\n---\n\n" return md # ================================================================ # UTILITIES # ================================================================ def _to_ris(self, a): ris = "TY - JOUR\n" if a.get("authors"): for au in a["authors"].split(", "): au = au.strip() if au: ris += "AU - " + au + "\n" title = a.get("title", "").rstrip(".") ris += "T1 - " + title + "\n" if a.get("journal"): ris += "JO - " + a["journal"] + "\n" if a.get("pubdate"): m = re.search(r"(\d{4})", a["pubdate"]) if m: ris += "PY - " + m.group(1) + "\n" ris += "DA - " + a["pubdate"] + "\n" if a.get("volume"): ris += "VL - " + a["volume"] + "\n" if a.get("issue"): ris += "IS - " + a["issue"] + "\n" if a.get("pages"): if "-" in a["pages"]: sp, ep = a["pages"].split("-", 1) ris += "SP - " + sp.strip() + "\n" ris += "EP - " + ep.strip() + "\n" else: ris += "SP - " + a["pages"] + "\n" if a.get("doi"): ris += "DO - " + a["doi"] + "\n" if a.get("url"): ris += "UR - " + a["url"] + "\n" if a.get("abstract"): abstract = a["abstract"][:2000] abstract = re.sub(r"\*\*[^*]+:\*\*\s*", "", abstract) ris += "AB - " + abstract + "\n" ris += "ER -\n\n" return ris def _extract_year(self, d): if not d: return None m = re.search(r"(\d{4})", str(d)) return int(m.group(1)) if m else None def _safe_int(self, v, default=10, mn=1, mx=200): try: r = int(float(str(v))) except (TypeError, ValueError): r = default return max(mn, min(mx, r)) def _error_msg(self, msg): return ( "**Search Error:** " + msg + "\n\n" "Try:\n" "- Simpler phrasing\n" "- `find_mesh` to check terms\n" "- `pico_search` for structured queries\n" )