File size: 1,881 Bytes
dd74be9 | 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 71 | import re
from sqlalchemy.orm import Session
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}
def score_department(department: Department, message: str) -> int:
query_value = message.lower().strip()
if not query_value:
return 0
score = 0
searchable_fields = [
department.name,
department.code,
department.description or "",
department.keywords or "",
department.query or "",
]
for field in searchable_fields:
field_value = field.lower()
if query_value in field_value:
score += 10
message_words = _normalize_words(query_value)
department_keywords = _normalize_words(department.keywords)
department_prompt_words = _normalize_words(department.query)
score += len(message_words & department_keywords) * 5
score += len(message_words & department_prompt_words) * 3
return score
def match_department(db: Session, message: str) -> tuple[Department | None, int]:
departments = (
db.query(Department)
.filter(Department.is_active.is_(True))
.all()
)
best_department = None
best_score = 0
for department in departments:
score = score_department(department, message)
if score > best_score:
best_department = department
best_score = score
if best_department is not None:
return best_department, best_score
fallback_department = next(
(department for department in departments if department.name.lower() == "admin"),
None,
)
if fallback_department is not None:
return fallback_department, 0
return (departments[0], 0) if departments else (None, 0)
|