| """HTML fragment renderers: source badges, reasoning trace blocks, AI-tools footer, contacts. |
| |
| Styling contract: renderers emit class names only. The answer sanitizer |
| (src/services/html_sanitizer.py) strips every style attribute and every svg element, |
| so inline styles never reached the browser. Presentation lives in the |
| frontend stylesheet: |
| - lb-* classes: LibBee server components (trace, source badge, events, hours, steps) |
| - kb-* classes: research-answer components already styled by the LibGuides page |
| (kb-ai-platform-*, kb-search-process*, kb-database-*, kb-boolean-query-*) |
| """ |
| import asyncio |
| import html |
| import json |
| import logging |
| import os |
| import re |
| import time |
| import uuid |
| from datetime import datetime |
| from typing import Any, Dict, List, Literal, Optional, Tuple |
| from urllib.parse import quote |
|
|
| import httpx |
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| from src.config import get_settings, LIBBEE_VERSION |
| from src.services.staff_service import ( |
| STAFF_DIRECTORY, |
| match_staff_name, |
| match_staff_role, |
| should_attempt_staff_lookup, |
| staff_name_answer, |
| staff_role_answer, |
| ) |
|
|
| from src.agentcore.models import SearchContextPayload |
| from src.agentcore.constants import ASK_LIBRARIAN_URL, PRIMO_AI_URL, _openathens_url |
| from src.agentcore.utils import ( |
| _escape, |
| _find_staff_by_token, |
| _primo_clean_url, |
| _shared_build_pubmed_url, |
| ) |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _SOURCE_BADGES = { |
| "rag": ("📖", "KU Library Knowledge Base", "kb"), |
| "primo": ("📚", "KU Library PRIMO", "library"), |
| "semantic": ("🎓", "Semantic Scholar", "scholarly"), |
| "consensus_badge": ("🧬", "Consensus", "scholarly"), |
| "semantic_badge": ("🎓", "Semantic Scholar", "scholarly"), |
| "pubmed": ("🏥", "PubMed", "scholarly"), |
| "web": ("🌐", "Web Search", "web"), |
| "web_live": ("🌐", "Web Search · Live", "web"), |
| "libbee": ("🐝", "LibBee", "libbee"), |
| "ku_library": ("📋", "KU Library", "library"), |
| "llm": ("🤖", "AI response · web not verified", "llm"), |
| } |
|
|
|
|
| def _source_badge(source: str, url: str = "") -> str: |
| icon, label, variant = _SOURCE_BADGES.get(source, ("ℹ️", source, "llm")) |
| inner = f"{icon} Source: {_escape(label)}" |
| if url: |
| inner = f'<a href="{_escape(url)}" target="_blank">{inner}</a>' |
| return f'<div class="lb-source lb-source--{variant}">{inner}</div>' |
|
|
|
|
| def _tool_urls(context: SearchContextPayload) -> Dict[str, str]: |
| ai_query = quote(context.ai_tool_query or context.display_topic or context.topic) |
| primo_ai_query = quote(context.ai_tool_query or context.display_topic or context.topic) |
| urls: Dict[str, str] = { |
| "leapspace": f"https://www.sciencedirect.com/leapspace?query={ai_query}", |
| "scopus_ai": _openathens_url(f"https://www.scopus.com/pages/ai?query={ai_query}"), |
| "ebsco_ai": ( |
| f"https://research.ebsco.com/c/yst6t2/search/results" |
| f"?q={ai_query}" |
| f"&autocorrect=y" |
| f"&expanders=concept" |
| f"&limiters=None" |
| f"&searchMode=enhanced" |
| f"&searchSegment=all-results" |
| ), |
| "primo_ai": f"{PRIMO_AI_URL}{primo_ai_query}", |
| "consensus": f"https://consensus.app/results/?q={ai_query}", |
| "primo_discovery": _primo_clean_url(context), |
| } |
| if context.intent == "search_medical": |
| urls["pubmed"] = _shared_build_pubmed_url( |
| context.topic, context.year_from, context.year_to, context.peer_reviewed, |
| open_access=context.open_access |
| ) |
| return urls |
|
|
|
|
| def _contact_footer_for_query(question: str, context: Optional[SearchContextPayload]) -> str: |
| q = (question or "").lower() |
| specialist: Optional[dict] = None |
| if any(t in q for t in ["interlibrary", "ill request", "ill form", "borrow from another", |
| "request article", "request a book", "not available at ku", |
| "not in ku library", "inter library", "inter-library"]): |
| footer = ( |
| "<br><br><strong>Library contact points</strong><br>" |
| f'General help: <a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a> · ' |
| '<a href="mailto:libse@ku.ac.ae">libse@ku.ac.ae</a><br>' |
| 'ILL requests: <a href="mailto:ill@ku.ac.ae">ill@ku.ac.ae</a> · ' |
| '<strong>Zainab Al Ali</strong> · ' |
| '<a href="mailto:zainab.alali@ku.ac.ae">zainab.alali@ku.ac.ae</a> · ' |
| '<a href="tel:+97123124321">+971 2 312 4321</a>' |
| ) |
| return footer |
| elif any(t in q for t in [ |
| "apc", "article processing charge", "article processing fee", |
| "open access fee", "oa fee", "oa fund", "oa publishing", |
| "publish open access", "open access publish", "how to publish open access", |
| "open access journal", "open access mandate", "open access policy", |
| "waiver", "gold open access", "diamond open access", "hybrid journal", |
| "creative commons", "cc by", "cc licence", "openaccess", "open access", |
| ]): |
| |
| specialist = _find_staff_by_token("walter") |
| elif any(t in q for t in [ |
| "research support", "research service", "research consultation", |
| "research librarian", "research help", "research advice", |
| "scholarly communication", "bibliometric", "research impact", |
| "citation analysis", "h-index", "journal ranking", "impact factor", |
| "orcid", "khazna", "repository", "scival", "scopus profile", |
| "author profile", "research data", "data management", |
| "libguide", "research guide", "ai tool", "ai for research", |
| "publication advice", "journal selection", "where to publish", |
| ]): |
| |
| specialist = _find_staff_by_token("nikesh") |
| elif any(t in q for t in ["database access", "proxy", "eresource", |
| "e-resource", "remote access", "access problem", "off campus"]): |
| specialist = _find_staff_by_token("rani") |
| elif any(t in q for t in ["website", "system", "technical", "login issue", "page not loading"]): |
| specialist = _find_staff_by_token("walter") |
| elif context and context.intent == "search_medical": |
| |
| specialist = _find_staff_by_token("jason") |
| elif context and context.intent == "search_academic": |
| specialist = _find_staff_by_token("nikesh") |
| footer = ( |
| "<br><br><strong>Library contact points</strong><br>" |
| f'General help: <a href="{ASK_LIBRARIAN_URL}" target="_blank">Ask a Librarian</a> · ' |
| '<a href="mailto:libse@ku.ac.ae">libse@ku.ac.ae</a>' |
| ) |
| if specialist: |
| footer += ( |
| f'<br>Relevant contact: <strong>{_escape(specialist.get("full_name", ""))}</strong> — ' |
| f'{_escape(specialist.get("role", ""))}<br>' |
| f'<a href="mailto:{_escape(specialist.get("email", ""))}">{_escape(specialist.get("email", ""))}</a>' |
| ) |
| phone = specialist.get("phone") or specialist.get("work_phone") |
| if phone: |
| footer += f' · <a href="tel:{_escape(str(phone).replace(" ", ""))}">{_escape(phone)}</a>' |
| return footer |
|
|
|
|
| def _thought_block(intent: str, tool: str, question: str, extra: str = "") -> str: |
| labels = { |
| "library_info": ("📋", "Library information lookup", "I searched the KU Library knowledge base for relevant content."), |
| "general": ("🌐", "General knowledge / web search", "I answered using a web search for current information."), |
| "general_recent": ("🔴", "Live web search", "I ran a live web search for the most up-to-date information."), |
| "social_greeting": ("🐝", "Conversational response", "I responded directly as LibBee, the KU Library AI Assistant."), |
| "social": ("🐝", "Conversational response", "I responded directly as LibBee, the KU Library AI Assistant."), |
| "general_knowledge": ("🤖", "AI response", "Current web information could not be verified for this answer."), |
| "events": ("📅", "Library events lookup", "I checked the library events service. Available events or the official calendar link appear below."), |
| "staff_lookup": ("👤", "Staff directory lookup", "I matched your request to the KU Library staff directory."), |
| "hours": ("⏰", "Library hours lookup", "I checked the library hours service. Available hours or the official schedule link appear below."), |
| "libcal_live": ("⏰", "Live library hours", "I checked the library hours service. Available hours or the official schedule link appear below."), |
| "campus": ("📍", "Campus information lookup", "I retrieved KU Library campus information from the knowledge base."), |
| "database_rec": ("🗄️", "Database recommendation", "I matched your subject to the KU Library e-resources database."), |
| "rag_search": ("📖", "Knowledge base search", "I searched the supplied KU Library knowledge base."), |
| } |
| icon, label, reasoning = labels.get(intent, ("🤖", "AI response", "I processed your question and generated a response.")) |
| if extra: |
| reasoning += " " + extra |
| return ( |
| '<details class="lb-trace">' |
| f'<summary class="lb-trace-summary"><span class="lb-trace-icon">{icon}</span> My approach</summary>' |
| '<div class="lb-trace-body">' |
| f'<strong>Question:</strong> {_escape(question)}<br>' |
| f'<strong>Classified as:</strong> {label}<br>' |
| f'<strong>How I answered:</strong> {reasoning}' |
| '</div></details>' |
| ) |
|
|
|
|
| def _search_trace_block(question: str, context: SearchContextPayload) -> str: |
| topic = _escape(context.display_topic or context.topic) |
| primo_boolean = _escape(context.primo_boolean_query or "") |
| source_label = "PubMed" if context.source == "pubmed" else "KU PRIMO Library Discovery" |
| intent_label = { |
| "search_academic": "academic literature search", |
| "search_medical": "medical / clinical literature search", |
| }.get(context.intent, "library search") |
| filters_parts = [] |
| if context.peer_reviewed: |
| filters_parts.append("peer reviewed requested (verify in PubMed)" if context.source == "pubmed" else "peer reviewed only") |
| if context.open_access: |
| filters_parts.append("free full text" if context.source == "pubmed" else "open access") |
| if context.year_from and context.year_to: |
| filters_parts.append(f"{context.year_from}–{context.year_to}") |
| elif context.year_from: |
| filters_parts.append(f"from {context.year_from}") |
| rtype = {"articles": "articles", "books": "books", "both": "articles and books"}.get( |
| context.resource_type, "articles" |
| ) |
| filters_parts.append(rtype) |
| filters_str = ", ".join(filters_parts) |
| reasoning = ( |
| f"I classified this as a <strong>{intent_label}</strong>, " |
| f"extracted the topic <strong>{topic}</strong>, " |
| f"prepared a search query and filters, " |
| f"for <strong>{source_label}</strong>: <strong>{filters_str}</strong>." |
| ) |
| if context.intent == "search_medical" and context.source != "pubmed": |
| reasoning += " For medical topics, PubMed and Embase are also recommended." |
| block = ( |
| '<details class="kb-search-process">' |
| '<summary class="kb-search-process-summary">' |
| '<span class="kb-search-process-icon">🔍</span> How I searched' |
| '</summary>' |
| '<div class="lb-trace-body">' |
| f'<strong>Your question:</strong> {_escape(question)}<br>' |
| f'<strong>Interpreted topic:</strong> {topic}<br>' |
| ) |
| if primo_boolean: |
| block += ( |
| f'<strong>Prepared query for {source_label}:</strong><br>' |
| f'<code class="lb-code">{primo_boolean}</code>' |
| ) |
| block += f'<strong>Reasoning:</strong> {reasoning}<br>' |
| block += '<span class="lb-muted">✨ AI research tools for deeper exploration are listed at the end of this answer.</span>' |
| block += '</div></details>' |
| return block |
|
|
|
|
| |
| _PLATFORM_ICONS = { |
| "primo": "📚", "leapspace": "🚀", "scopus": "🔎", "ebsco": "📘", |
| "consensus": "🧬", "semanticscholar": "🎓", "perplexity": "💬", "pubmed": "🏥", |
| } |
|
|
| _DB_ICONS = { |
| "PubMed": "🔬", "PubMed@KU": "🔬", "Embase": "🏥", "CINAHL Ultimate": "🏥", |
| "Cochrane Library": "🔬", "UpToDate": "🏥", |
| "IEEE Xplore": "💻", "ACM Digital Library": "💻", |
| "ScienceDirect": "🔬", "Scopus": "🔬", "Web of Science": "🔬", "SpringerLink": "📚", |
| "ACS Publications": "⚛️", "APS Journals": "⚛️", "Knovel": "🔧", |
| "Business Source Complete": "📊", "Emerald Insight": "📊", |
| "Taylor & Francis": "📚", "EBSCO": "📚", |
| } |
|
|
|
|
| def _ai_tools_footer(context): |
| from urllib.parse import quote as _q |
|
|
| ai_q = _q(context.ai_tool_query or context.display_topic or context.topic) |
| bool_q = context.primo_boolean_query or "" |
| primo_ai = PRIMO_AI_URL + ai_q |
| is_med = context.intent == "search_medical" |
| pubmed_url = _shared_build_pubmed_url( |
| context.topic, context.year_from, context.year_to, context.peer_reviewed, |
| open_access=context.open_access |
| ) if is_med else "" |
|
|
| platforms = [ |
| ("PRIMO AI Assistant", primo_ai, "primo"), |
| ("LeapSpace", "https://www.sciencedirect.com/leapspace?query=" + ai_q, "leapspace"), |
| ("Scopus AI", _openathens_url("https://www.scopus.com/pages/ai?query=" + ai_q), "scopus"), |
| ("EBSCO Research AI", |
| "https://research.ebsco.com/c/yst6t2/search/results?q=" + ai_q |
| + "&autocorrect=y&expanders=concept&limiters=None&searchMode=enhanced&searchSegment=all-results", "ebsco"), |
| ("Consensus", "https://consensus.app/results/?q=" + ai_q, "consensus"), |
| ("Semantic Scholar", "https://www.semanticscholar.org/search?q=" + ai_q + "&sort=relevance", "semanticscholar"), |
| ("Perplexity", "https://www.perplexity.ai/search?q=" + ai_q, "perplexity"), |
| ] |
| if is_med and pubmed_url: |
| platforms.append(("PubMed", pubmed_url, "pubmed")) |
|
|
| def _btn(label, url, icon_key): |
| icon = _PLATFORM_ICONS.get(icon_key, "🐝") |
| return ( |
| f'<a class="kb-ai-platform-button" href="{_escape(url)}" target="_blank">' |
| f'<span class="kb-ai-platform-icon">{icon}</span>{_escape(label)}</a>' |
| ) |
|
|
| btn_rows = "" |
| for k in range(0, len(platforms), 2): |
| pair = platforms[k:k + 2] |
| cells = "".join(f"<td>{_btn(lbl, url, key)}</td>" for lbl, url, key in pair) |
| btn_rows += f"<tr>{cells}</tr>" |
|
|
| platforms_html = ( |
| '<div class="kb-ai-platform-panel">' |
| '<div class="kb-ai-platform-heading">' |
| '<span class="kb-ai-platform-heading-icon">🤖</span> Search your query across AI platforms</div>' |
| '<div class="kb-ai-platform-subtitle">' |
| 'Your query is pre-loaded. Click any platform to search instantly.</div>' |
| f'<table class="kb-ai-platform-grid"><tbody>{btn_rows}</tbody></table>' |
| '</div>' |
| ) |
|
|
| if bool_q: |
| and_count = bool_q.upper().count(" AND ") |
| if and_count >= 1: |
| bool_label = "Use this in PRIMO Advanced Search or any database for precise results:" |
| else: |
| bool_label = "Try adding synonyms with OR to find more relevant results:" |
| bool_html = ( |
| '<div class="kb-boolean-query-panel">' |
| '<div class="kb-boolean-query-title">💡 Boolean search tip</div>' |
| f'<div class="lb-boolean-hint">{bool_label}</div>' |
| f'<code class="kb-boolean-query-text">{_escape(bool_q)}</code>' |
| '<div class="lb-boolean-hint">Paste into the Advanced Search box of any database below.</div>' |
| '</div>' |
| ) |
| else: |
| bool_html = "" |
|
|
| all_lower = ((context.ai_tool_query or "") + " " + (context.topic or "")).lower() |
|
|
| def _db_row(name, url, desc): |
| icon = _DB_ICONS.get(name, "🐝") |
| return ( |
| '<div class="kb-database-row">' |
| f'<span class="kb-database-icon">{icon}</span>' |
| f'<a class="kb-database-name" href="{_escape(url)}" target="_blank">{_escape(name)}</a>' |
| f'<span class="lb-muted"> — {_escape(desc)}</span>' |
| '</div>' |
| ) |
|
|
| if any(k in all_lower for k in ["medicine","medical","clinical","nursing","pharmacy","health","pubmed","embase","cinahl","cochrane","uptodate"]): |
| db_subject = "Medicine & Health Sciences" |
| dbs = [ |
| ("PubMed","https://pubmed.ncbi.nlm.nih.gov/","Core biomedical literature"), |
| ("Embase",_openathens_url("https://www.embase.com"),"Biomedical, drugs, devices, clinical medicine"), |
| ("CINAHL Ultimate",_openathens_url("https://search.ebscohost.com/login.aspx?authtype=ip,uid&profile=ehost&defaultdb=cul"),"Nursing and allied health"), |
| ("Cochrane Library",_openathens_url("https://www.cochranelibrary.com"),"Systematic reviews and evidence-based medicine"), |
| ("UpToDate",_openathens_url("https://www.uptodate.com"),"Point-of-care clinical decision support"), |
| ] |
| elif any(k in all_lower for k in ["machine learning","deep learning","artificial intelligence","neural network","computer science","software","algorithm","cybersecurity","robotics","nlp"]): |
| db_subject = "Computer Science & AI" |
| dbs = [ |
| ("IEEE Xplore",_openathens_url("https://ieeexplore.ieee.org"),"Core CS, AI, electronics and systems"), |
| ("ACM Digital Library",_openathens_url("https://dl.acm.org"),"Computer science journals, proceedings and books"), |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Full text across CS and applied computing"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Discovery and citation tracking"), |
| ("Web of Science",_openathens_url("https://www.webofscience.com"),"Citation index"), |
| ] |
| elif any(k in all_lower for k in ["climate","global warming","environment","sustainability","ecology","carbon","emission","renewable","sea level","maldives","island"]): |
| db_subject = "Environment & Climate" |
| dbs = [ |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Full text across science, environment and engineering"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Broad discovery and citation tracking"), |
| ("Web of Science",_openathens_url("https://www.webofscience.com"),"High-impact journal coverage and citation index"), |
| ("SpringerLink",_openathens_url("https://link.springer.com"),"Springer journals and books"), |
| ] |
| elif any(k in all_lower for k in ["engineering","mechanical","civil","electrical","aerospace","structural","petroleum","chemical","materials"]): |
| db_subject = "Engineering" |
| dbs = [ |
| ("IEEE Xplore",_openathens_url("https://ieeexplore.ieee.org"),"Electrical, electronics and systems engineering"), |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Full text across engineering disciplines"), |
| ("Knovel",_openathens_url("https://app.knovel.com"),"Engineering handbooks and reference data"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Discovery and citation tracking"), |
| ("Web of Science",_openathens_url("https://www.webofscience.com"),"Citation index"), |
| ] |
| elif any(k in all_lower for k in ["business","management","finance","economics","leadership","marketing","strategy","entrepreneurship"]): |
| db_subject = "Business & Management" |
| dbs = [ |
| ("Business Source Complete",_openathens_url("https://search.ebscohost.com"),"Core business and management literature"), |
| ("Emerald Insight",_openathens_url("https://www.emerald.com/insight"),"Management, strategy and leadership"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Broad discovery and citation tracking"), |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Business and social sciences full text"), |
| ] |
| elif any(k in all_lower for k in ["physics","chemistry","quantum","nuclear","materials science","nanotechnology","optics"]): |
| db_subject = "Physics, Chemistry & Materials" |
| dbs = [ |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Full text across physical sciences"), |
| ("ACS Publications",_openathens_url("https://pubs.acs.org"),"Core chemistry and chemical engineering"), |
| ("APS Journals",_openathens_url("https://journals.aps.org"),"American Physical Society core physics journals"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Discovery and citation tracking"), |
| ("Web of Science",_openathens_url("https://www.webofscience.com"),"Citation index"), |
| ] |
| else: |
| db_subject = "General" |
| dbs = [ |
| ("EBSCO",_openathens_url("https://search.ebscohost.com"),"Multidisciplinary full text"), |
| ("ScienceDirect",_openathens_url("https://www.sciencedirect.com"),"Science, engineering and medicine"), |
| ("SpringerLink",_openathens_url("https://link.springer.com"),"Journals and books across all disciplines"), |
| ("Taylor & Francis",_openathens_url("https://www.tandfonline.com"),"Humanities, social sciences and STEM"), |
| ("Emerald Insight",_openathens_url("https://www.emerald.com/insight"),"Management and social sciences"), |
| ("Scopus",_openathens_url("https://www.scopus.com"),"Citation discovery, all disciplines"), |
| ("Web of Science",_openathens_url("https://www.webofscience.com"),"Citation index, all disciplines"), |
| ] |
|
|
| dbs_html_rows = "".join(_db_row(name, url, desc) for name, url, desc in dbs) |
|
|
| db_section = ( |
| '<div class="kb-database-list">' |
| f'<div class="lb-db-heading">📂 Recommended KU databases — {_escape(db_subject)}</div>' |
| + dbs_html_rows + |
| '<div class="lb-db-more">' |
| '<a href="https://library.ku.ac.ae/eresources" target="_blank">' |
| '🗄️ Browse the current KU resource directory →</a></div>' |
| '</div>' |
| ) |
|
|
| filter_note = "" |
| if is_med and context.peer_reviewed: |
| filter_note += '<p><small>PubMed has no peer-review filter. Use PRIMO’s Peer Reviewed facet and verify the journal’s publication process.</small></p>' |
| if is_med and context.open_access: |
| filter_note += '<p><small>The PubMed link uses its Free full text filter; free access does not establish an open reuse licence.</small></p>' |
| return platforms_html + filter_note + bool_html + db_section |
|
|
