| """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: |
| |
| 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) |
|
|
