"""HTML fragment renderers: source badges, reasoning trace blocks, AI-tools footer, contacts.""" 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 from src.agentcore.utils import ( _escape, _find_staff_by_token, _primo_clean_url, _shared_build_pubmed_url, ) logger = logging.getLogger(__name__) def _source_badge(source: str, url: str = "") -> str: styles = { "rag": ("📖", "KU Library Knowledge Base", "#fef3c7", "#92400e"), "primo": ("📚", "KU Library PRIMO", "#eff6ff", "#1e40af"), "semantic": ("🎓", "Semantic Scholar", "#f0fdf4", "#166534"), "consensus_badge": ("🧬", "Consensus", "#fdf4ff", "#86198f"), "semantic_badge": ("🎓", "Semantic Scholar", "#f0fdf4", "#166534"), "pubmed": ("🏥", "PubMed", "#d1fae5", "#065f46"), "web": ("🌐", "Web Search", "#f5f3ff", "#4c1d95"), "web_live": ("🌐", "Web Search · Live", "#fdf4ff", "#86198f"), "libbee": ("🐝", "LibBee", "#fef9c3", "#713f12"), "ku_library": ("📋", "KU Library", "#eff6ff", "#1e40af"), "llm": ("🤖", "AI Knowledge Base", "#f3f4f6", "#374151"), } icon, label, bg, fg = styles.get(source, ("ℹ️", source, "#f3f4f6", "#374151")) inner = f"{icon} Source: {label}" if url: inner = f'{inner}' return ( f'
{inner}
' ) 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": f"https://www-scopus-com.khalifa.idm.oclc.org/pages/ai?query={ai_query}", "ebsco_ai": ( f"https://research-ebsco-com.khalifa.idm.oclc.org/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 ) 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 = ( "

Library contact points
" f'General help: Ask a Librarian · ' 'libse@ku.ac.ae
' 'ILL requests: ill@ku.ac.ae · ' 'Suaad Al Jneibi · ' 'suaad.aljneibi@ku.ac.ae · ' '+971 2 312 4278' ) 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", ]): # Open Access & APC → Walter Brian Hall (OA lead at KU Library) 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", ]): # Research services → Nikesh Narayanan (Research & Access Services Librarian) 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": # Medical/clinical research questions only → Jason Fetty (Medical Librarian) specialist = _find_staff_by_token("jason") elif context and context.intent == "search_academic": specialist = _find_staff_by_token("nikesh") footer = ( "

Library contact points
" f'General help: Ask a Librarian · ' 'libse@ku.ac.ae' ) if specialist: footer += ( f'
Relevant contact: {_escape(specialist.get("full_name", ""))} — ' f'{_escape(specialist.get("role", ""))}
' f'{_escape(specialist.get("email", ""))}' ) phone = specialist.get("phone") or specialist.get("work_phone") if phone: footer += f' · {_escape(phone)}' 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."), "staff_lookup": ("👤", "Staff directory lookup", "I matched your request to the KU Library staff directory."), "hours": ("⏰", "Library hours lookup", "I fetched live library hours from the KU LibCal API."), "libcal_live": ("⏰", "Live library hours", "I fetched real-time hours from the KU LibCal API."), "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 ran a hybrid semantic + keyword search across the 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 ( f'
' f'' f'{icon} My approach — click to see' f'
' f'Question: {_escape(question)}
' f'Classified as: {label}
' f'How I answered: {reasoning}' f'
' ) 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 only") if context.open_access: filters_parts.append("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 {intent_label}, " f"extracted the topic {topic}, " f"built an LLM-optimised boolean query with quoted phrases and OR synonyms, " f"then searched {source_label} for {filters_str}." ) if context.intent == "search_medical" and context.source != "pubmed": reasoning += " For medical topics, PubMed and Embase are also recommended." block = ( f'
' f'' f'🔍 How I searched — click to expand' f'' f'
' f'Your question: {_escape(question)}
' f'Interpreted topic: {topic}
' ) if primo_boolean: block += ( f'Boolean query sent to {source_label}:
' f'{primo_boolean}' ) block += f'Reasoning: {reasoning}
' block += '✨ AI research tools for deeper exploration are listed at the end of this answer.' block += '
' return block 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_url = _primo_clean_url(context) 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 ) if is_med else "" KU = "https://khalifa.idm.oclc.org/login?url=" platforms = [ ("PRIMO AI Assistant", primo_ai, "primo"), ("LeapSpace", "https://www.sciencedirect.com/leapspace?query=" + ai_q, "leapspace"), ("Scopus AI", "https://www-scopus-com.khalifa.idm.oclc.org/pages/ai?query=" + ai_q, "scopus"), ("EBSCO Research AI", "https://research-ebsco-com.khalifa.idm.oclc.org/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")) _SVGS = { "bee": ( '' '' '' '' '' '' '' '' '' '' '' '' '' '' 'LB' '' ), "book": ( '' '' '' '' '' '' '' '' ), "microscope": ( '' '' '' '' '' '' '' '' '' ), "circuit": ( '' '' '' '' '' '' '' '' '' '' '' '' ), "leaf": ( '' '' '' '' '' '' '' '' ), "gear": ( '' '' '' '' '' ), "chart": ( '' '' '' '' '' '' '' ), "atom": ( '' '' '' '' '' '' ), "medical": ( '' '' '' '' '' ), "wrench": ( '' '' '' '' ), "primo": ( '' '' '' '' '' '' '' '' '' ), "scopus": ( '' '' 'Sc' '' ), "ebsco": ( '' '' 'EBS' '' ), "consensus": ( '' '' '' '' '' '' ), "leapspace": ( '' '' '' '' '' ), "perplexity": ( '' '' '' '' '' '' ), "semanticscholar": ( '' '' '' '' '' ), "pubmed": ( '' '' 'PubMed' '' ), } _DB_MASCOT = { "PubMed@KU": "microscope", "Embase": "medical", "CINAHL Ultimate": "medical", "Cochrane Library": "microscope", "UpToDate": "medical", "IEEE Xplore": "circuit", "ACM Digital Library": "circuit", "ScienceDirect": "microscope", "Scopus": "microscope", "Web of Science": "microscope", "SpringerLink": "book", "ACS Publications": "atom", "APS Journals": "atom", "Knovel": "wrench", "Business Source Complete": "chart", "Emerald Insight": "chart", "Taylor & Francis": "book", "EBSCO": "book", } _BEE = _SVGS["bee"] def _btn(label, url, mascot_key="bee"): icon = _SVGS.get(mascot_key, _SVGS["bee"]) style = ( 'display:inline-flex;align-items:center;gap:7px;' 'padding:8px 14px;border-radius:10px;text-decoration:none;' 'font-size:.8rem;font-weight:600;color:#1a1a2e;' 'background:#fff;border:1.5px solid #d1d5db;' 'transition:box-shadow .2s;width:100%;box-sizing:border-box' ) return '' + icon + label + '' btn_rows = "" for k in range(0, len(platforms), 2): pair = platforms[k:k+2] cells = "" for lbl, url, mascot_key in pair: cells += '' + _btn(lbl, url, mascot_key) + '' if len(pair) == 1: cells += '' btn_rows += "" + cells + "" platforms_html = ( '
' '
' '🤖 Search your query across AI platforms
' '
' 'Your query is pre-loaded. Click any platform to search instantly.
' '' + btn_rows + '
' '
' ) 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 = ( '
' '
' '\U0001f4a1 Boolean search tip
' '
' + bool_label + '
' '' + _escape(bool_q) + '' '
' 'Paste into the Advanced Search box of any database below.
' '
' ) else: bool_html = "" all_lower = ((context.ai_tool_query or "") + " " + (context.topic or "")).lower() def _db_row(name, url, desc, favicon=None): icon = _SVGS.get(_DB_MASCOT.get(name, "bee"), _SVGS["bee"]) return ( '
' + icon + '
' + name + '' ' — ' + desc + '
' '
' ) 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@KU","http://www.ncbi.nlm.nih.gov/pubmed/?otool=iaekustlib","Core biomedical literature"), ("Embase",KU+"https://www.embase.com","Biomedical, drugs, devices, clinical medicine"), ("CINAHL Ultimate",KU+"https://search.ebscohost.com/login.aspx?authtype=ip,uid&profile=ehost&defaultdb=cul","Nursing and allied health"), ("Cochrane Library",KU+"https://www.cochranelibrary.com","Systematic reviews and evidence-based medicine"), ("UpToDate",KU+"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",KU+"https://ieeexplore.ieee.org","Core CS, AI, electronics and systems"), ("ACM Digital Library",KU+"https://dl.acm.org","Computer science journals, proceedings and books"), ("ScienceDirect",KU+"https://www.sciencedirect.com","Full text across CS and applied computing"), ("Scopus",KU+"https://www.scopus.com","Discovery and citation tracking"), ("Web of Science",KU+"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",KU+"https://www.sciencedirect.com","Full text across science, environment and engineering"), ("Scopus",KU+"https://www.scopus.com","Broad discovery and citation tracking"), ("Web of Science",KU+"https://www.webofscience.com","High-impact journal coverage and citation index"), ("SpringerLink",KU+"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",KU+"https://ieeexplore.ieee.org","Electrical, electronics and systems engineering"), ("ScienceDirect",KU+"https://www.sciencedirect.com","Full text across engineering disciplines"), ("Knovel",KU+"https://app.knovel.com","Engineering handbooks and reference data"), ("Scopus",KU+"https://www.scopus.com","Discovery and citation tracking"), ("Web of Science",KU+"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",KU+"https://search.ebscohost.com","Core business and management literature"), ("Emerald Insight",KU+"https://www.emerald.com/insight","Management, strategy and leadership"), ("Scopus",KU+"https://www.scopus.com","Broad discovery and citation tracking"), ("ScienceDirect",KU+"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",KU+"https://www.sciencedirect.com","Full text across physical sciences"), ("ACS Publications",KU+"https://pubs.acs.org","Core chemistry and chemical engineering"), ("APS Journals",KU+"https://journals.aps.org","American Physical Society core physics journals"), ("Scopus",KU+"https://www.scopus.com","Discovery and citation tracking"), ("Web of Science",KU+"https://www.webofscience.com","Citation index"), ] else: db_subject = "General" dbs = [ ("EBSCO",KU+"https://search.ebscohost.com","Multidisciplinary full text"), ("ScienceDirect",KU+"https://www.sciencedirect.com","Science, engineering and medicine"), ("SpringerLink",KU+"https://link.springer.com","Journals and books across all disciplines"), ("Taylor & Francis",KU+"https://www.tandfonline.com","Humanities, social sciences and STEM"), ("Emerald Insight",KU+"https://www.emerald.com/insight","Management and social sciences"), ("Scopus",KU+"https://www.scopus.com","Citation discovery, all disciplines"), ("Web of Science",KU+"https://www.webofscience.com","Citation index, all disciplines"), ] dbs_html_rows = "" for name, url, desc in dbs: dbs_html_rows += _db_row(name, url, desc) db_section = ( '
' '
' '\U0001f4c2 Recommended KU databases \u2014 ' + db_subject + '
' + dbs_html_rows + '
' '' '\U0001f5c4\ufe0f Browse all 50+ KU databases \u2192
' '
' ) return platforms_html + bool_html + db_section