| import re | |
| from sqlalchemy.orm import Session | |
| from app.models.department import Department | |
| def _normalize_words(value: str | None) -> set[str]: | |
| 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 _normalize_words(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 None, 0 | |