| """ |
| Khazna AI Assistant — natural-language search over KU's Elsevier Pure repository. |
| |
| Environment variables (set as HF Space *Secrets*, never in code): |
| PURE_API_KEY required Pure API key (read-only) |
| OPENAI_API_KEY required planner + answer model (gpt-4o-mini) |
| ANTHROPIC_API_KEY optional enables model="claude" (Claude Haiku 4.5) |
| ADMIN_PASSWORD optional protects /admin/* and /debug/* |
| Variables (non-secret, optional): |
| PURE_BASE_URL default https://khazna.ku.ac.ae/ws/api |
| PURE_PORTAL_URL default https://khazna.ku.ac.ae |
| PURE_PUBLIC_ONLY default true (drop records whose visibility is not public) |
| TOP_K default 8 (records passed to the answer model) |
| ALLOWED_ORIGINS default * comma-separated, e.g. https://ku-ae.libguides.com,https://library.ku.ac.ae |
| RATE_LIMIT_PER_MIN default 12 per client IP on /agent |
| FRONTEND_URL default https://library.ku.ac.ae/khaznabot (the official LibGuides page; "/" redirects there, |
| the Space's own test chat stays at /test). Set to "" to serve the test chat at "/". |
| ANNOUNCEMENT optional banner text shown by the LibGuides frontend (via /config) |
| MAINTENANCE_MODE "true" disables asking in the frontend; MAINTENANCE_MESSAGE sets the text |
| """ |
|
|
| from __future__ import annotations |
|
|
| import os |
| import re |
| import secrets |
| import sqlite3 |
| import time |
| from collections import defaultdict, deque |
| from contextlib import asynccontextmanager |
| from datetime import datetime, timezone |
| from urllib.parse import quote_plus |
|
|
| from fastapi import FastAPI, Header, HTTPException, Request, Response |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.responses import FileResponse, JSONResponse, RedirectResponse |
| from pydantic import BaseModel |
|
|
| import pipeline |
| from pure_client import PureClient, PureError |
|
|
| PURE_BASE_URL = os.environ.get("PURE_BASE_URL", "https://khazna.ku.ac.ae/ws/api") |
| PURE_PORTAL_URL = os.environ.get("PURE_PORTAL_URL", "https://khazna.ku.ac.ae") |
| PUBLIC_ONLY = os.environ.get("PURE_PUBLIC_ONLY", "true").lower() != "false" |
| ACTIVE_PERSONS_ONLY = os.environ.get("PURE_ACTIVE_PERSONS_ONLY", "true").lower() != "false" |
| TOP_K = int(os.environ.get("TOP_K", "8")) |
| ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "") |
| ALLOWED_ORIGINS = [o.strip() for o in os.environ.get("ALLOWED_ORIGINS", "*").split(",") if o.strip()] |
| FRONTEND_URL = os.environ.get("FRONTEND_URL", "https://library.ku.ac.ae/khaznabot").strip() |
| RATE_LIMIT = int(os.environ.get("RATE_LIMIT_PER_MIN", "12")) |
| DB_PATH = "/data/khazna_analytics.db" if os.path.isdir("/data") else "khazna_analytics.db" |
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| SCOPES = ("all", "publications", "theses", "persons", "organisations", "projects", "equipment", "collaborations", "datasets") |
| SCOPE_PORTAL_PATH = {"publications": "publications", "theses": "studentTheses", "persons": "persons", |
| "organisations": "organisations", "projects": "projects", "equipment": "equipments", |
| "collaborations": "publications", "datasets": "datasets"} |
| INTENT_SCOPE = {"thesis_search": "theses", "project_search": "projects", "equipment_search": "equipment", |
| "collab_search": "collaborations", "topic_search": "publications", "author_search": "publications", "count": "publications", |
| "expert_search": "persons", "org_search": "organisations", "dataset_search": "datasets"} |
|
|
| pure: PureClient | None = None |
|
|
|
|
| |
| def init_db(): |
| with sqlite3.connect(DB_PATH) as c: |
| c.execute("""CREATE TABLE IF NOT EXISTS queries ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, question TEXT, intent TEXT, |
| keywords TEXT, results INTEGER, model TEXT, seconds REAL, error TEXT)""") |
| c.execute("""CREATE TABLE IF NOT EXISTS feedback ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, question TEXT, rating TEXT, comment TEXT)""") |
| cols = {r[1] for r in c.execute("PRAGMA table_info(queries)")} |
| if "scope" not in cols: |
| c.execute("ALTER TABLE queries ADD COLUMN scope TEXT") |
| for col in ("origin", "request_id", "session_id"): |
| if col not in cols: |
| c.execute(f"ALTER TABLE queries ADD COLUMN {col} TEXT") |
| fcols = {r[1] for r in c.execute("PRAGMA table_info(feedback)")} |
| for col in ("request_id", "session_id"): |
| if col not in fcols: |
| c.execute(f"ALTER TABLE feedback ADD COLUMN {col} TEXT") |
|
|
|
|
| def log_query(question, intent, keywords, results, model, seconds, error=None, scope="all", origin="", |
| request_id="", session_id=""): |
| try: |
| with sqlite3.connect(DB_PATH) as c: |
| c.execute( |
| "INSERT INTO queries (ts,question,intent,keywords,results,model,seconds,error,scope,origin,request_id,session_id) " |
| "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", |
| (datetime.now(timezone.utc).isoformat(), question, intent, " | ".join(keywords or []), results, model, |
| round(seconds, 2), error, scope, origin, request_id, session_id), |
| ) |
| except Exception as e: |
| print(f"[log] {e}") |
|
|
|
|
| |
| _hits: dict[str, deque] = defaultdict(deque) |
|
|
|
|
| def _client_ip(request: Request) -> str: |
| fwd = request.headers.get("x-forwarded-for", "") |
| return fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else "?") |
|
|
|
|
| def _rate_limited(ip: str) -> bool: |
| now, q = time.time(), _hits[ip] |
| while q and now - q[0] > 60: |
| q.popleft() |
| if len(q) >= RATE_LIMIT: |
| return True |
| q.append(now) |
| if len(_hits) > 5000: |
| _hits.clear() |
| return False |
|
|
|
|
| |
| @asynccontextmanager |
| async def lifespan(app: FastAPI): |
| global pure |
| init_db() |
| key = os.environ.get("PURE_API_KEY", "") |
| if not key: |
| print("WARNING: PURE_API_KEY not set — searches will fail.") |
| pure = PureClient(PURE_BASE_URL, key, PURE_PORTAL_URL, public_only=PUBLIC_ONLY, active_persons_only=ACTIVE_PERSONS_ONLY) |
| yield |
| await pure.aclose() |
|
|
|
|
| app = FastAPI(title="Khazna AI Assistant", lifespan=lifespan) |
| app.add_middleware(CORSMiddleware, allow_origins=ALLOWED_ORIGINS, allow_methods=["GET", "POST"], allow_headers=["*"], |
| expose_headers=["X-Request-ID", "Retry-After"]) |
|
|
|
|
| class ChatMessage(BaseModel): |
| role: str |
| content: str |
|
|
|
|
| class AgentRequest(BaseModel): |
| question: str |
| model: str = "gpt" |
| history: list[ChatMessage] = [] |
| top_k: int | None = None |
| scope: str = "all" |
| person_uuid: str | None = None |
| person_name: str | None = None |
|
|
|
|
| class FeedbackRequest(BaseModel): |
| |
| request_id: str = "" |
| feedback: str = "" |
| session_id: str = "" |
| question: str = "" |
| rating: str = "" |
| comment: str = "" |
|
|
|
|
| def _require_admin(pw: str | None): |
| if not ADMIN_PASSWORD or pw != ADMIN_PASSWORD: |
| raise HTTPException(401, "Admin password required (X-Admin-Password header).") |
|
|
|
|
| def _public_result(o: dict, n: int) -> dict: |
| kind = o.get("kind", "publication") |
| if kind == "person": |
| keep = ("kind", "uuid", "name", "job_titles", "organisations", "keywords", "orcid", "portal_url", "evidence", "matching_outputs", |
| "supervised", "supervised_theses", "choose") |
| elif kind == "organisation": |
| keep = ("kind", "uuid", "name", "type", "parents", "portal_url") |
| elif kind == "collaborator": |
| keep = ("kind", "uuid", "name", "type", "country", "joint_outputs", "examples", "portal_url") |
| elif kind == "project": |
| keep = ("kind", "uuid", "title", "subtitle", "type", "year", "period", "status", "venue", "funders", "keywords", "portal_url") |
| elif kind == "equipment": |
| keep = ("kind", "uuid", "title", "type", "category", "venue", "location", "keywords", "portal_url") |
| else: |
| keep = ("kind", "uuid", "title", "subtitle", "type", "year", "date", "authors", "ku_authors", "venue", |
| "doi", "doi_url", "open_access", "file_available", "portal_url", "keywords") |
| out = {k: o.get(k) for k in keep} |
| if kind == "thesis": |
| out["supervisors"] = [{"name": x.get("name"), "uuid": x.get("uuid")} for x in (o.get("supervisors") or [])] |
| if kind == "project": |
| out["participants"] = [{"name": x.get("name"), "role": x.get("role")} for x in (o.get("participants") or [])] |
| if kind == "equipment": |
| out["contacts"] = [{"name": x.get("name")} for x in (o.get("contacts") or [])] |
| out["n"] = n |
| if kind in ("publication", "dataset", "thesis", "project", "equipment"): |
| out["abstract"] = (o.get("abstract") or "")[:400] |
| return out |
|
|
|
|
| def _reply(t0: float, **kw) -> dict: |
| kw.setdefault("search_results", []) |
| kw["response_time"] = round(time.time() - t0, 2) |
| return kw |
|
|
|
|
| _SESSION_RE = re.compile(r"^[A-Za-z0-9_-]{8,80}$") |
| FIRST_PAGE = 12 |
|
|
|
|
| from pure_client import PERSONS_PORTAL_PARAMS |
|
|
|
|
| def _portal_url(path: str, term: str, search_by: str | None = None) -> str: |
| """Khazna portal search link. People searches use the portal's own parameters with |
| community=false, so external 'Pure Community' profiles are not listed.""" |
| url = f"{PURE_PORTAL_URL}/en/{path}/?search={quote_plus(term)}" |
| return url + PERSONS_PORTAL_PARAMS if path == "persons" else url |
|
|
|
|
| def _followups(plan, retrieval: dict | None) -> list[dict]: |
| """Next-step suggestions the frontend renders as chips: {label, question, scope}.""" |
| kw = (plan.keyword_queries[0] if plan.keyword_queries else "").strip() |
| year = datetime.now().year |
| out: list[dict] = [] |
| if plan.intent == "author_search" and retrieval and retrieval.get("person"): |
| name = retrieval["person"]["name"] |
| out.append({"label": f"Latest by {name}", "question": f"Latest publications by {name}", "scope": "publications"}) |
| top_kw = (retrieval["person"].get("keywords") or [kw])[0] if (retrieval["person"].get("keywords") or kw) else "" |
| if top_kw: |
| out.append({"label": f"Others working on {top_kw}", "question": f"Who else at KU works on {top_kw}?", "scope": "persons"}) |
| return out |
| if not kw: |
| return out |
| if plan.intent in ("topic_search", "count"): |
| out += [{"label": "Who at KU works on this?", "question": f"Who at KU works on {kw}?", "scope": "persons"}, |
| {"label": "KU theses on this", "question": f"KU theses on {kw}", "scope": "theses"}, |
| {"label": "Related projects", "question": f"KU research projects on {kw}", "scope": "projects"}, |
| {"label": "Who does KU collaborate with?", "question": f"Which organisations does KU collaborate with on {kw}?", "scope": "collaborations"}] |
| if not plan.year_from: |
| out.append({"label": f"Only since {year - 3}", "question": f"KU publications on {kw} since {year - 3}", "scope": "publications"}) |
| elif plan.intent == "project_search": |
| out += [{"label": "Who works on this?", "question": f"Who at KU works on {kw}?", "scope": "persons"}, |
| {"label": "Publications from this area", "question": f"KU publications on {kw}", "scope": "publications"}] |
| elif plan.intent == "equipment_search": |
| out += [{"label": "Which departments use this?", "question": f"Which KU departments or centres work on {kw}?", "scope": "organisations"}, |
| {"label": "Experts in this area", "question": f"Who at KU works on {kw}?", "scope": "persons"}] |
| elif plan.intent == "collab_search": |
| out += [{"label": "The joint publications", "question": f"KU publications on {kw}", "scope": "publications"}, |
| {"label": "KU experts in this area", "question": f"Who at KU works on {kw}?", "scope": "persons"}] |
| elif plan.intent == "thesis_search": |
| out += [{"label": "Who supervises in this area?", "question": f"Who at KU supervises theses on {kw}?", "scope": "persons"}, |
| {"label": "Related publications", "question": f"KU publications on {kw}", "scope": "publications"}] |
| elif plan.intent == "expert_search": |
| out += [{"label": "Show the publications", "question": f"KU publications on {kw}", "scope": "publications"}, |
| {"label": "Which departments or centres?", "question": f"Which KU departments or centres work on {kw}?", "scope": "organisations"}] |
| elif plan.intent == "org_search": |
| out += [{"label": "Experts in this area", "question": f"Who at KU works on {kw}?", "scope": "persons"}, |
| {"label": "Publications on this", "question": f"KU publications on {kw}", "scope": "publications"}] |
| elif plan.intent == "dataset_search": |
| out += [{"label": "Publications on this", "question": f"KU publications on {kw}", "scope": "publications"}, |
| {"label": "Who works on this?", "question": f"Who at KU works on {kw}?", "scope": "persons"}] |
| return out[:4] |
|
|
|
|
| |
| @app.get("/") |
| def ui(): |
| |
| if FRONTEND_URL.startswith("https://"): |
| return RedirectResponse(FRONTEND_URL, status_code=302) |
| return FileResponse(os.path.join(HERE, "static", "index.html")) |
|
|
|
|
| @app.get("/test") |
| def test_ui(): |
| """The Space's own chat page (widget.js), for testing backend changes before they reach LibGuides.""" |
| return FileResponse(os.path.join(HERE, "static", "index.html")) |
|
|
|
|
| @app.get("/widget.js") |
| def widget_js(): |
| |
| return FileResponse(os.path.join(HERE, "static", "widget.js"), media_type="application/javascript", |
| headers={"Cache-Control": "public, max-age=300"}) |
|
|
|
|
| @app.get("/config") |
| def public_config(): |
| return { |
| "announcement": os.environ.get("ANNOUNCEMENT", ""), |
| "maintenance_mode": os.environ.get("MAINTENANCE_MODE", "false").lower() == "true", |
| "maintenance_message": os.environ.get("MAINTENANCE_MESSAGE", "The Khazna assistant is temporarily unavailable for maintenance."), |
| "scopes": list(SCOPES), |
| } |
|
|
|
|
| @app.get("/health") |
| def health(): |
| return { |
| "status": "ok", |
| "pure_base_url": PURE_BASE_URL, |
| "pure_key_configured": bool(os.environ.get("PURE_API_KEY")), |
| "frontend_url": FRONTEND_URL, |
| "portal_people_status": getattr(pure, "portal_status", "not tried") if pure else "not started", |
| "public_only": PUBLIC_ONLY, |
| "scopes": SCOPES, |
| "models": {"gpt": bool(os.environ.get("OPENAI_API_KEY")), "claude": bool(os.environ.get("ANTHROPIC_API_KEY"))}, |
| } |
|
|
|
|
| @app.post("/agent") |
| async def agent(req: AgentRequest, request: Request, response: Response): |
| t0 = time.time() |
| request_id = secrets.token_hex(12) |
| response.headers["X-Request-ID"] = request_id |
| session_id = request.headers.get("x-khazna-session", "") |
| session_id = session_id if _SESSION_RE.match(session_id) else "" |
| if _rate_limited(_client_ip(request)): |
| return JSONResponse(status_code=429, headers={"Retry-After": "60", "X-Request-ID": request_id}, |
| content=_reply(t0, intent="rate_limited", request_id=request_id, |
| answer="You're sending questions very quickly — please wait a minute and try again.")) |
| if os.environ.get("MAINTENANCE_MODE", "false").lower() == "true": |
| return JSONResponse(status_code=503, headers={"X-Request-ID": request_id}, |
| content=_reply(t0, intent="maintenance", request_id=request_id, |
| answer=os.environ.get("MAINTENANCE_MESSAGE", "The Khazna assistant is temporarily unavailable."))) |
| q = req.question.strip()[:1000] |
| if not q: |
| return _reply(t0, intent="empty", answer="Please type a question.") |
| history = [m.model_dump() for m in req.history][-8:] |
| model = req.model if req.model in ("gpt", "claude") else "gpt" |
| scope = req.scope if req.scope in SCOPES else "all" |
| top_k = max(1, min(req.top_k or TOP_K, 15)) |
| origin = request.headers.get("origin", "") |
|
|
| plan = await pipeline.plan_query(q, history, model, datetime.now().year, scope) |
| base = {"intent": plan.intent, "plan": plan.model_dump(), "model_used": model, "scope": scope, "request_id": request_id} |
| logx = {"scope": scope, "origin": origin, "request_id": request_id, "session_id": session_id} |
|
|
| if plan.intent == "social": |
| return _reply(t0, **base, answer=pipeline.SOCIAL_ANSWER) |
| if plan.intent == "out_of_scope": |
| return _reply(t0, **base, answer=pipeline.OUT_OF_SCOPE_ANSWER) |
| if plan.intent == "repository_help": |
| ans = await pipeline.answer_repository_help(q, model) |
| log_query(q, plan.intent, [], 0, model, time.time() - t0, **logx) |
| return _reply(t0, **base, answer=ans) |
|
|
| person_uuid = req.person_uuid if req.person_uuid and re.fullmatch(r"[A-Za-z0-9-]{1,64}", req.person_uuid) else None |
| try: |
| person_name = (req.person_name or "").strip()[:120] or None |
| retrieval = await pipeline.retrieve(pure, plan, person_uuid=person_uuid, person_name=person_name) |
| except PureError as e: |
| log_query(q, plan.intent, plan.keyword_queries, 0, model, time.time() - t0, str(e), **logx) |
| if e.status == 403 and plan.intent == "dataset_search": |
| msg = ("Dataset search isn't switched on yet: the Khazna API key needs the /data-sets/* endpoint. " |
| "Meanwhile you can browse datasets at https://khazna.ku.ac.ae/en/datasets/.") |
| else: |
| msg = ("Sorry — I couldn't reach Khazna just now. Please try again in a moment, " |
| "or search directly at https://khazna.ku.ac.ae/.") |
| return _reply(t0, **base, answer=msg, error=str(e)) |
|
|
| results = retrieval["results"] |
| if retrieval.get("clarify"): |
| c = retrieval["clarify"] |
| more = (" Use **Show more** below the list to see everyone, or add a first name or department to narrow it down." |
| if c["shown"] > FIRST_PAGE else "") |
| if c["count"] > c["shown"]: |
| more += f" (Showing the first {c['shown']} of {c['count']}.)" |
| answer = (f"I found **{c['count']} KU researchers** whose name matches “{c['name']}”. Which one do you mean?\n" |
| f"Choose **Explore this researcher** on the right person below, or add their full name or department to your question.{more}") |
| log_query(q, "author_clarify", [c["name"]], len(results), model, time.time() - t0, **logx) |
| return _reply( |
| t0, **{**base, "intent": "author_clarify"}, |
| answer=answer, clarify=True, original_question=q, |
| search_results=[_public_result(o, i) for i, o in enumerate(results[:FIRST_PAGE], 1)], |
| more_results=[_public_result(o, i) for i, o in enumerate(results[FIRST_PAGE:], FIRST_PAGE + 1)], |
| total_found=c["count"], after_filters=len(results), |
| trace=[{k: v for k, v in t.items() if k != "error"} for t in retrieval["trace"]], |
| portal_search_url=_portal_url("persons", c["name"], "PartOfNameOrTitle"), |
| follow_up_suggestions=[], |
| ) |
| if not results and retrieval.get("no_person"): |
| name = retrieval["no_person"] |
| answer = (f"I couldn't find a current KU researcher named “{name}” in Khazna. " |
| "Check the spelling or try their full name. Former staff and external co-authors are not listed.") |
| log_query(q, "author_not_found", [name], 0, model, time.time() - t0, **logx) |
| return _reply(t0, **base, answer=answer, trace=[{k: v for k, v in t.items() if k != "error"} for t in retrieval["trace"]], |
| portal_search_url=_portal_url("persons", name), follow_up_suggestions=[]) |
| if not results and retrieval.get("person"): |
| pname = retrieval["person"].get("name", "this researcher") |
| answer = (f"Here is **{pname}**'s Khazna profile. I couldn't find matching publications through search" |
| + (" for that topic or filter" if (plan.keyword_queries and plan.intent == "author_search") else "") |
| + "; open the profile for their full list of outputs.") |
| elif not results: |
| filt = bool(plan.year_from or plan.year_to or plan.output_types or plan.open_access_only) |
| answer = ("I couldn't find Khazna records matching that" + (" with the filters you asked for" if filt else "") |
| + ". Try broader or different keywords" + (", or remove the year/type filter." if filt else ".")) |
| else: |
| try: |
| answer = await pipeline.synthesize(q, plan, retrieval, history, model, top_k) |
| except Exception as e: |
| print(f"[synth] {e}") |
| answer = f"I found {len(results)} matching Khazna records; here are the most relevant." |
|
|
| elapsed = time.time() - t0 |
| log_query(q, plan.intent, plan.keyword_queries, len(results), model, elapsed, **logx) |
| portal_path = SCOPE_PORTAL_PATH.get(INTENT_SCOPE.get(plan.intent, "publications"), "publications") |
| search_term = plan.author_name if plan.intent == "author_search" and plan.author_name else (plan.keyword_queries[0] if plan.keyword_queries else q) |
| person = retrieval.get("person") |
| return _reply( |
| t0, **base, |
| answer=answer, |
| search_results=[_public_result(o, i) for i, o in enumerate(results[:top_k], 1)], |
| more_results=[_public_result(o, i) for i, o in enumerate(results[top_k:top_k + 12], top_k + 1)], |
| total_found=retrieval["total"], |
| after_filters=len(results), |
| person=_public_result(person, 0) if person else None, |
| database_query=" | ".join(t.get("query", "") for t in retrieval["trace"] if t.get("query")), |
| trace=[{k: v for k, v in t.items() if k != "error"} for t in retrieval["trace"]], |
| portal_search_url=_portal_url(portal_path, search_term, |
| "PartOfNameOrTitle" if plan.intent == "author_search" and plan.author_name |
| else "RelatedConcepts" if plan.intent == "expert_search" else None), |
| follow_up_suggestions=_followups(plan, retrieval) if results else [], |
| ) |
|
|
|
|
| @app.post("/feedback") |
| def feedback(req: FeedbackRequest): |
| rating = (req.feedback or req.rating or "")[:10] |
| if rating not in ("up", "down"): |
| raise HTTPException(422, "feedback must be 'up' or 'down'") |
| question = req.question |
| with sqlite3.connect(DB_PATH) as c: |
| if req.request_id and not question: |
| row = c.execute("SELECT question FROM queries WHERE request_id=?", (req.request_id[:40],)).fetchone() |
| question = row[0] if row else "" |
| c.execute("INSERT INTO feedback (ts,question,rating,comment,request_id,session_id) VALUES (?,?,?,?,?,?)", |
| (datetime.now(timezone.utc).isoformat(), question[:1000], rating, req.comment[:1000], |
| req.request_id[:40], req.session_id[:80])) |
| return {"status": "ok"} |
|
|
|
|
| @app.get("/admin/stats") |
| def admin_stats(x_admin_password: str | None = Header(default=None)): |
| _require_admin(x_admin_password) |
| with sqlite3.connect(DB_PATH) as c: |
| c.row_factory = sqlite3.Row |
| total = c.execute("SELECT COUNT(*) n, AVG(seconds) s FROM queries").fetchone() |
| by_intent = c.execute("SELECT intent, COUNT(*) n FROM queries GROUP BY intent ORDER BY n DESC").fetchall() |
| by_origin = c.execute("SELECT origin, COUNT(*) n FROM queries GROUP BY origin ORDER BY n DESC").fetchall() |
| zero = c.execute("SELECT question, intent, keywords, ts FROM queries WHERE results=0 AND intent NOT IN " |
| "('social','repository_help','out_of_scope') ORDER BY id DESC LIMIT 30").fetchall() |
| recent = c.execute("SELECT ts, question, intent, scope, keywords, results, seconds, error FROM queries ORDER BY id DESC LIMIT 50").fetchall() |
| fb = c.execute("SELECT rating, COUNT(*) n FROM feedback GROUP BY rating").fetchall() |
| return {"queries": total["n"], "avg_seconds": round(total["s"] or 0, 2), |
| "by_intent": {r["intent"]: r["n"] for r in by_intent}, |
| "by_origin": {(r["origin"] or "direct"): r["n"] for r in by_origin}, |
| "zero_result_queries": [dict(r) for r in zero], "recent": [dict(r) for r in recent], |
| "feedback": {r["rating"]: r["n"] for r in fb}} |
|
|
|
|
| DEBUG_ENDPOINTS = {"research-outputs": "normalize_research_output", "persons": "normalize_person", |
| "organizations": "normalize_organization", "data-sets": "normalize_dataset", |
| "student-theses": "normalize_student_thesis", "projects": "normalize_project", |
| "equipment": "normalize_equipment", "external-organizations": "normalize_external_organization"} |
|
|
|
|
| @app.get("/debug/persons") |
| async def debug_persons(q: str = "Kumar", x_admin_password: str | None = Header(default=None)): |
| """Why each person returned by Pure for a name is kept or dropped (visibility, staff appointments).""" |
| _require_admin(x_admin_password) |
| import pure_client |
| from datetime import date as _date |
| raw = await pure.request("POST", "/persons/search", json={"searchString": q, "size": 50, "offset": 0}) |
| items = raw.get("items") or [] |
| field_returned = any("staffOrganizationAssociations" in i for i in items) |
| kept_uuids = {p["uuid"] for p in await pure.search_persons(q, size=50)} |
| matched, _ = pipeline.match_people(q, [pure_client.normalize_person(i, PURE_PORTAL_URL) for i in items]) |
| matched_uuids = {p["uuid"] for p in matched} |
| today = _date.today().isoformat() |
| rows = [] |
| for i in items: |
| vis = i.get("visibility") |
| status = pure_client.staff_status(i) |
| assocs = [] |
| for a in i.get("staffOrganizationAssociations") or []: |
| per = a.get("period") or {} |
| org = a.get("organization") or {} |
| assocs.append({ |
| "organisation": pure_client._loc(org.get("name")) or await pure.organization_name(org.get("uuid", "")) if org.get("uuid") else "", |
| "job_title": pure_client._loc(a.get("jobTitle")), |
| "start": pure_client._date_parts(per.get("startDate"))[1], "end": pure_client._date_parts(per.get("endDate"))[1], |
| "current": pure_client._association_active(a, today)}) |
| reason = ("not public" if not pure_client.is_public(i) else |
| "no current staff appointment" if status is False else |
| "name does not match" if i.get("uuid") not in matched_uuids else "kept") |
| rows.append({"name": pure_client._person_name(i.get("name")), "uuid": i.get("uuid"), |
| "visibility": vis.get("key") if isinstance(vis, dict) else vis, |
| "staff_appointments": assocs, |
| "other_association_fields": [k for k in i if k.endswith("Associations") and k != "staffOrganizationAssociations" and i.get(k)], |
| "kept_by_filter": i.get("uuid") in kept_uuids, "result": reason}) |
| portal = await pure.portal_person_search(q) |
| merged = pipeline.merge_portal_people(portal or [], await pure.search_persons(q, size=50)) if portal is not None else None |
| shown_by_bot, _ = pipeline.match_people(q, merged if merged is not None else []) |
| return {"query": q, "portal_reachable": portal is not None, |
| "portal_people": [{"name": p["name"], "portal_url": p["portal_url"], "api_uuid": p.get("uuid") or None, |
| "shown_by_bot": p in shown_by_bot} for p in (merged or [])], |
| "api_count": raw.get("count"), "api_returned": len(items), "staff_field_returned": field_returned, |
| "api_people": rows} |
|
|
|
|
| @app.get("/debug/pure") |
| async def debug_pure(q: str = "machine learning", endpoint: str = "research-outputs", |
| x_admin_password: str | None = Header(default=None)): |
| """Raw vs normalised first hit for any endpoint — verifies field mappings against your Pure version.""" |
| _require_admin(x_admin_password) |
| if endpoint not in DEBUG_ENDPOINTS: |
| raise HTTPException(400, f"endpoint must be one of {list(DEBUG_ENDPOINTS)}") |
| import pure_client |
| try: |
| raw = await pure.request("POST", f"/{endpoint}/search", json={"searchString": q, "size": 1, "offset": 0}) |
| except PureError as e: |
| return {"endpoint": endpoint, "error": str(e), "status": e.status} |
| items = raw.get("items") or [] |
| norm = getattr(pure_client, DEBUG_ENDPOINTS[endpoint]) |
| return { |
| "endpoint": endpoint, |
| "count": raw.get("count"), |
| "raw_first_item": items[0] if items else None, |
| "is_public": pure_client.is_public(items[0]) if items else None, |
| "normalized": norm(items[0], PURE_PORTAL_URL) if items else None, |
| } |
|
|
