File size: 2,222 Bytes
dd74be9 eeff6a0 dd74be9 eeff6a0 dd74be9 eeff6a0 dd74be9 eeff6a0 dd74be9 eeff6a0 dd74be9 39a56a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | from sqlalchemy.orm import Session, selectinload
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:
if score <= 0:
return 0.5
return round(min(0.99, 0.55 + score / 20), 2)
def create_query(db: Session, *, current_user: User, payload: QueryCreate) -> Query:
message = payload.message.strip()
if not message:
raise ValueError("Query message cannot be empty")
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,
message=message,
department_id=department.id if department is not None else None,
status="Routed" if department is not None else "Open",
priority="Normal",
confidence=confidence,
)
db.add(query)
db.commit()
db.refresh(query)
return query
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 (
queries
.order_by(Query.created_at.desc(), Query.id.desc())
.all()
)
def list_all_queries(db: Session) -> list[Query]:
return (
db.query(Query)
.options(selectinload(Query.department))
.order_by(Query.created_at.desc(), Query.id.desc())
.all()
)
|