Add Gemini query routing and department-scoped ticket access

app/schemas/department.py

@@ -13,5 +13,4 @@ class DepartmentCreate(BaseModel):
 class DepartmentResponse(DepartmentCreate):
     id: int
     is_active: bool
-
     model_config = ConfigDict(from_attributes=True) 

app/services/classification_service.py

@@ -0,0 +1,127 @@
+import json
+import logging
+import os
+
+import httpx
+from google import genai
+from google.genai import errors, types
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+import app.config  # Loads Backend/.env
+from app.models.department import Department
+
+logger = logging.getLogger(__name__)
+
+
+class ClassificationResult(BaseModel):
+    department_id: int | None
+    confidence: float = Field(ge=0, le=1)
+
+
+def classify_query(
+    db: Session,
+    message: str,
+) -> tuple[Department, float] | None:
+    """Return a validated Gemini result, or None to use keyword routing."""
+    gemini_keys = [
+    os.getenv("GEMINI_API_KEY_1"),
+    os.getenv("GEMINI_API_KEY_2"),
+    os.getenv("GEMINI_API_KEY_3"),
+    os.getenv("GEMINI_API_KEY_4"),
+    os.getenv("GEMINI_API_KEY_5"),
+    os.getenv("GEMINI_API_KEY_6"),
+    os.getenv("GEMINI_API_KEY_7"),
+    os.getenv("GEMINI_API_KEY_8"),
+    os.getenv("GEMINI_API_KEY_9"),
+]
+
+
+    gemini_keys = [
+    key for key in gemini_keys if key
+]
+
+
+    api_key = gemini_keys[0] if gemini_keys else None
+    if not api_key:
+        logger.warning("Gemini key missing; using keyword routing.")
+        return None
+
+    departments = (
+        db.query(Department)
+        .filter(Department.is_active.is_(True))
+        .all()
+    )
+    if not departments:
+        return None
+
+    department_data = [
+        {
+            "id": department.id,
+            "name": department.name,
+            "code": department.code,
+            "description": department.description or "",
+            "keywords": department.keywords or "",
+            "example_queries": department.query or "",
+        }
+        for department in departments
+    ]
+
+    try:
+        with genai.Client(
+            api_key=api_key,
+            http_options=types.HttpOptions(
+                timeout=15000,
+                retry_options=types.HttpRetryOptions(attempts=1),
+            ),
+        ) as client:
+            response = client.models.generate_content(
+                  model="gemini-3.6-flash",
+                contents=json.dumps({
+                    "departments": department_data,
+                    "student_message": message,
+                }),
+                config=types.GenerateContentConfig(
+                    system_instruction=(
+                     "You are an AI query routing assistant for  a university. "
+                      "Classify student queries and select the most suitable department. "
+                      "Use only the supplied department information. "
+                      "Return only the department_id and confidence score. "
+                      "Do not answer the student query."
+),
+                    response_mime_type="application/json",
+                    response_schema=ClassificationResult,
+                    automatic_function_calling=(
+                        types.AutomaticFunctionCallingConfig(disable=True)
+                    ),
+                ),
+            )
+
+        result = ClassificationResult.model_validate_json(
+            response.text or ""
+        )
+
+        department = next(
+            (
+                department
+                for department in departments
+                if department.id == result.department_id
+            ),
+            None,
+        )
+
+        if department is None:
+            logger.warning(
+                "Gemini returned no valid department; using keyword routing."
+            )
+            return None
+
+        return department, result.confidence
+
+    except (errors.APIError, httpx.HTTPError, ValueError) as exc:
+        # Log the error type without exposing credentials or query content.
+        logger.warning(
+            "Gemini classification failed (%s); using keyword routing.",
+            type(exc).__name__,
+        )
+        return None

app/services/query_service.py

@@ -4,6 +4,7 @@ from app.models.query import Query
 from app.models.user import User
 from app.schemas.query import QueryCreate
 from app.services.routing_service import match_department
