diff --git a/pubmed_search_tool_with_reference_formatting.py b/pubmed_search_tool_with_reference_formatting.py new file mode 100644 index 0000000..bef9717 --- /dev/null +++ b/pubmed_search_tool_with_reference_formatting.py @@ -0,0 +1,1163 @@ +""" +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" + )