from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope from govoplan_core.audit.logging import audit_from_principal from govoplan_core.core.search import CAPABILITY_SEARCH_INDEX_WRITER from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.search import SearchQuery from govoplan_core.db.session import get_session from govoplan_search.backend.manifest import ADMIN_SCOPE, READ_SCOPE from govoplan_search.backend.schemas import ( SearchChangeDispatchResponse, SearchDiagnosticsResponse, SearchIndexStateResponse, SearchModuleReconcileResponse, SearchProviderListResponse, SearchProviderResponse, SearchRebuildResponse, SearchResourceTypeResponse, SearchResponse, SearchResultResponse, ) from govoplan_search.backend.service import ( SearchIndexService, aggregate_search_page, ) router = APIRouter(prefix="/search", tags=["search"]) def _registry(request: Request) -> PlatformRegistry: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Search registry is not available.", ) return registry def _require_read(principal: ApiPrincipal) -> None: if not has_scope(principal, READ_SCOPE): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {READ_SCOPE}", ) def _require_admin(principal: ApiPrincipal) -> None: if not has_scope(principal, ADMIN_SCOPE): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {ADMIN_SCOPE}", ) def _service(registry: PlatformRegistry) -> SearchIndexService: capability = registry.capability( CAPABILITY_SEARCH_INDEX_WRITER ) if not isinstance(capability, SearchIndexService): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Search index service is not available.", ) return capability def _search_resource_catalogue( registry: PlatformRegistry, ) -> list[SearchResourceTypeResponse]: resources = { ( descriptor.module_id, descriptor.resource_type, descriptor.provider_id, ): SearchResourceTypeResponse( provider_id=descriptor.provider_id, module_id=descriptor.module_id, resource_type=descriptor.resource_type, label=descriptor.label, order=registered.registration.order, ) for registered, source in registry.search_sources() for descriptor in source.resource_types() } return [ resources[key] for key in sorted( resources, key=lambda item: ( resources[item].order, resources[item].label.casefold(), item, ), ) ] @router.get("", response_model=SearchResponse) def api_search( request: Request, q: str = Query(default="", max_length=500), module: list[str] = Query(default=[]), resource_type: list[str] = Query(default=[]), context_kind: str = Query(default="global", pattern="^(global|module|resource)$"), context_id: str | None = Query(default=None, max_length=255), limit: int = Query(default=25, ge=1, le=100), offset: int = Query(default=0, ge=0, le=10_000), cursor: str | None = Query(default=None, max_length=2000), language: str = Query( default="simple", max_length=32, pattern=r"^[A-Za-z_]+$", ), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchResponse: _require_read(principal) registry = _registry(request) try: query = SearchQuery( text=q, tenant_id=principal.tenant_id, module_ids=tuple(dict.fromkeys(module)), resource_types=tuple(dict.fromkeys(resource_type)), context_kind=context_kind, # type: ignore[arg-type] context_id=context_id, limit=limit, offset=offset, cursor=cursor, language=language.casefold(), ) page = aggregate_search_page( registry, session, principal, query=query, ) except ValueError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc), ) from exc return SearchResponse( query=query.text, results=[ SearchResultResponse.model_validate( { "provider_id": result.provider_id, "module_id": result.module_id, "resource_type": result.resource_type, "resource_id": result.resource_id, "title": result.title, "summary": result.summary, "url": result.url, "score": result.score, "highlights": list(result.highlights), "breadcrumbs": list(result.breadcrumbs), "external_reference": ( result.external_reference.to_dict() if result.external_reference is not None else None ), "metadata": dict(result.metadata), "source_revision": result.source_revision, "provenance": dict(result.provenance), } ) for result in page.results ], diagnostics=list(page.diagnostics), next_cursor=page.next_cursor, has_more=page.next_cursor is not None, ) @router.get("/providers", response_model=SearchProviderListResponse) def api_search_providers( request: Request, principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchProviderListResponse: _require_read(principal) registry = _registry(request) registrations = registry.search_provider_registrations() return SearchProviderListResponse( providers=[ SearchProviderResponse( id=item.registration.id, module_id=item.module_id, resource_types=list(item.registration.resource_types), order=item.registration.order, ) for item in registrations ], resources=_search_resource_catalogue(registry), ) @router.get( "/admin/diagnostics", response_model=SearchDiagnosticsResponse, ) def api_search_diagnostics( request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchDiagnosticsResponse: _require_admin(principal) registry = _registry(request) return SearchDiagnosticsResponse.model_validate( _service(registry).diagnostics(session, principal) ) @router.post( "/admin/reconcile-modules", response_model=SearchModuleReconcileResponse, ) def api_reconcile_search_modules( request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchModuleReconcileResponse: _require_admin(principal) result = _service(_registry(request)).reconcile_active_modules( session, tenant_id=principal.tenant_id, ) audit_from_principal( session, principal, action="search.modules.reconciled", scope="tenant", object_type="search_index", details=result, ) session.commit() return SearchModuleReconcileResponse(**result) @router.post( "/admin/changes/process", response_model=SearchChangeDispatchResponse, ) def api_process_search_changes( request: Request, limit: int = Query(default=100, ge=1, le=500), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchChangeDispatchResponse: _require_admin(principal) result = _service(_registry(request)).process_changes( session, limit=limit, tenant_id=principal.tenant_id, ) audit_from_principal( session, principal, action="search.changes.processed", scope="tenant", object_type="search_index", details=result, ) session.commit() return SearchChangeDispatchResponse(**result) @router.post( "/admin/rebuilds/{provider_id}/{resource_type}/start", response_model=SearchRebuildResponse, ) def api_start_search_rebuild( provider_id: str, resource_type: str, request: Request, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchRebuildResponse: _require_admin(principal) try: state = _service(_registry(request)).start_rebuild( session, principal, provider_id=provider_id, resource_type=resource_type, ) except ValueError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc), ) from exc audit_from_principal( session, principal, action="search.rebuild.started", scope="tenant", object_type="search_index", object_id=state.id, details={ "provider_id": provider_id, "resource_type": resource_type, "rebuild_id": state.rebuild_id, }, ) session.commit() return SearchRebuildResponse( state=_search_state_response(state) ) @router.post( "/admin/rebuilds/{provider_id}/{resource_type}/continue", response_model=SearchRebuildResponse, ) def api_continue_search_rebuild( provider_id: str, resource_type: str, request: Request, limit: int = Query(default=100, ge=1, le=500), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> SearchRebuildResponse: _require_admin(principal) service = _service(_registry(request)) try: state = service.continue_rebuild( session, principal, provider_id=provider_id, resource_type=resource_type, limit=limit, ) except ValueError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc), ) from exc audit_from_principal( session, principal, action="search.rebuild.continued", scope="tenant", object_type="search_index", object_id=state.id, details={ "provider_id": provider_id, "resource_type": resource_type, "status": state.status, "indexed_documents": state.indexed_documents, }, ) session.commit() return SearchRebuildResponse( state=_search_state_response(state) ) def _search_state_response(state: object) -> SearchIndexStateResponse: return SearchIndexStateResponse( provider_id=str(getattr(state, "provider_id")), module_id=str(getattr(state, "module_id")), resource_type=str(getattr(state, "resource_type")), index_version=int(getattr(state, "index_version")), status=str(getattr(state, "status")), checkpoint_cursor=getattr(state, "checkpoint_cursor"), high_watermark=getattr(state, "high_watermark"), last_change_cursor=getattr(state, "last_change_cursor"), indexed_documents=int( getattr(state, "indexed_documents") ), rejected_documents=int( getattr(state, "rejected_documents") ), rebuild_started_at=getattr(state, "rebuild_started_at"), rebuild_completed_at=getattr( state, "rebuild_completed_at", ), last_success_at=getattr(state, "last_success_at"), last_error=getattr(state, "last_error"), ) __all__ = ["router"]