+from app.services.classification_service import classify_query
 
 
 def _confidence_from_score(score: int) -> float:
@@ -17,7 +18,14 @@ def create_query(db: Session, *, current_user: User, payload: QueryCreate) -> Qu
     if not message:
         raise ValueError("Query message cannot be empty")
 
-    department, score = match_department(db, message)
+    classification = classify_query(db, message)
+
+    if classification is None:
+        department, score = match_department(db, message)
+        confidence = _confidence_from_score(score)
+    else:
+        department, confidence = classification
+
     query = Query(
         student_id=current_user.id,
         sender_email=current_user.email,
@@ -25,7 +33,7 @@ def create_query(db: Session, *, current_user: User, payload: QueryCreate) -> Qu
         department_id=department.id if department is not None else None,
         status="Routed" if department is not None else "Open",
         priority="Normal",
-        confidence=_confidence_from_score(score),
+        confidence=confidence,
     )
     db.add(query)
     db.commit()
@@ -34,10 +42,19 @@ def create_query(db: Session, *, current_user: User, payload: QueryCreate) -> Qu
 
 
 def list_queries_for_user(db: Session, *, current_user: User) -> list[Query]:
+    queries = db.query(Query).options(selectinload(Query.department))
+    if current_user.is_admin:
+        return queries.order_by(Query.created_at.desc(), Query.id.desc()).all()
+    if current_user.role.strip().lower() == "student":
+        queries = queries.filter(Query.student_id == current_user.id)
+    elif current_user.department_id is not None:
+        queries = queries.filter(Query.department_id == current_user.department_id)
+    else:
+        # Staff without an assigned department cannot read unassigned queries.
+        return []
+
     return (
-        db.query(Query)
-        .options(selectinload(Query.department))
-        .filter(Query.student_id == current_user.id)
+        queries
         .order_by(Query.created_at.desc(), Query.id.desc())
         .all()
     )

app/services/routing_service.py

@@ -6,15 +6,26 @@ from app.models.department import Department
 
 
 def _normalize_words(value: str | None) -> set[str]:
-    if not value:
-        return set()
-
-    return {word for word in re.split(r"[^a-zA-Z0-9]+", value.lower()) if word}
+    stop_words = {
+        "i", "me", "my", "we", "our", "you", "your",
+        "a", "an", "the", "is", "are", "am", "was", "were",
+        "be", "been", "have", "has", "had", "do", "does",
+        "and", "or", "but", "to", "of", "for", "in",
+        "on", "at", "with", "this", "that", "these", "those",
+        "please", "dear", "sir", "thank", "thanks", "also",
+        "want", "need", "can", "could", "would",
+    }
+
+    return {
+        word
+        for word in re.split(r"[^a-zA-Z0-9]+", (value or "").lower())
+        if word and word not in stop_words
+    }
 
 
 def score_department(department: Department, message: str) -> int:
     query_value = message.lower().strip()
-    if not query_value:
+    if not _normalize_words(query_value):
         return 0
 
     score = 0
@@ -67,4 +78,4 @@ def match_department(db: Session, message: str) -> tuple[Department | None, int]
     if fallback_department is not None:
         return fallback_department, 0
 
-    return (departments[0], 0) if departments else (None, 0)
+    return None, 0

requirements.txt

@@ -4,3 +4,4 @@ sqlalchemy
 psycopg2-binary
 bcrypt
 PyJWT
+google-genai

tests/test_permissions.py

