File size: 5,746 Bytes
c597e9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """Server-side PRIMO records for chat; never substitute another provider's hits."""
import asyncio
import html
import logging
from urllib.parse import urlsplit
from src.config import get_settings
from src.agentcore.models import SearchContextPayload
logger = logging.getLogger(__name__)
def _text(value) -> str:
if isinstance(value, list):
return "; ".join(str(item) for item in value if item)
return str(value or "")
def _safe_record_url(value) -> str:
value = _text(value).strip()
try:
parts = urlsplit(value)
if (parts.scheme in {"http", "https"} and parts.hostname
and not parts.username and not parts.password
and not any(ord(c) < 32 for c in value)):
return value
except ValueError:
pass
return ""
def primo_filter_note(context: SearchContextPayload) -> str:
"""Visible applied dates, material and ordering, including the 'recent' default."""
filters = [{"books": "Books", "articles": "Articles", "both": "Books and articles"}[context.resource_type]]
if context.year_from and context.year_to:
filters.append(f"publication years {context.year_from}–{context.year_to} (inclusive)")
elif context.year_from:
filters.append(f"published from {context.year_from}")
elif context.year_to:
filters.append(f"published through {context.year_to}")
if context.peer_reviewed:
filters.append("peer reviewed")
if context.open_access:
filters.append("open access")
filters.append({"rank": "by relevance", "date_d": "newest first", "date_a": "oldest first"}[context.sort])
return "; ".join(filters) + "."
def render_primo_results(payload: dict, context: SearchContextPayload) -> tuple[list[dict], str]:
"""Escape provider metadata; a result link never implies local availability."""
records = payload.get("results") or []
records = [record for record in records if isinstance(record, dict)][:10] if isinstance(records, list) else []
failed = bool(payload.get("error")) or payload.get("status", "ok") != "ok"
if failed:
records = []
esc = html.escape
parts = ['<div class="libbee-primo-results" aria-label="PRIMO search results">']
parts.append('<h3 class="libbee-primo-results-title">Results from KU Library PRIMO</h3>')
parts.append(f'<p class="libbee-primo-results-note">{esc(primo_filter_note(context))}</p>')
if failed:
parts.append('<p class="libbee-primo-results-note">PRIMO results could not load right now. Open the search link above to continue with these filters.</p>')
elif not records:
parts.append('<p class="libbee-primo-results-note">PRIMO returned no results for this search. Open the link above to broaden the keywords or adjust the filters.</p>')
else:
total = payload.get("total")
count = len(records)
count_text = f"Showing {count} result{'s' if count != 1 else ''}"
if isinstance(total, int) and total >= count:
count_text += f" of {total:,}"
parts.append(f'<p class="libbee-primo-results-note">{count_text}.</p>')
parts.append('<ol class="libbee-primo-results-list">')
for record in records:
title = esc(_text(record.get("title")) or "Untitled record")
url = _safe_record_url(record.get("primo_url")) or _safe_record_url(record.get("link"))
title_html = (f'<a class="libbee-primo-result-title" href="{esc(url, quote=True)}" target="_blank" rel="noopener noreferrer">{title}</a>'
if url else f'<strong class="libbee-primo-result-title">{title}</strong>')
metadata = [_text(record.get(key)) for key in ("creator", "date", "type")]
metadata = " · ".join(item for item in metadata if item)
parts.append(f'<li class="libbee-primo-result">{title_html}')
if metadata:
parts.append(f'<p class="libbee-primo-result-meta">{esc(metadata)}</p>')
parts.append('</li>')
parts.append('</ol>')
if context.resource_type == "books":
parts.append('<p class="libbee-primo-results-note">The Books filter is already applied. In PRIMO, refine the format or online availability for e-books, and open a record to check editions and access. Book records are not necessarily designated textbooks.</p>')
parts.append('</div>')
return records, "".join(parts)
async def fetch_primo_panel(context: SearchContextPayload) -> tuple[list[dict], str]:
settings = get_settings()
api_key = getattr(settings, "primo_api_key", "")
payload = {"status": "not_configured", "results": []}
if api_key:
try:
# Lazy import keeps offline/query-only use independent of the API framework.
from src.api.search import SearchRequest, search_primo
request = SearchRequest(
query=context.display_topic or context.topic,
boolean_query=context.primo_boolean_query,
source="primo", limit=10, resource_type=context.resource_type,
peer_reviewed=context.peer_reviewed, open_access=context.open_access,
year_from=context.year_from, year_to=context.year_to,
query_field=context.query_field, sort=context.sort,
)
payload = await asyncio.wait_for(
search_primo(request, api_key, region=getattr(settings, "primo_api_region", "auto")),
timeout=6.5,
)
except Exception as exc:
logger.warning("PRIMO chat results unavailable: %s", type(exc).__name__)
payload = {"status": "unavailable", "results": []}
return render_primo_results(payload, context)
|