from __future__ import annotations import hashlib import uuid from typing import Literal from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.orm import Session from app.api.deps import get_current_user, get_db from app.core.config import get_settings from app.core.ratelimit import SlidingWindowLimiter from app.models.audit import AuditEvent from app.schemas import CurrentUser from app.services.audit import record_audit_event from app.services.knowledge import GroundedAnswer, KnowledgeHealth, get_knowledge_provider router = APIRouter(prefix="/api/v1/knowledge", tags=["knowledge"]) settings = get_settings() _question_limiter = ( SlidingWindowLimiter( max_requests=settings.knowledge_max_requests, window_seconds=settings.knowledge_rate_limit_window_seconds, ) if settings.knowledge_max_requests > 0 else None ) SupportedLanguage = Literal["nl-BE", "en-GB", "fr-BE"] class AskQuestionRequest(BaseModel): question: str = Field(min_length=3, max_length=1000) language: SupportedLanguage = "en-GB" class KnowledgeFeedbackRequest(BaseModel): correlation_id: uuid.UUID helpful: bool @router.post("/questions", response_model=GroundedAnswer) def ask_question( body: AskQuestionRequest, request: Request, db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> GroundedAnswer: if _question_limiter is not None: forwarded = request.headers.get("x-forwarded-for", "") client_ip = ( forwarded.split(",")[-1].strip() if forwarded else request.client.host if request.client else "unknown" ) token = request.cookies.get(settings.session_cookie_name, "") session_key = hashlib.sha256(token.encode("utf-8")).hexdigest() retry_after = max( _question_limiter.consume(f"ip:{client_ip}"), _question_limiter.consume(f"session:{session_key}"), ) if retry_after: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many knowledge questions. Try again later.", headers={"Retry-After": str(retry_after)}, ) correlation_id = str(uuid.uuid4()) provider = get_knowledge_provider() answer = provider.ask(body.question, correlation_id, body.language) record_audit_event( db, actor_type="user", actor_label=user.display_name, action="knowledge_question_asked", entity_type="knowledge", correlation_id=uuid.UUID(correlation_id), metadata={ "evidence_state": answer.evidence_state, "provider": answer.provider, "source_ids": [s.document_id for s in answer.sources], "question_length": len(body.question), "language": body.language, }, ) db.commit() return answer @router.post("/feedback") def record_feedback( body: KnowledgeFeedbackRequest, db: Session = Depends(get_db), user: CurrentUser = Depends(get_current_user), ) -> dict[str, str]: question_event = db.scalar( select(AuditEvent.id).where( AuditEvent.action == "knowledge_question_asked", AuditEvent.correlation_id == body.correlation_id, AuditEvent.actor_label == user.display_name, ) ) if question_event is None: raise HTTPException(status_code=404, detail="Knowledge exchange not found") existing = db.scalar( select(AuditEvent).where( AuditEvent.action == "knowledge_feedback_recorded", AuditEvent.correlation_id == body.correlation_id, AuditEvent.actor_label == user.display_name, ) ) if existing is not None: existing.metadata_json = {"helpful": body.helpful} else: record_audit_event( db, actor_type="user", actor_label=user.display_name, action="knowledge_feedback_recorded", entity_type="knowledge", correlation_id=body.correlation_id, metadata={"helpful": body.helpful}, ) db.commit() return {"status": "recorded"} @router.get("/status", response_model=KnowledgeHealth) def knowledge_status( language: SupportedLanguage = "en-GB", db: Session = Depends(get_db), _user: CurrentUser = Depends(get_current_user), ) -> KnowledgeHealth: health = get_knowledge_provider().health(language) if health.provider != "ragcore": return health latest_sync = db.scalar( select(AuditEvent) .where(AuditEvent.action == "n8n_procedures_synced") .order_by(AuditEvent.occurred_at.desc()) .limit(1) ) if latest_sync is None: return health reported = latest_sync.after_json or {} synced = reported.get("synced") failed = reported.get("failed") return health.model_copy( update={ "reported_synced_document_count": synced if isinstance(synced, int) else None, "reported_failed_document_count": failed if isinstance(failed, int) else None, "last_sync_at": latest_sync.occurred_at, # A persisted sync callback is useful additional provenance, but must not # downgrade stronger provider-side verification to merely "reported". "statistics_state": ( health.statistics_state if health.statistics_state == "verified" else "sync_reported" ), } )