@@ -0,0 +1,186 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.core.database import Base
+from app.core.security import create_access_token, hash_password
+from app.dependencies import get_db
+from app.models.department import Department
+from app.models.query import Query
+from app.models.user import User
+from app.routes import admin, auth, queries
+
+
+@pytest.fixture(scope="module")
+def password_hash():
+    return hash_password("test-password-123")
+
+
+@pytest.fixture
+def workspace(password_hash):
+    engine = create_engine(
+        "sqlite+pysqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    with sessionmaker(bind=engine)() as db:
+        departments = [
+            Department(id=11, name="Examination Department", code="EXAM", keywords="exam"),
+            Department(id=22, name="Finance Department", code="FIN", keywords="fees"),
+            Department(id=33, name="Registrar", code="REG", keywords="transcript"),
+            Department(id=44, name="IT Dept", code="IT", keywords="wifi"),
+            Department(id=55, name="Instructor", code="INST", keywords="feedback"),
+            Department(id=66, name="Admin", code="ADMIN", keywords="general"),
+        ]
+        db.add_all(departments)
+        db.flush()
+        users = {
+            "student": User(email="student@example.com", role="STUDENT"),
+            "other_student": User(email="other@example.com", role="STUDENT"),
+            "admin": User(email="admin@example.com", role="ADMIN", is_admin=True),
+            "unassigned_staff": User(email="unassigned@example.com", role="STAFF"),
+        }
+        for department in departments:
+            users[department.code] = User(
+                email=f"staff{department.id}@example.com",
+                role=department.name,
+                department_id=department.id,
+            )
+        for user in users.values():
+            user.hashed_password = password_hash
+        db.add_all(users.values())
+        db.flush()
+        for query_id, department in enumerate(departments, start=101):
+            db.add(Query(
+                id=query_id, student_id=users["student"].id,
+                sender_email=users["student"].email,
+                message=f"Question for {department.name}",
+                department_id=department.id,
+            ))
+        db.add_all([
+            Query(id=107, student_id=users["other_student"].id, sender_email="other@example.com",
+                  message="Another student's fees query", department_id=22),
+            Query(id=108, student_id=users["other_student"].id, sender_email="other@example.com",
+                  message="Unassigned query"),
+            Query(id=109, student_id=users["FIN"].id, sender_email=users["FIN"].email,
+                  message="Staff's own exam question", department_id=11),
+        ])
+        db.commit()
+
+        app = FastAPI()
+        for router in (auth.router, queries.router, admin.router):
+            app.include_router(router)
+
+        def test_db():
+            yield db
+
+        app.dependency_overrides[get_db] = test_db
+        with TestClient(app) as client:
+            yield client, db, users
+    engine.dispose()
+
+
+def headers(user):
+    return {"Authorization": f"Bearer {create_access_token(str(user.id))}"}
+
+
+@pytest.mark.parametrize(("account", "expected_ids"), [
+    ("EXAM", {101, 109}), ("FIN", {102, 107}), ("REG", {103}),
+    ("IT", {104}), ("INST", {105}), ("ADMIN", {106}),
+])
+def test_department_staff_only_receive_assigned_queries(workspace, account, expected_ids):
+    client, _, users = workspace
+    response = client.get("/queries/", headers=headers(users[account]))
+    assert response.status_code == 200
+    assert {item["id"] for item in response.json()} == expected_ids
+
+
+@pytest.mark.parametrize("path", ["/queries/", "/admin/queries"])
+def test_admin_receives_all_departments_and_unassigned_queries(workspace, path):
+    client, _, users = workspace
+    response = client.get(path, headers=headers(users["admin"]))
+    assert response.status_code == 200
+    assert {item["id"] for item in response.json()} == set(range(101, 110))
+
+
+@pytest.mark.parametrize(("account", "expected_ids"), [
+    ("student", {101, 102, 103, 104, 105, 106}), ("other_student", {107, 108}),
+])
+def test_students_only_receive_their_own_queries(workspace, account, expected_ids):
+    client, _, users = workspace
+    response = client.get("/queries/", headers=headers(users[account]))
+    assert response.status_code == 200
+    assert {item["id"] for item in response.json()} == expected_ids
+
+
+def test_student_role_does_not_grant_department_access(workspace):
+    client, db, users = workspace
+    users["student"].department_id = 22
+    db.commit()
+    response = client.get("/queries/", headers=headers(users["student"]))
+    assert {item["id"] for item in response.json()} == {101, 102, 103, 104, 105, 106}
+
+
+def test_staff_without_department_cannot_see_unassigned_queries(workspace):
+    client, _, users = workspace
+    response = client.get("/queries/", headers=headers(users["unassigned_staff"]))
+    assert response.status_code == 200
+    assert response.json() == []
+
+
+@pytest.mark.parametrize("account", ["EXAM", "FIN", "REG", "IT", "INST", "ADMIN", "student"])
+def test_only_admin_can_use_all_queries_endpoint(workspace, account):
+    client, _, users = workspace
+    assert client.get("/admin/queries", headers=headers(users[account])).status_code == 403
+
+
+@pytest.mark.parametrize("path", ["/queries/", "/admin/queries"])
+def test_reading_queries_requires_login(workspace, path):
+    client, _, _ = workspace
+    assert client.get(path).status_code == 401
+
+
+def test_request_parameters_cannot_change_a_staff_members_scope(workspace):
+    client, _, users = workspace
+    response = client.get("/queries/?department_id=11&role=Admin&is_admin=true",
+                          headers=headers(users["FIN"]))
+    assert response.status_code == 200
+    assert {item["id"] for item in response.json()} == {102, 107}
+
+
+@pytest.mark.parametrize(("keyword", "department_id", "staff_account"), [
+    ("exam", 11, "EXAM"), ("fees", 22, "FIN"), ("transcript", 33, "REG"),
+    ("wifi", 44, "IT"), ("feedback", 55, "INST"), ("general", 66, "ADMIN"),
+])
+def test_submitted_query_is_visible_to_its_department_student_and_admin(workspace, keyword, department_id, staff_account):
+    client, _, users = workspace
+    response = client.post("/queries/", headers=headers(users["student"]), json={"message": keyword})
+    assert response.status_code == 201
+    query_id = response.json()["id"]
+    assert response.json()["department_id"] == department_id
+    for account in ("student", staff_account, "admin"):
+        response = client.get("/queries/", headers=headers(users[account]))
+        assert query_id in {item["id"] for item in response.json()}
+    for account in ("EXAM", "FIN", "REG", "IT", "INST", "ADMIN", "other_student"):
+        if account != staff_account:
+            response = client.get("/queries/", headers=headers(users[account]))
+            assert query_id not in {item["id"] for item in response.json()}
+
+
+def test_login_uses_saved_account_role_for_query_access(workspace):
+    client, _, users = workspace
+    response = client.post("/auth/login", json={
+        "email": users["FIN"].email, "password": "test-password-123",
+        "selectedRole": "Admin", "is_admin": True,
+    })
+    assert response.status_code == 200
+    assert response.json()["user"]["is_admin"] is False
+    assert response.json()["user"]["department_id"] == 22
+    token_headers = {"Authorization": f"Bearer {response.json()['access_token']}"}
+    response = client.get("/queries/", headers=token_headers)
+    assert {item["id"] for item in response.json()} == {102, 107}
+    assert client.get("/admin/queries", headers=token_headers).status_code == 403

tests/test_routing.py

@@ -0,0 +1,32 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from app.core.database import Base
+from app.models.department import Department
+from app.services.routing_service import match_department
+
+
+@pytest.mark.parametrize("include_finance", [False, True])
+def test_unmatched_query_without_admin_stays_unassigned(include_finance):
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    try:
+        Base.metadata.create_all(engine)
+        with Session(engine) as db:
+            if include_finance:
+                db.add(Department(
+                    name="Finance",
+                    code="FIN",
+                    keywords="tuition, fee, refund",
+                    query="Sir i want refund my fee",
+                ))
+                db.commit()
+
+            result = match_department(
+                db,
+                "I have updated the syllabus. Please review the course outline.",
+            )
+
+            assert result == (None, 0)
+    finally:
+        engine.dispose()