1828 lines
78 KiB
Python
1828 lines
78 KiB
Python
"""
|
|
AI Service - Base class for all AI interactions using LiteLLM
|
|
|
|
This module provides a unified interface for AI operations using LiteLLM
|
|
with OpenRouter as the provider. This replaces the stub AI engine.
|
|
|
|
Features:
|
|
- Complaint analysis (severity, priority classification)
|
|
- Chat completion for general AI tasks
|
|
- Sentiment analysis
|
|
- Entity extraction
|
|
- Language detection
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Dict, List, Optional, Any
|
|
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AIServiceError(Exception):
|
|
"""Custom exception for AI service errors"""
|
|
|
|
pass
|
|
|
|
|
|
class AIService:
|
|
"""
|
|
Base AI Service class using LiteLLM with OpenRouter.
|
|
|
|
This is the single source of truth for all AI interactions in the application.
|
|
"""
|
|
|
|
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
|
|
# OPENROUTER_API_KEY = "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"
|
|
OPENROUTER_API_KEY = "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"
|
|
|
|
# Default configuration
|
|
DEFAULT_MODEL = "openrouter/google/gemini-2.5-flash-lite"
|
|
# DEFAULT_MODEL = "openrouter/xiaomi/mimo-v2-flash:free"
|
|
DEFAULT_TEMPERATURE = 0.3
|
|
DEFAULT_MAX_TOKENS = 500
|
|
DEFAULT_TIMEOUT = 30
|
|
|
|
# Severity choices
|
|
SEVERITY_CHOICES = ["low", "medium", "high", "critical"]
|
|
|
|
# Priority choices
|
|
PRIORITY_CHOICES = ["low", "medium", "high"]
|
|
|
|
@classmethod
|
|
def _get_api_key(cls) -> str:
|
|
"""Get OpenRouter API key from settings"""
|
|
# Use 'or' operator to fall back to DEFAULT_API_KEY when setting is empty or not set
|
|
api_key = cls.OPENROUTER_API_KEY
|
|
os.environ["OPENROUTER_API_KEY"] = api_key
|
|
os.environ["OPENROUTER_API_BASE"] = cls.OPENROUTER_BASE_URL
|
|
return api_key
|
|
|
|
@classmethod
|
|
def _get_model(cls) -> str:
|
|
"""Get AI model from settings"""
|
|
return getattr(settings, "AI_MODEL") or cls.DEFAULT_MODEL
|
|
|
|
@classmethod
|
|
def _get_temperature(cls) -> float:
|
|
"""Get AI temperature from settings"""
|
|
return float(getattr(settings, "AI_TEMPERATURE")) or cls.DEFAULT_TEMPERATURE
|
|
|
|
@classmethod
|
|
def _get_max_tokens(cls) -> int:
|
|
"""Get max tokens from settings"""
|
|
return int(getattr(settings, "AI_MAX_TOKENS")) or cls.DEFAULT_MAX_TOKENS
|
|
|
|
@classmethod
|
|
def _get_complaint_categories(cls) -> List[str]:
|
|
"""Get complaint categories from settings"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
|
|
return ComplaintCategory.objects.all().values_list("name_en", flat=True)
|
|
|
|
@classmethod
|
|
def _get_complaint_sub_categories(cls, category) -> List[str]:
|
|
"""Get complaint subcategories for a given category name"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
|
|
if category:
|
|
try:
|
|
# Find the category by name and get its subcategories
|
|
category_obj = ComplaintCategory.objects.filter(name_en=category).first()
|
|
if category_obj:
|
|
return ComplaintCategory.objects.filter(parent=category_obj).values_list("name_en", flat=True)
|
|
except Exception as e:
|
|
logger.error(f"Error fetching subcategories: {e}")
|
|
return []
|
|
|
|
@classmethod
|
|
def _get_all_categories_with_subcategories(cls) -> Dict[str, List[str]]:
|
|
"""Get all categories with their subcategories in a structured format"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
|
|
result = {}
|
|
try:
|
|
# Get all parent categories (no parent or parent is null)
|
|
parent_categories = ComplaintCategory.objects.filter(parent__isnull=True).all()
|
|
|
|
for category in parent_categories:
|
|
# Get subcategories for this parent
|
|
subcategories = list(
|
|
ComplaintCategory.objects.filter(parent=category).values_list("name_en", flat=True)
|
|
)
|
|
result[category.name_en] = subcategories if subcategories else []
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching categories with subcategories: {e}")
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def _get_taxonomy_hierarchy(cls) -> Dict:
|
|
"""
|
|
Get complete 4-level SHCT taxonomy hierarchy for AI classification.
|
|
|
|
Returns a structured dictionary representing the full taxonomy tree:
|
|
{
|
|
'domains': [
|
|
{
|
|
'code': 'CLINICAL',
|
|
'name_en': 'Clinical',
|
|
'name_ar': 'سريري',
|
|
'categories': [
|
|
{
|
|
'code': 'QUALITY',
|
|
'name_en': 'Quality',
|
|
'name_ar': 'الجودة',
|
|
'subcategories': [
|
|
{
|
|
'code': 'EXAMINATION',
|
|
'name_en': 'Examination',
|
|
'name_ar': 'الفحص',
|
|
'classifications': [
|
|
{
|
|
'code': 'exam_not_performed',
|
|
'name_en': 'Examination not performed',
|
|
'name_ar': 'لم يتم إجراء الفحص'
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
|
|
result = {"domains": []}
|
|
|
|
try:
|
|
# Get Level 1: Domains
|
|
domains = ComplaintCategory.objects.filter(
|
|
level=ComplaintCategory.LevelChoices.DOMAIN, is_active=True
|
|
).order_by("domain_type", "order")
|
|
|
|
for domain in domains:
|
|
domain_data = {
|
|
"code": domain.code or domain.name_en.upper(),
|
|
"name_en": domain.name_en,
|
|
"name_ar": domain.name_ar or "",
|
|
"categories": [],
|
|
}
|
|
|
|
# Get Level 2: Categories for this domain
|
|
categories = ComplaintCategory.objects.filter(
|
|
parent=domain, level=ComplaintCategory.LevelChoices.CATEGORY, is_active=True
|
|
).order_by("order")
|
|
|
|
for category in categories:
|
|
category_data = {
|
|
"code": category.code or category.name_en.upper(),
|
|
"name_en": category.name_en,
|
|
"name_ar": category.name_ar or "",
|
|
"subcategories": [],
|
|
}
|
|
|
|
# Get Level 3: Subcategories for this category
|
|
subcategories = ComplaintCategory.objects.filter(
|
|
parent=category, level=ComplaintCategory.LevelChoices.SUBCATEGORY, is_active=True
|
|
).order_by("order")
|
|
|
|
for subcategory in subcategories:
|
|
subcategory_data = {
|
|
"code": subcategory.code or subcategory.name_en.upper(),
|
|
"name_en": subcategory.name_en,
|
|
"name_ar": subcategory.name_ar or "",
|
|
"classifications": [],
|
|
}
|
|
|
|
# Get Level 4: Classifications for this subcategory
|
|
classifications = ComplaintCategory.objects.filter(
|
|
parent=subcategory, level=ComplaintCategory.LevelChoices.CLASSIFICATION, is_active=True
|
|
).order_by("order")
|
|
|
|
for classification in classifications:
|
|
classification_data = {
|
|
"code": classification.code,
|
|
"name_en": classification.name_en,
|
|
"name_ar": classification.name_ar or "",
|
|
}
|
|
subcategory_data["classifications"].append(classification_data)
|
|
|
|
category_data["subcategories"].append(subcategory_data)
|
|
|
|
domain_data["categories"].append(category_data)
|
|
|
|
result["domains"].append(domain_data)
|
|
|
|
logger.info(f"Taxonomy hierarchy loaded: {len(result['domains'])} domains")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching taxonomy hierarchy: {e}")
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def _find_category_by_name_or_code(
|
|
cls, name_or_code: str, level: int, parent_id: str = None, fuzzy_threshold: float = 0.85
|
|
) -> dict:
|
|
"""
|
|
Find a ComplaintCategory by name (English/Arabic) or code with fuzzy matching.
|
|
|
|
Args:
|
|
name_or_code: The name or code to search for
|
|
level: The level of category to find (1-4)
|
|
parent_id: Optional parent category ID for hierarchical search
|
|
fuzzy_threshold: Minimum similarity ratio for fuzzy matching (0.0 to 1.0)
|
|
|
|
Returns:
|
|
Dictionary with category details or None if not found:
|
|
{
|
|
'id': str,
|
|
'code': str,
|
|
'name_en': str,
|
|
'name_ar': str,
|
|
'level': int,
|
|
'parent_id': str or None,
|
|
'confidence': float
|
|
}
|
|
"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
from difflib import SequenceMatcher
|
|
|
|
if not name_or_code or not name_or_code.strip():
|
|
return None
|
|
|
|
search_term = name_or_code.strip().lower()
|
|
matches = []
|
|
|
|
# Build base query
|
|
query = ComplaintCategory.objects.filter(level=level, is_active=True)
|
|
if parent_id:
|
|
query = query.filter(parent_id=parent_id)
|
|
|
|
categories = list(query)
|
|
|
|
# Try exact matches first (English name, Arabic name, code)
|
|
for cat in categories:
|
|
# Exact match on code
|
|
if cat.code and cat.code.lower() == search_term:
|
|
return {
|
|
"id": str(cat.id),
|
|
"code": cat.code,
|
|
"name_en": cat.name_en,
|
|
"name_ar": cat.name_ar or "",
|
|
"level": cat.level,
|
|
"parent_id": str(cat.parent_id) if cat.parent else None,
|
|
"confidence": 1.0,
|
|
"match_type": "exact_code",
|
|
}
|
|
|
|
# Exact match on English name
|
|
if cat.name_en.lower() == search_term:
|
|
return {
|
|
"id": str(cat.id),
|
|
"code": cat.code or "",
|
|
"name_en": cat.name_en,
|
|
"name_ar": cat.name_ar or "",
|
|
"level": cat.level,
|
|
"parent_id": str(cat.parent_id) if cat.parent else None,
|
|
"confidence": 0.95,
|
|
"match_type": "exact_name_en",
|
|
}
|
|
|
|
# Exact match on Arabic name
|
|
if cat.name_ar and cat.name_ar.lower() == search_term:
|
|
return {
|
|
"id": str(cat.id),
|
|
"code": cat.code or "",
|
|
"name_en": cat.name_en,
|
|
"name_ar": cat.name_ar,
|
|
"level": cat.level,
|
|
"parent_id": str(cat.parent_id) if cat.parent else None,
|
|
"confidence": 0.95,
|
|
"match_type": "exact_name_ar",
|
|
}
|
|
|
|
# No exact match found, try fuzzy matching
|
|
for cat in categories:
|
|
# Try English name
|
|
ratio_en = SequenceMatcher(None, search_term, cat.name_en.lower()).ratio()
|
|
if ratio_en >= fuzzy_threshold:
|
|
matches.append(
|
|
{
|
|
"id": str(cat.id),
|
|
"code": cat.code or "",
|
|
"name_en": cat.name_en,
|
|
"name_ar": cat.name_ar or "",
|
|
"level": cat.level,
|
|
"parent_id": str(cat.parent_id) if cat.parent else None,
|
|
"confidence": ratio_en * 0.85, # Lower confidence for fuzzy matches
|
|
"match_type": "fuzzy_name_en",
|
|
}
|
|
)
|
|
|
|
# Try Arabic name
|
|
if cat.name_ar:
|
|
ratio_ar = SequenceMatcher(None, search_term, cat.name_ar.lower()).ratio()
|
|
if ratio_ar >= fuzzy_threshold:
|
|
# Avoid duplicate matches
|
|
if not any(m["id"] == str(cat.id) for m in matches):
|
|
matches.append(
|
|
{
|
|
"id": str(cat.id),
|
|
"code": cat.code or "",
|
|
"name_en": cat.name_en,
|
|
"name_ar": cat.name_ar,
|
|
"level": cat.level,
|
|
"parent_id": str(cat.parent_id) if cat.parent else None,
|
|
"confidence": ratio_ar * 0.85,
|
|
"match_type": "fuzzy_name_ar",
|
|
}
|
|
)
|
|
|
|
# Sort by confidence and return best match
|
|
if matches:
|
|
matches.sort(key=lambda x: x["confidence"], reverse=True)
|
|
logger.info(
|
|
f"Fuzzy match found for '{name_or_code}': {matches[0]['name_en']} (confidence: {matches[0]['confidence']:.2f})"
|
|
)
|
|
return matches[0]
|
|
|
|
logger.warning(f"No match found for taxonomy term: '{name_or_code}' (level: {level})")
|
|
return None
|
|
|
|
@classmethod
|
|
def _map_ai_taxonomy_to_db(cls, taxonomy_data: Dict) -> Dict:
|
|
"""
|
|
Map AI taxonomy classification to database objects.
|
|
|
|
Takes AI-provided taxonomy classification (codes/names for domain, category, subcategory, classification)
|
|
and maps them to actual ComplaintCategory database objects with fuzzy matching fallback.
|
|
|
|
Args:
|
|
taxonomy_data: Dictionary from AI with taxonomy classifications:
|
|
{
|
|
'domain': {'code': 'CLINICAL', 'name_en': 'Clinical', ...},
|
|
'category': {'code': 'QUALITY', 'name_en': 'Quality', ...},
|
|
'subcategory': {'code': 'EXAMINATION', 'name_en': 'Examination', ...},
|
|
'classification': {'code': 'exam_not_performed', 'name_en': 'Examination not performed', ...}
|
|
}
|
|
|
|
Returns:
|
|
Dictionary with mapped database IDs and confidence scores:
|
|
{
|
|
'domain': {'id': str, 'confidence': float, 'match_type': str} or None,
|
|
'category': {'id': str, 'confidence': float, 'match_type': str} or None,
|
|
'subcategory': {'id': str, 'confidence': float, 'match_type': str} or None,
|
|
'classification': {'id': str, 'confidence': float, 'match_type': str} or None,
|
|
'errors': list
|
|
}
|
|
"""
|
|
from apps.complaints.models import ComplaintCategory
|
|
|
|
result = {"domain": None, "category": None, "subcategory": None, "classification": None, "errors": []}
|
|
|
|
# Level 1: Domain (no parent)
|
|
if "domain" in taxonomy_data and taxonomy_data["domain"]:
|
|
domain_data = taxonomy_data["domain"]
|
|
domain_code = domain_data.get("code")
|
|
domain_name = domain_data.get("name_en")
|
|
|
|
# Try code first, then name
|
|
search_term = domain_code or domain_name
|
|
if search_term:
|
|
result["domain"] = cls._find_category_by_name_or_code(
|
|
name_or_code=search_term, level=ComplaintCategory.LevelChoices.DOMAIN, parent_id=None
|
|
)
|
|
if not result["domain"]:
|
|
result["errors"].append(f"Domain not found: {search_term}")
|
|
|
|
# Level 2: Category (child of domain)
|
|
if "category" in taxonomy_data and taxonomy_data["category"] and result["domain"]:
|
|
category_data = taxonomy_data["category"]
|
|
category_code = category_data.get("code")
|
|
category_name = category_data.get("name_en")
|
|
|
|
search_term = category_code or category_name
|
|
if search_term:
|
|
result["category"] = cls._find_category_by_name_or_code(
|
|
name_or_code=search_term,
|
|
level=ComplaintCategory.LevelChoices.CATEGORY,
|
|
parent_id=result["domain"]["id"],
|
|
)
|
|
if not result["category"]:
|
|
result["errors"].append(
|
|
f"Category not found: {search_term} (under domain: {result['domain']['name_en']})"
|
|
)
|
|
|
|
# Level 3: Subcategory (child of category)
|
|
if "subcategory" in taxonomy_data and taxonomy_data["subcategory"] and result["category"]:
|
|
subcategory_data = taxonomy_data["subcategory"]
|
|
subcategory_code = subcategory_data.get("code")
|
|
subcategory_name = subcategory_data.get("name_en")
|
|
|
|
search_term = subcategory_code or subcategory_name
|
|
if search_term:
|
|
result["subcategory"] = cls._find_category_by_name_or_code(
|
|
name_or_code=search_term,
|
|
level=ComplaintCategory.LevelChoices.SUBCATEGORY,
|
|
parent_id=result["category"]["id"],
|
|
)
|
|
if not result["subcategory"]:
|
|
result["errors"].append(
|
|
f"Subcategory not found: {search_term} (under category: {result['category']['name_en']})"
|
|
)
|
|
|
|
# Level 4: Classification (child of subcategory)
|
|
if "classification" in taxonomy_data and taxonomy_data["classification"] and result["subcategory"]:
|
|
classification_data = taxonomy_data["classification"]
|
|
classification_code = classification_data.get("code")
|
|
classification_name = classification_data.get("name_en")
|
|
|
|
search_term = classification_code or classification_name
|
|
if search_term:
|
|
result["classification"] = cls._find_category_by_name_or_code(
|
|
name_or_code=search_term,
|
|
level=ComplaintCategory.LevelChoices.CLASSIFICATION,
|
|
parent_id=result["subcategory"]["id"],
|
|
)
|
|
if not result["classification"]:
|
|
result["errors"].append(
|
|
f"Classification not found: {search_term} (under subcategory: {result['subcategory']['name_en']})"
|
|
)
|
|
|
|
logger.info(
|
|
f"Taxonomy mapping complete: domain={result['domain']}, category={result['category']}, subcategory={result['subcategory']}, classification={result['classification']}, errors={len(result['errors'])}"
|
|
)
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def _get_hospital_departments(cls, hospital_id: int) -> List[str]:
|
|
"""Get all departments for a specific hospital"""
|
|
from apps.organizations.models import Department
|
|
|
|
try:
|
|
departments = Department.objects.filter(hospital_id=hospital_id, status="active").values_list(
|
|
"name", flat=True
|
|
)
|
|
return list(departments)
|
|
except Exception as e:
|
|
logger.error(f"Error fetching hospital departments: {e}")
|
|
return []
|
|
|
|
@classmethod
|
|
def chat_completion(
|
|
cls,
|
|
prompt: str,
|
|
model: Optional[str] = None,
|
|
temperature: Optional[float] = None,
|
|
max_tokens: Optional[int] = None,
|
|
system_prompt: Optional[str] = None,
|
|
response_format: Optional[str] = None,
|
|
) -> str:
|
|
"""
|
|
Perform a chat completion using LiteLLM.
|
|
|
|
Args:
|
|
prompt: User prompt
|
|
model: AI model (uses default if not provided)
|
|
temperature: Temperature for randomness (uses default if not provided)
|
|
max_tokens: Maximum tokens to generate
|
|
system_prompt: System prompt to set context
|
|
response_format: Response format ('text' or 'json_object')
|
|
|
|
Returns:
|
|
Generated text response
|
|
|
|
Raises:
|
|
AIServiceError: If API call fails
|
|
"""
|
|
try:
|
|
from litellm import completion
|
|
|
|
api_key = cls._get_api_key()
|
|
|
|
model_name = model or cls._get_model()
|
|
temp = temperature if temperature is not None else cls._get_temperature()
|
|
max_tok = max_tokens or cls._get_max_tokens()
|
|
|
|
# Build messages
|
|
messages = []
|
|
if system_prompt:
|
|
messages.append({"role": "system", "content": system_prompt})
|
|
messages.append({"role": "user", "content": prompt})
|
|
|
|
# Build kwargs
|
|
model_name = model or cls._get_model()
|
|
kwargs = {"model": model_name, "messages": messages}
|
|
|
|
if response_format:
|
|
kwargs["response_format"] = {"type": response_format}
|
|
|
|
logger.info(f"AI Request: model={model_name}, temp={temp}")
|
|
|
|
response = completion(**kwargs)
|
|
|
|
content = response.choices[0].message.content
|
|
logger.info(f"AI Response: length={len(content)}")
|
|
|
|
return content
|
|
|
|
except Exception as e:
|
|
logger.error(f"AI service error: {str(e)}")
|
|
raise AIServiceError(f"Failed to get AI response: {str(e)}")
|
|
|
|
@classmethod
|
|
def analyze_complaint(
|
|
cls,
|
|
title: Optional[str] = None,
|
|
description: str = "",
|
|
category: Optional[str] = None,
|
|
hospital_id: Optional[int] = None,
|
|
use_taxonomy: bool = True,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Analyze a complaint and determine type (complaint vs appreciation), title, severity, priority,
|
|
4-level SHCT taxonomy (Domain, Category, Subcategory, Classification), and department.
|
|
|
|
Args:
|
|
title: Complaint title (optional, will be generated if not provided)
|
|
description: Complaint description
|
|
category: Complaint category (deprecated, kept for backward compatibility)
|
|
hospital_id: Hospital ID to fetch departments
|
|
use_taxonomy: Whether to use 4-level SHCT taxonomy classification (default: True)
|
|
|
|
Returns:
|
|
Dictionary with analysis:
|
|
{
|
|
'complaint_type': 'complaint' | 'appreciation', # Type of feedback
|
|
'title_en': str, # Generated or provided title (English)
|
|
'title_ar': str, # Generated or provided title (Arabic)
|
|
'short_description_en': str, # 2-3 sentence summary (English)
|
|
'short_description_ar': str, # 2-3 sentence summary (Arabic)
|
|
'suggested_action_en': str, # Suggested action (English)
|
|
'suggested_action_ar': str, # Suggested action (Arabic)
|
|
'severity': 'low' | 'medium' | 'high' | 'critical',
|
|
'priority': 'low' | 'medium' | 'high',
|
|
'category': str, # Legacy category name (deprecated, kept for backward compatibility)
|
|
'subcategory': str, # Legacy subcategory name (deprecated, kept for backward compatibility)
|
|
'department': str, # Name of department
|
|
'taxonomy': { # NEW: 4-level SHCT taxonomy classification
|
|
'domain': {
|
|
'code': 'CLINICAL',
|
|
'name_en': 'Clinical',
|
|
'name_ar': 'سريري',
|
|
'confidence': 0.95
|
|
},
|
|
'category': {
|
|
'code': 'QUALITY',
|
|
'name_en': 'Quality',
|
|
'name_ar': 'الجودة',
|
|
'confidence': 0.88
|
|
},
|
|
'subcategory': {
|
|
'code': 'EXAMINATION',
|
|
'name_en': 'Examination',
|
|
'name_ar': 'الفحص',
|
|
'confidence': 0.82
|
|
},
|
|
'classification': {
|
|
'code': 'exam_not_performed',
|
|
'name_en': 'Examination not performed',
|
|
'name_ar': 'لم يتم إجراء الفحص',
|
|
'confidence': 0.75
|
|
}
|
|
},
|
|
'staff_names': list, # All staff names mentioned
|
|
'primary_staff_name': str, # Primary staff name
|
|
'reasoning_en': str, # Explanation for classification (English)
|
|
'reasoning_ar': str # Explanation for classification (Arabic)
|
|
}
|
|
"""
|
|
# Check cache first
|
|
cache_key = f"complaint_analysis:{hash(str(title) + description + str(hospital_id) + str(use_taxonomy))}"
|
|
cached_result = cache.get(cache_key)
|
|
if cached_result:
|
|
logger.info("Using cached complaint analysis")
|
|
return cached_result
|
|
|
|
# Get 4-level SHCT taxonomy hierarchy
|
|
taxonomy_hierarchy = cls._get_taxonomy_hierarchy()
|
|
|
|
# Format taxonomy for prompt
|
|
taxonomy_text = cls._format_taxonomy_for_prompt(taxonomy_hierarchy)
|
|
|
|
# Get hospital departments if hospital_id is provided
|
|
departments_text = ""
|
|
if hospital_id:
|
|
departments = cls._get_hospital_departments(hospital_id)
|
|
if departments:
|
|
departments_text = f"\nAvailable Departments for this hospital:\n"
|
|
for dept in departments:
|
|
departments_text += f"- {dept}\n"
|
|
departments_text += "\n"
|
|
|
|
# Build prompt
|
|
title_text = f"Complaint Title: {title}\n" if title else ""
|
|
prompt = f"""Analyze this healthcare complaint and classify it using the 4-level SHCT taxonomy.
|
|
|
|
Complaint Description: {description}
|
|
{title_text}{departments_text}Severity Classification (choose one):
|
|
- low: Minor issues, no impact on patient care, routine matters
|
|
- medium: Moderate issues, some patient dissatisfaction, not urgent
|
|
- high: Serious issues, significant patient impact, requires timely attention
|
|
- critical: Emergency, immediate threat to patient safety, requires instant action
|
|
|
|
Priority Classification (choose one):
|
|
- low: Can be addressed within 1-2 weeks
|
|
- medium: Should be addressed within 3-5 days
|
|
- high: Requires immediate attention (within 24 hours)
|
|
|
|
4-Level SHCT Taxonomy Hierarchy:
|
|
{taxonomy_text}
|
|
|
|
Instructions:
|
|
1. If no title is provided, generate a concise title (max 10 words) that summarizes the complaint in BOTH English and Arabic
|
|
2. Generate a brief_summary (exactly 2-3 words) that serves as a quick tag/label for the complaint in BOTH English and Arabic. Examples: "Wait Time", "Staff Attitude", "Medication Error", "Billing Issue", "Facility Cleanliness", "Privacy Concern"
|
|
3. Generate a short_description (2-3 sentences) that captures the main issue and context in BOTH English and Arabic
|
|
3. Classify the complaint using the 4-level SHCT taxonomy:
|
|
a. Select the most appropriate DOMAIN (Level 1)
|
|
b. Select the most appropriate CATEGORY within that domain (Level 2)
|
|
c. Select the most appropriate SUBCATEGORY within that category (Level 3)
|
|
d. Select the most appropriate CLASSIFICATION within that subcategory (Level 4)
|
|
e. Use the CODE and NAME from the taxonomy above - DO NOT invent new categories
|
|
4. For each taxonomy level, assign a confidence score (0.0 to 1.0) reflecting how certain you are
|
|
5. Select the most appropriate department from the hospital's departments (if available)
|
|
6. Extract ALL staff members mentioned in the complaint (physicians, nurses, etc.)
|
|
7. Return ALL staff names WITHOUT titles (Dr., Nurse, دكتور, ممرض, etc.)
|
|
8. Identify the PRIMARY staff member (the one most relevant to the complaint)
|
|
9. If no staff is mentioned, return empty arrays for staff names
|
|
10. Generate 3-5 suggested_actions as a JSON list, each with:
|
|
- action: Specific, actionable step
|
|
- priority: high|medium|low
|
|
- category: clinical_quality|patient_safety|service_quality|staff_behavior|facility|process_improvement|other
|
|
Provide all actions in BOTH English and Arabic
|
|
|
|
IMPORTANT: ALL TEXT FIELDS MUST BE PROVIDED IN BOTH ENGLISH AND ARABIC
|
|
- title: Provide in both English and Arabic
|
|
- short_description: Provide in both English and Arabic
|
|
- suggested_actions: Provide as a list with English and Arabic for each action
|
|
- reasoning: Provide in both English and Arabic
|
|
|
|
Provide your analysis in JSON format:
|
|
{{
|
|
"title_en": "concise title in English summarizing the complaint (max 10 words)",
|
|
"title_ar": "العنوان بالعربية",
|
|
"brief_summary_en": "exactly 2-3 word tag/label in English (e.g., 'Wait Time', 'Staff Attitude', 'Billing Issue')",
|
|
"brief_summary_ar": "وصف مختصر من 2-3 كلمات بالعربية",
|
|
"short_description_en": "2-3 sentence summary in English of the complaint that captures the main issue and context",
|
|
"short_description_ar": "ملخص من 2-3 جمل بالعربية",
|
|
"severity": "low|medium|high|critical",
|
|
"priority": "low|medium|high",
|
|
"category": "exact category name from Level 2 of taxonomy (for backward compatibility)",
|
|
"subcategory": "exact subcategory name from Level 3 of taxonomy (for backward compatibility)",
|
|
"department": "exact department name from the hospital's departments, or empty string if not applicable",
|
|
"staff_names": ["name1", "name2", "name3"],
|
|
"primary_staff_name": "name of PRIMARY staff member (the one most relevant to the complaint), or empty string if no staff mentioned",
|
|
"suggested_actions": [
|
|
{{
|
|
"action_en": "Specific actionable step in English",
|
|
"action_ar": "خطوة محددة بالعربية",
|
|
"priority": "high|medium|low",
|
|
"category": "clinical_quality|patient_safety|service_quality|staff_behavior|facility|process_improvement|other"
|
|
}},
|
|
{{
|
|
"action_en": "Another action in English",
|
|
"action_ar": "إجراء آخر بالعربية",
|
|
"priority": "medium",
|
|
"category": "process_improvement"
|
|
}}
|
|
],
|
|
"reasoning_en": "Brief explanation in English of your classification (2-3 sentences)",
|
|
"reasoning_ar": "شرح مختصر بالعربية",
|
|
"taxonomy": {{
|
|
"domain": {{
|
|
"code": "exact code from taxonomy (e.g., CLINICAL)",
|
|
"name_en": "exact English name from taxonomy",
|
|
"name_ar": "exact Arabic name from taxonomy",
|
|
"confidence": 0.95
|
|
}},
|
|
"category": {{
|
|
"code": "exact code from taxonomy (e.g., QUALITY)",
|
|
"name_en": "exact English name from taxonomy",
|
|
"name_ar": "exact Arabic name from taxonomy",
|
|
"confidence": 0.88
|
|
}},
|
|
"subcategory": {{
|
|
"code": "exact code from taxonomy (e.g., EXAMINATION)",
|
|
"name_en": "exact English name from taxonomy",
|
|
"name_ar": "exact Arabic name from taxonomy",
|
|
"confidence": 0.82
|
|
}},
|
|
"classification": {{
|
|
"code": "exact code from taxonomy (e.g., exam_not_performed)",
|
|
"name_en": "exact English name from taxonomy",
|
|
"name_ar": "exact Arabic name from taxonomy",
|
|
"confidence": 0.75
|
|
}}
|
|
}}
|
|
}}"""
|
|
|
|
system_prompt = """You are a healthcare complaint analysis expert fluent in both English and Arabic.
|
|
Your job is to classify complaints using the 4-level SHCT taxonomy (Domain, Category, Subcategory, Classification).
|
|
Always use EXACT names and codes from the provided taxonomy - do not invent new categories.
|
|
Be conservative - when in doubt, choose a higher severity/priority.
|
|
Generate clear, concise titles that accurately summarize the complaint in BOTH English and Arabic.
|
|
Provide all text fields in both languages.
|
|
Assign realistic confidence scores based on how clearly the complaint fits each taxonomy level.
|
|
|
|
ARABIC LANGUAGE REQUIREMENTS:
|
|
- Use Modern Standard Arabic (Fusha) - الفصحى
|
|
- Use formal, professional medical terminology
|
|
- Ensure Arabic text is culturally appropriate for Saudi Arabian healthcare context
|
|
- Arabic translations should be natural and fluent, not literal word-for-word translations
|
|
- Use proper Arabic grammar and sentence structure
|
|
- Avoid mixing Arabic and English characters (no Arabizi)
|
|
- If the complaint is in Arabic, improve and professionalize the Arabic text while keeping the meaning
|
|
- If the complaint is in English, provide high-quality professional Arabic translations"""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
response_format="json_object",
|
|
temperature=0.2, # Lower temperature for consistent classification
|
|
)
|
|
|
|
# Parse JSON response
|
|
result = json.loads(response)
|
|
|
|
# Detect complaint type
|
|
complaint_type = cls._detect_complaint_type(description + " " + (title or ""))
|
|
result["complaint_type"] = complaint_type
|
|
|
|
# Map AI taxonomy to database objects
|
|
if use_taxonomy and "taxonomy" in result:
|
|
taxonomy_mapping = cls._map_ai_taxonomy_to_db(result["taxonomy"])
|
|
# Replace AI taxonomy IDs with database IDs
|
|
result["taxonomy_mapping"] = taxonomy_mapping
|
|
result["taxonomy"] = result["taxonomy"] # Keep original AI response
|
|
|
|
# Use provided title if available, otherwise use AI-generated title
|
|
if title:
|
|
result["title"] = title
|
|
|
|
# Validate severity
|
|
if result.get("severity") not in cls.SEVERITY_CHOICES:
|
|
result["severity"] = "medium"
|
|
logger.warning(f"Invalid severity, defaulting to medium")
|
|
|
|
# Validate priority
|
|
if result.get("priority") not in cls.PRIORITY_CHOICES:
|
|
result["priority"] = "medium"
|
|
logger.warning(f"Invalid priority, defaulting to medium")
|
|
|
|
# Validate Arabic text quality
|
|
result = cls._validate_arabic_text(result)
|
|
|
|
# Ensure title exists (for backward compatibility)
|
|
if not result.get("title"):
|
|
result["title"] = "Complaint"
|
|
|
|
# Cache result for 1 hour
|
|
cache.set(cache_key, result, timeout=3600)
|
|
|
|
logger.info(
|
|
f"Complaint analyzed: title={result['title']}, severity={result['severity']}, "
|
|
f"priority={result['priority']}, taxonomy={result.get('taxonomy', {}).get('domain', {}).get('name_en', 'N/A')}"
|
|
)
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Failed to parse AI response: {e}")
|
|
# Return defaults
|
|
return {
|
|
"title": title or "Complaint",
|
|
"title_en": title or "Complaint",
|
|
"title_ar": title or "شكوى",
|
|
"short_description_en": description[:200] if description else "",
|
|
"short_description_ar": description[:200] if description else "",
|
|
"severity": "medium",
|
|
"priority": "medium",
|
|
"category": "other",
|
|
"subcategory": "",
|
|
"department": "",
|
|
"staff_names": [],
|
|
"primary_staff_name": "",
|
|
"suggested_action_en": "",
|
|
"suggested_action_ar": "",
|
|
"reasoning_en": "AI analysis failed, using default values",
|
|
"reasoning_ar": "فشل تحليل الذكاء الاصطناعي، استخدام القيم الافتراضية",
|
|
"taxonomy": None,
|
|
"taxonomy_mapping": None,
|
|
}
|
|
except AIServiceError as e:
|
|
logger.error(f"AI service error: {e}")
|
|
return {
|
|
"title": title or "Complaint",
|
|
"title_en": title or "Complaint",
|
|
"title_ar": title or "شكوى",
|
|
"short_description_en": description[:200] if description else "",
|
|
"short_description_ar": description[:200] if description else "",
|
|
"severity": "medium",
|
|
"priority": "medium",
|
|
"category": "other",
|
|
"subcategory": "",
|
|
"department": "",
|
|
"staff_names": [],
|
|
"primary_staff_name": "",
|
|
"suggested_action_en": "",
|
|
"suggested_action_ar": "",
|
|
"reasoning_en": f"AI service unavailable: {str(e)}",
|
|
"reasoning_ar": f"خدمة الذكاء الاصطناعي غير متوفرة: {str(e)}",
|
|
"taxonomy": None,
|
|
"taxonomy_mapping": None,
|
|
}
|
|
|
|
@classmethod
|
|
def analyze_inquiry(
|
|
cls,
|
|
subject: str,
|
|
message: str,
|
|
category: Optional[str] = None,
|
|
hospital_id: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Analyze an inquiry to determine priority, department, and generate bilingual summaries.
|
|
|
|
Args:
|
|
subject: Inquiry subject/title
|
|
message: Inquiry message/body
|
|
category: Inquiry category (appointment, billing, etc.)
|
|
hospital_id: Hospital ID to fetch departments
|
|
|
|
Returns:
|
|
Dictionary with analysis:
|
|
{
|
|
'title_en': str,
|
|
'title_ar': str,
|
|
'brief_summary_en': str,
|
|
'brief_summary_ar': str,
|
|
'short_description_en': str,
|
|
'short_description_ar': str,
|
|
'priority': 'low' | 'medium' | 'high',
|
|
'department': str,
|
|
'reasoning_en': str,
|
|
'reasoning_ar': str,
|
|
}
|
|
"""
|
|
departments_text = ""
|
|
if hospital_id:
|
|
departments = cls._get_hospital_departments(hospital_id)
|
|
if departments:
|
|
departments_text = f"\nAvailable Departments for this hospital:\n"
|
|
for dept in departments:
|
|
departments_text += f"- {dept}\n"
|
|
departments_text += "\n"
|
|
|
|
prompt = f"""Analyze this healthcare inquiry and classify it appropriately.
|
|
|
|
Inquiry Subject: {subject}
|
|
Inquiry Message: {message}
|
|
Category: {category or "Not specified"}
|
|
{departments_text}Priority Classification (choose one):
|
|
- low: General information, routine questions, can be addressed within 1-2 weeks
|
|
- medium: Needs attention, moderate urgency, should be addressed within 3-5 days
|
|
- high: Urgent inquiry, requires immediate attention (within 24 hours)
|
|
|
|
Instructions:
|
|
1. Generate a concise title (max 10 words) that summarizes the inquiry in BOTH English and Arabic
|
|
2. Generate a brief_summary (exactly 2-3 words) as a quick tag/label in BOTH English and Arabic
|
|
3. Generate a short_description (2-3 sentences) capturing the main inquiry in BOTH English and Arabic
|
|
4. Select the most appropriate priority level
|
|
5. Select the most appropriate department from the hospital's departments (if available)
|
|
6. Provide reasoning for your classification in BOTH English and Arabic
|
|
|
|
IMPORTANT: ALL TEXT FIELDS MUST BE PROVIDED IN BOTH ENGLISH AND ARABIC
|
|
|
|
Provide your analysis in JSON format:
|
|
{{
|
|
"title_en": "concise title in English summarizing the inquiry (max 10 words)",
|
|
"title_ar": "العنوان بالعربية",
|
|
"brief_summary_en": "exactly 2-3 word tag/label in English",
|
|
"brief_summary_ar": "وصف مختصر من 2-3 كلمات بالعربية",
|
|
"short_description_en": "2-3 sentence summary in English of the inquiry",
|
|
"short_description_ar": "ملخص من 2-3 جمل بالعربية",
|
|
"priority": "low|medium|high",
|
|
"department": "exact department name from the hospital's departments, or empty string if not applicable",
|
|
"reasoning_en": "Brief explanation in English of your classification (2-3 sentences)",
|
|
"reasoning_ar": "شرح مختصر بالعربية"
|
|
}}"""
|
|
|
|
system_prompt = """You are a healthcare inquiry analysis expert fluent in both English and Arabic.
|
|
Your job is to analyze patient inquiries, determine priority, suggest appropriate department, and generate bilingual summaries.
|
|
Be accurate in priority assessment - err on the side of higher priority for sensitive medical topics.
|
|
Generate clear, concise summaries that accurately capture the inquiry in BOTH English and Arabic.
|
|
|
|
ARABIC LANGUAGE REQUIREMENTS:
|
|
- Use Modern Standard Arabic (Fusha) - الفصحى
|
|
- Use formal, professional medical terminology
|
|
- Ensure Arabic text is culturally appropriate for Saudi Arabian healthcare context
|
|
- Arabic translations should be natural and fluent, not literal word-for-word translations"""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
response_format="json_object",
|
|
temperature=0.2,
|
|
)
|
|
|
|
result = json.loads(response)
|
|
|
|
if result.get("priority") not in cls.PRIORITY_CHOICES:
|
|
result["priority"] = "medium"
|
|
logger.warning(f"Invalid priority for inquiry, defaulting to medium")
|
|
|
|
result = cls._validate_arabic_text(result)
|
|
|
|
logger.info(f"Inquiry analyzed: title={result.get('title_en')}, priority={result.get('priority')}")
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Failed to parse inquiry AI response: {e}")
|
|
return {
|
|
"title_en": subject,
|
|
"title_ar": subject,
|
|
"brief_summary_en": "Inquiry",
|
|
"brief_summary_ar": "استفسار",
|
|
"short_description_en": message[:200] if message else "",
|
|
"short_description_ar": message[:200] if message else "",
|
|
"priority": "medium",
|
|
"department": "",
|
|
"reasoning_en": "AI analysis failed, using default values",
|
|
"reasoning_ar": "فشل تحليل الذكاء الاصطناعي، استخدام القيم الافتراضية",
|
|
}
|
|
except AIServiceError as e:
|
|
logger.error(f"AI service error for inquiry: {e}")
|
|
return {
|
|
"title_en": subject,
|
|
"title_ar": subject,
|
|
"brief_summary_en": "Inquiry",
|
|
"brief_summary_ar": "استفسار",
|
|
"short_description_en": message[:200] if message else "",
|
|
"short_description_ar": message[:200] if message else "",
|
|
"priority": "medium",
|
|
"department": "",
|
|
"reasoning_en": f"AI service unavailable: {str(e)}",
|
|
"reasoning_ar": f"خدمة الذكاء الاصطناعي غير متوفرة: {str(e)}",
|
|
}
|
|
|
|
@classmethod
|
|
def analyze_observation(
|
|
cls,
|
|
title: Optional[str] = None,
|
|
description: str = "",
|
|
hospital_id: Optional[int] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Analyze a staff observation to determine severity, category, and generate bilingual summaries
|
|
and suggested PX actions for quality improvement.
|
|
|
|
Args:
|
|
title: Observation title (optional, will be generated if not provided)
|
|
description: Observation description
|
|
hospital_id: Hospital ID to fetch departments
|
|
|
|
Returns:
|
|
Dictionary with analysis:
|
|
{
|
|
'title_en': str,
|
|
'title_ar': str,
|
|
'brief_summary_en': str,
|
|
'brief_summary_ar': str,
|
|
'short_description_en': str,
|
|
'short_description_ar': str,
|
|
'severity': 'low' | 'medium' | 'high' | 'critical',
|
|
'category': str,
|
|
'department': str,
|
|
'suggested_actions': [
|
|
{'action_en': str, 'action_ar': str, 'priority': str, 'category': str}
|
|
],
|
|
'suggested_action_en': str,
|
|
'suggested_action_ar': str,
|
|
'reasoning_en': str,
|
|
'reasoning_ar': str,
|
|
}
|
|
"""
|
|
cache_key = f"observation_analysis:{hash(str(title) + description + str(hospital_id))}"
|
|
cached_result = cache.get(cache_key)
|
|
if cached_result:
|
|
logger.info("Using cached observation analysis")
|
|
return cached_result
|
|
|
|
departments_text = ""
|
|
if hospital_id:
|
|
departments = cls._get_hospital_departments(hospital_id)
|
|
if departments:
|
|
departments_text = f"\nAvailable Departments for this hospital:\n"
|
|
for dept in departments:
|
|
departments_text += f"- {dept}\n"
|
|
departments_text += "\n"
|
|
|
|
title_text = f"Observation Title: {title}\n" if title else ""
|
|
prompt = f"""Analyze this healthcare staff observation and classify it appropriately.
|
|
|
|
Observation Description: {description}
|
|
{title_text}{departments_text}Severity Classification (choose one):
|
|
- low: Minor issues, no impact on patient care or safety, routine matters
|
|
- medium: Moderate issues, potential for patient dissatisfaction, not urgent
|
|
- high: Serious issues, significant impact on patient care or safety, requires timely attention
|
|
- critical: Emergency, immediate threat to patient safety, requires instant action
|
|
|
|
Observation Categories (choose one):
|
|
- Patient Safety (سلامة المرضى)
|
|
- Clinical Quality (الجودة السريرية)
|
|
- Infection Control (مكافحة العدوى)
|
|
- Medication Safety (سلامة الأدوية)
|
|
- Equipment & Devices (المعدات والأجهزة)
|
|
- Facility & Environment (المرافق والبيئة)
|
|
- Staff Behavior (سلوك الموظفين)
|
|
- Communication (التواصل)
|
|
- Documentation (التوثيق)
|
|
- Process & Workflow (العمليات وسير العمل)
|
|
- Security (الأمن)
|
|
- IT & Systems (تقنية المعلومات والأنظمة)
|
|
- Housekeeping (التدبير المنزلي)
|
|
- Food Services (خدمات الطعام)
|
|
- Other (أخرى)
|
|
|
|
Instructions:
|
|
1. If no title is provided, generate a concise title (max 10 words) that summarizes the observation in BOTH English and Arabic
|
|
2. Generate a brief_summary (exactly 2-3 words) that serves as a quick tag/label for the observation in BOTH English and Arabic. Examples: "Fall Risk", "Hand Hygiene", "Equipment Failure", "Cleanliness Issue"
|
|
3. Generate a short_description (2-3 sentences) that captures the main observation and context in BOTH English and Arabic
|
|
4. Classify the observation using the appropriate category from the list above
|
|
5. Classify severity accurately - be conservative, when in doubt choose higher severity for patient safety
|
|
6. Select the most appropriate department from the hospital's departments (if available)
|
|
7. Generate 3-5 suggested_actions as improvement actions (PX Actions), each with:
|
|
- action: Specific, actionable improvement step
|
|
- priority: high|medium|low
|
|
- category: patient_safety|clinical_quality|service_quality|staff_behavior|facility|process_improvement|infection_control|medication_safety|other
|
|
Provide all actions in BOTH English and Arabic
|
|
8. Provide reasoning for your classification in BOTH English and Arabic
|
|
|
|
IMPORTANT: ALL TEXT FIELDS MUST BE PROVIDED IN BOTH ENGLISH AND ARABIC
|
|
- title: Provide in both English and Arabic
|
|
- brief_summary: Provide in both English and Arabic
|
|
- short_description: Provide in both English and Arabic
|
|
- suggested_actions: Provide as a list with English and Arabic for each action
|
|
- reasoning: Provide in both English and Arabic
|
|
|
|
Provide your analysis in JSON format:
|
|
{{
|
|
"title_en": "concise title in English summarizing the observation (max 10 words)",
|
|
"title_ar": "العنوان بالعربية",
|
|
"brief_summary_en": "exactly 2-3 word tag/label in English (e.g., 'Fall Risk', 'Hand Hygiene')",
|
|
"brief_summary_ar": "وصف مختصر من 2-3 كلمات بالعربية",
|
|
"short_description_en": "2-3 sentence summary in English of the observation that captures the main issue and context",
|
|
"short_description_ar": "ملخص من 2-3 جمل بالعربية",
|
|
"severity": "low|medium|high|critical",
|
|
"category": "exact category name from the list above",
|
|
"department": "exact department name from the hospital's departments, or empty string if not applicable",
|
|
"suggested_actions": [
|
|
{{
|
|
"action_en": "Specific actionable improvement step in English",
|
|
"action_ar": "خطوة محددة للتحسين بالعربية",
|
|
"priority": "high|medium|low",
|
|
"category": "patient_safety|clinical_quality|service_quality|staff_behavior|facility|process_improvement|infection_control|medication_safety|other"
|
|
}},
|
|
{{
|
|
"action_en": "Another action in English",
|
|
"action_ar": "إجراء آخر بالعربية",
|
|
"priority": "medium",
|
|
"category": "process_improvement"
|
|
}}
|
|
],
|
|
"suggested_action_en": "top suggested action in English (backward compatibility)",
|
|
"suggested_action_ar": "أهم إجراء مقترح بالعربية (لل توافق مع الأنظمة السابقة)",
|
|
"reasoning_en": "Brief explanation in English of your classification (2-3 sentences)",
|
|
"reasoning_ar": "شرح مختصر بالعربية"
|
|
}}"""
|
|
|
|
system_prompt = """You are a healthcare observation analysis expert fluent in both English and Arabic.
|
|
Your job is to analyze staff-reported observations in a healthcare setting, determine severity,
|
|
classify the observation category, and suggest actionable improvement steps (PX Actions).
|
|
Be conservative with severity - when in doubt, choose higher severity for patient safety concerns.
|
|
Generate clear, concise summaries that accurately capture the observation in BOTH English and Arabic.
|
|
Suggested actions should be specific, actionable, and focused on quality improvement.
|
|
|
|
ARABIC LANGUAGE REQUIREMENTS:
|
|
- Use Modern Standard Arabic (Fusha) - الفصحى
|
|
- Use formal, professional medical terminology
|
|
- Ensure Arabic text is culturally appropriate for Saudi Arabian healthcare context
|
|
- Arabic translations should be natural and fluent, not literal word-for-word translations
|
|
- Use proper Arabic grammar and sentence structure
|
|
- Avoid mixing Arabic and English characters (no Arabizi)
|
|
- If the observation is in Arabic, improve and professionalize the Arabic text while keeping the meaning
|
|
- If the observation is in English, provide high-quality professional Arabic translations"""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
response_format="json_object",
|
|
temperature=0.2,
|
|
)
|
|
|
|
result = json.loads(response)
|
|
|
|
if title:
|
|
result["title"] = title
|
|
|
|
if result.get("severity") not in cls.SEVERITY_CHOICES:
|
|
result["severity"] = "medium"
|
|
logger.warning(f"Invalid severity for observation, defaulting to medium")
|
|
|
|
result = cls._validate_arabic_text(result)
|
|
|
|
if not result.get("title"):
|
|
result["title"] = "Observation"
|
|
|
|
if not result.get("suggested_action_en") and result.get("suggested_actions"):
|
|
first_action = result["suggested_actions"][0]
|
|
result["suggested_action_en"] = first_action.get("action_en", "")
|
|
result["suggested_action_ar"] = first_action.get("action_ar", "")
|
|
|
|
cache.set(cache_key, result, timeout=3600)
|
|
|
|
logger.info(
|
|
f"Observation analyzed: title={result.get('title_en')}, severity={result.get('severity')}, "
|
|
f"category={result.get('category', 'N/A')}"
|
|
)
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Failed to parse observation AI response: {e}")
|
|
return {
|
|
"title": title or "Observation",
|
|
"title_en": title or "Observation",
|
|
"title_ar": title or "ملاحظة",
|
|
"brief_summary_en": "Observation",
|
|
"brief_summary_ar": "ملاحظة",
|
|
"short_description_en": description[:200] if description else "",
|
|
"short_description_ar": description[:200] if description else "",
|
|
"severity": "medium",
|
|
"category": "Other",
|
|
"department": "",
|
|
"suggested_actions": [],
|
|
"suggested_action_en": "",
|
|
"suggested_action_ar": "",
|
|
"reasoning_en": "AI analysis failed, using default values",
|
|
"reasoning_ar": "فشل تحليل الذكاء الاصطناعي، استخدام القيم الافتراضية",
|
|
}
|
|
except AIServiceError as e:
|
|
logger.error(f"AI service error for observation: {e}")
|
|
return {
|
|
"title": title or "Observation",
|
|
"title_en": title or "Observation",
|
|
"title_ar": title or "ملاحظة",
|
|
"brief_summary_en": "Observation",
|
|
"brief_summary_ar": "ملاحظة",
|
|
"short_description_en": description[:200] if description else "",
|
|
"short_description_ar": description[:200] if description else "",
|
|
"severity": "medium",
|
|
"category": "Other",
|
|
"department": "",
|
|
"suggested_actions": [],
|
|
"suggested_action_en": "",
|
|
"suggested_action_ar": "",
|
|
"reasoning_en": f"AI service unavailable: {str(e)}",
|
|
"reasoning_ar": f"خدمة الذكاء الاصطناعي غير متوفرة: {str(e)}",
|
|
}
|
|
|
|
@classmethod
|
|
def _format_taxonomy_for_prompt(cls, taxonomy_hierarchy: Dict) -> str:
|
|
"""
|
|
Format taxonomy hierarchy for AI prompt.
|
|
|
|
Args:
|
|
taxonomy_hierarchy: Dictionary from _get_taxonomy_hierarchy()
|
|
|
|
Returns:
|
|
Formatted string representation of taxonomy
|
|
"""
|
|
text = ""
|
|
|
|
for domain in taxonomy_hierarchy.get("domains", []):
|
|
text += f"\nDOMAIN: {domain['code']} - {domain['name_en']} ({domain['name_ar']})\n"
|
|
|
|
for category in domain.get("categories", []):
|
|
text += f" CATEGORY: {category['code']} - {category['name_en']} ({category['name_ar']})\n"
|
|
|
|
for subcategory in category.get("subcategories", []):
|
|
text += f" SUBCATEGORY: {subcategory['code']} - {subcategory['name_en']} ({subcategory['name_ar']})\n"
|
|
|
|
for classification in subcategory.get("classifications", []):
|
|
text += f" CLASSIFICATION: {classification['code']} - {classification['name_en']} ({classification['name_ar']})\n"
|
|
|
|
return text
|
|
|
|
@classmethod
|
|
def _validate_arabic_text(cls, result: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Validate and clean Arabic text fields to ensure quality.
|
|
|
|
Args:
|
|
result: Dictionary containing AI analysis results
|
|
|
|
Returns:
|
|
Validated and cleaned result dictionary
|
|
"""
|
|
import re
|
|
|
|
def is_valid_arabic(text: str) -> bool:
|
|
"""Check if text contains valid Arabic characters"""
|
|
if not text:
|
|
return False
|
|
# Check for Arabic Unicode range (including Arabic Presentation Forms)
|
|
arabic_pattern = re.compile(r"[\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF\uFB50-\uFDFF\uFE70-\uFEFF]")
|
|
return bool(arabic_pattern.search(text))
|
|
|
|
def clean_arabic_text(text: str) -> str:
|
|
"""Clean and normalize Arabic text"""
|
|
if not text:
|
|
return ""
|
|
# Remove excessive whitespace
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
# Remove mixed Arabic-English words (Arabizi)
|
|
# This is a basic check - more sophisticated detection could be added
|
|
return text
|
|
|
|
# Fields to validate
|
|
arabic_fields = ["title_ar", "brief_summary_ar", "short_description_ar", "reasoning_ar"]
|
|
|
|
for field in arabic_fields:
|
|
if field in result and result[field]:
|
|
# Clean the text
|
|
result[field] = clean_arabic_text(result[field])
|
|
|
|
# Check if valid Arabic
|
|
if not is_valid_arabic(result[field]):
|
|
logger.warning(f"Field {field} does not contain valid Arabic text")
|
|
# Generate fallback if needed
|
|
if field == "title_ar" and result.get("title_en"):
|
|
result[field] = f"[AR] {result['title_en']}" # Mark for manual translation
|
|
elif field == "brief_summary_ar" and result.get("brief_summary_en"):
|
|
result[field] = f"[AR] {result['brief_summary_en']}"
|
|
elif field == "short_description_ar" and result.get("short_description_en"):
|
|
result[field] = f"[AR] {result['short_description_en']}"
|
|
elif field == "reasoning_ar" and result.get("reasoning_en"):
|
|
result[field] = f"[AR] {result['reasoning_en']}"
|
|
|
|
# Validate suggested actions
|
|
if "suggested_actions" in result and isinstance(result["suggested_actions"], list):
|
|
for action in result["suggested_actions"]:
|
|
if "action_ar" in action and action["action_ar"]:
|
|
action["action_ar"] = clean_arabic_text(action["action_ar"])
|
|
if not is_valid_arabic(action["action_ar"]):
|
|
logger.warning("Suggested action does not contain valid Arabic text")
|
|
if "action_en" in action:
|
|
action["action_ar"] = f"[AR] {action['action_en']}"
|
|
|
|
# Validate taxonomy Arabic names
|
|
if "taxonomy" in result and result["taxonomy"]:
|
|
for level in ["domain", "category", "subcategory", "classification"]:
|
|
if level in result["taxonomy"] and result["taxonomy"][level]:
|
|
level_data = result["taxonomy"][level]
|
|
if "name_ar" in level_data and level_data["name_ar"]:
|
|
level_data["name_ar"] = clean_arabic_text(level_data["name_ar"])
|
|
if not is_valid_arabic(level_data["name_ar"]):
|
|
logger.warning(f"Taxonomy {level} does not contain valid Arabic name")
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def classify_sentiment(cls, text: str) -> Dict[str, Any]:
|
|
"""
|
|
Classify sentiment of text.
|
|
|
|
Args:
|
|
text: Text to analyze
|
|
|
|
Returns:
|
|
Dictionary with sentiment analysis:
|
|
{
|
|
'sentiment': 'positive' | 'neutral' | 'negative',
|
|
'score': float, # -1.0 to 1.0
|
|
'confidence': float # 0.0 to 1.0
|
|
}
|
|
"""
|
|
prompt = f"""Analyze the sentiment of this text:
|
|
|
|
{text}
|
|
|
|
Provide your analysis in JSON format:
|
|
{{
|
|
"sentiment": "positive|neutral|negative",
|
|
"score": float, # -1.0 (very negative) to 1.0 (very positive)
|
|
"confidence": float # 0.0 to 1.0
|
|
}}"""
|
|
|
|
system_prompt = """You are a sentiment analysis expert.
|
|
Analyze the emotional tone of the text accurately."""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt, system_prompt=system_prompt, response_format="json_object", temperature=0.1
|
|
)
|
|
|
|
result = json.loads(response)
|
|
return result
|
|
|
|
except (json.JSONDecodeError, AIServiceError) as e:
|
|
logger.error(f"Sentiment analysis failed: {e}")
|
|
return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0}
|
|
|
|
@classmethod
|
|
def analyze_emotion(cls, text: str) -> Dict[str, Any]:
|
|
"""
|
|
Analyze emotion in text to identify primary emotion and intensity.
|
|
|
|
Args:
|
|
text: Text to analyze (supports English and Arabic)
|
|
|
|
Returns:
|
|
Dictionary with emotion analysis:
|
|
{
|
|
'emotion': 'anger' | 'sadness' | 'confusion' | 'fear' | 'neutral',
|
|
'intensity': float, # 0.0 to 1.0 (how strong the emotion is)
|
|
'confidence': float # 0.0 to 1.0 (how confident AI is)
|
|
}
|
|
"""
|
|
prompt = f"""Analyze the primary emotion in this text (supports English and Arabic):
|
|
|
|
{text}
|
|
|
|
Identify the PRIMARY emotion from these options:
|
|
- anger: Strong feelings of displeasure, hostility, or rage
|
|
- sadness: Feelings of sorrow, grief, or unhappiness
|
|
- confusion: Lack of understanding, bewilderment, or uncertainty
|
|
- fear: Feelings of anxiety, worry, or being afraid
|
|
- neutral: No strong emotion detected
|
|
|
|
Provide your analysis in JSON format:
|
|
{{
|
|
"emotion": "anger|sadness|confusion|fear|neutral",
|
|
"intensity": float, # 0.0 (very weak) to 1.0 (extremely strong)
|
|
"confidence": float # 0.0 to 1.0 (how confident you are)
|
|
}}
|
|
|
|
Examples:
|
|
- "This is unacceptable! I demand to speak to management!" -> emotion: "anger", intensity: 0.9
|
|
- "I'm very disappointed with the care my father received" -> emotion: "sadness", intensity: 0.7
|
|
- "I don't understand what happened, can you explain?" -> emotion: "confusion", intensity: 0.5
|
|
- "I'm worried about the side effects of this medication" -> emotion: "fear", intensity: 0.6
|
|
- "I would like to report a minor issue" -> emotion: "neutral", intensity: 0.2
|
|
"""
|
|
|
|
system_prompt = """You are an emotion analysis expert fluent in both English and Arabic.
|
|
Analyze the text to identify the PRIMARY emotion and its intensity.
|
|
Be accurate in distinguishing between different emotions.
|
|
Provide intensity scores that reflect how strongly the emotion is expressed (0.0 to 1.0)."""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt, system_prompt=system_prompt, response_format="json_object", temperature=0.1
|
|
)
|
|
|
|
result = json.loads(response)
|
|
|
|
# Validate emotion
|
|
valid_emotions = ["anger", "sadness", "confusion", "fear", "neutral"]
|
|
if result.get("emotion") not in valid_emotions:
|
|
result["emotion"] = "neutral"
|
|
logger.warning(f"Invalid emotion detected, defaulting to neutral")
|
|
|
|
# Validate intensity
|
|
intensity = float(result.get("intensity", 0.0))
|
|
if not (0.0 <= intensity <= 1.0):
|
|
intensity = max(0.0, min(1.0, intensity))
|
|
result["intensity"] = intensity
|
|
logger.warning(f"Intensity out of range, clamping to {intensity}")
|
|
|
|
# Validate confidence
|
|
confidence = float(result.get("confidence", 0.0))
|
|
if not (0.0 <= confidence <= 1.0):
|
|
confidence = max(0.0, min(1.0, confidence))
|
|
result["confidence"] = confidence
|
|
logger.warning(f"Confidence out of range, clamping to {confidence}")
|
|
|
|
logger.info(f"Emotion analysis: {result['emotion']}, intensity={intensity}, confidence={confidence}")
|
|
return result
|
|
|
|
except (json.JSONDecodeError, AIServiceError) as e:
|
|
logger.error(f"Emotion analysis failed: {e}")
|
|
return {"emotion": "neutral", "intensity": 0.0, "confidence": 0.0}
|
|
|
|
@classmethod
|
|
def extract_entities(cls, text: str) -> List[Dict[str, str]]:
|
|
prompt = f"""Extract named entities from this text:
|
|
"{text}"
|
|
|
|
Focus heavily on PERSON names.
|
|
IMPORTANT: Extract the clean name only. Remove titles like 'Dr.', 'Nurse', 'Mr.', 'Professor', 'دكتور', 'ممرض'.
|
|
|
|
Provide entities in JSON format:
|
|
{{
|
|
"entities": [
|
|
{{"text": "Name", "type": "PERSON"}},
|
|
{{"text": "DepartmentName", "type": "ORGANIZATION"}}
|
|
]
|
|
}}"""
|
|
|
|
system_prompt = (
|
|
"You are an expert in bilingual NER (Arabic and English). Extract formal names for database lookup."
|
|
)
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt, system_prompt=system_prompt, response_format="json_object", temperature=0.0
|
|
)
|
|
return json.loads(response).get("entities", [])
|
|
except (json.JSONDecodeError, AIServiceError):
|
|
return []
|
|
|
|
@classmethod
|
|
def generate_summary(cls, text: str, max_length: int = 200) -> str:
|
|
"""
|
|
Generate a summary of text.
|
|
|
|
Args:
|
|
text: Text to summarize
|
|
max_length: Maximum length of summary
|
|
|
|
Returns:
|
|
Summary text
|
|
"""
|
|
prompt = f"""Summarize this text in {max_length} characters or less:
|
|
|
|
{text}"""
|
|
|
|
system_prompt = """You are a text summarization expert.
|
|
Create a concise summary that captures the main points."""
|
|
|
|
try:
|
|
response = cls.chat_completion(prompt=prompt, system_prompt=system_prompt, temperature=0.3, max_tokens=150)
|
|
|
|
return response.strip()
|
|
|
|
except AIServiceError as e:
|
|
logger.error(f"Summary generation failed: {e}")
|
|
return text[:max_length]
|
|
|
|
@classmethod
|
|
def create_px_action_from_complaint(cls, complaint) -> Dict[str, Any]:
|
|
"""
|
|
Generate PX Action data from a complaint using AI analysis.
|
|
|
|
Args:
|
|
complaint: Complaint model instance
|
|
|
|
Returns:
|
|
Dictionary with PX Action data:
|
|
{
|
|
'title': str,
|
|
'description': str,
|
|
'category': str,
|
|
'priority': str,
|
|
'severity': str,
|
|
'reasoning': str
|
|
}
|
|
"""
|
|
# Get complaint data
|
|
title = complaint.title
|
|
description = complaint.description
|
|
complaint_category = complaint.category.name_en if complaint.category else "other"
|
|
severity = complaint.severity
|
|
priority = complaint.priority
|
|
|
|
# Build prompt for AI to generate action details
|
|
prompt = f"""Generate a PX Action from this complaint:
|
|
|
|
Complaint Title: {title}
|
|
Complaint Description: {description}
|
|
Complaint Category: {complaint_category}
|
|
Severity: {severity}
|
|
Priority: {priority}
|
|
|
|
Available PX Action Categories:
|
|
- clinical_quality: Issues related to medical care quality, diagnosis, treatment
|
|
- patient_safety: Issues that could harm patients, safety violations, risks
|
|
- service_quality: Issues with service delivery, wait times, customer service
|
|
- staff_behavior: Issues with staff professionalism, attitude, conduct
|
|
- facility: Issues with facilities, equipment, environment, cleanliness
|
|
- process_improvement: Issues with processes, workflows, procedures
|
|
- other: General issues that don't fit specific categories
|
|
|
|
Instructions:
|
|
1. Generate a clear, action-oriented title for the PX Action (max 15 words)
|
|
2. Create a detailed description that explains what needs to be done
|
|
3. Select the most appropriate PX Action category from the list above
|
|
4. Keep the same severity and priority as the complaint
|
|
5. Provide reasoning for your choices
|
|
|
|
Provide your response in JSON format:
|
|
{{
|
|
"title": "Action-oriented title (max 15 words)",
|
|
"description": "Detailed description of what needs to be done to address this complaint",
|
|
"category": "exact category name from the list above",
|
|
"priority": "low|medium|high",
|
|
"severity": "low|medium|high|critical",
|
|
"reasoning": "Brief explanation of why this category and action are appropriate"
|
|
}}"""
|
|
|
|
system_prompt = """You are a healthcare quality improvement expert.
|
|
Generate PX Actions that are actionable, specific, and focused on improvement.
|
|
The action should clearly state what needs to be done to address the complaint.
|
|
Be specific and practical in your descriptions."""
|
|
|
|
try:
|
|
response = cls.chat_completion(
|
|
prompt=prompt, system_prompt=system_prompt, response_format="json_object", temperature=0.3
|
|
)
|
|
|
|
# Parse JSON response
|
|
result = json.loads(response)
|
|
|
|
# Validate category
|
|
valid_categories = [
|
|
"clinical_quality",
|
|
"patient_safety",
|
|
"service_quality",
|
|
"staff_behavior",
|
|
"facility",
|
|
"process_improvement",
|
|
"other",
|
|
]
|
|
if result.get("category") not in valid_categories:
|
|
# Fallback: map complaint category to action category
|
|
result["category"] = cls._map_category_to_action_category(complaint_category)
|
|
|
|
# Validate severity
|
|
if result.get("severity") not in cls.SEVERITY_CHOICES:
|
|
result["severity"] = severity # Use complaint severity as fallback
|
|
|
|
# Validate priority
|
|
if result.get("priority") not in cls.PRIORITY_CHOICES:
|
|
result["priority"] = priority # Use complaint priority as fallback
|
|
|
|
logger.info(f"PX Action generated: title={result['title']}, category={result['category']}")
|
|
return result
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Failed to parse AI response: {e}")
|
|
# Return fallback based on complaint data
|
|
return {
|
|
"title": f"Address: {title}",
|
|
"description": f"Resolve the complaint: {description}",
|
|
"category": cls._map_category_to_action_category(complaint_category),
|
|
"priority": priority,
|
|
"severity": severity,
|
|
"reasoning": "AI generation failed, using complaint data as fallback",
|
|
}
|
|
except AIServiceError as e:
|
|
logger.error(f"AI service error: {e}")
|
|
# Return fallback based on complaint data
|
|
return {
|
|
"title": f"Address: {title}",
|
|
"description": f"Resolve the complaint: {description}",
|
|
"category": cls._map_category_to_action_category(complaint_category),
|
|
"priority": priority,
|
|
"severity": severity,
|
|
"reasoning": f"AI service unavailable: {str(e)}",
|
|
}
|
|
|
|
@classmethod
|
|
def _map_category_to_action_category(cls, complaint_category: str) -> str:
|
|
"""
|
|
Map complaint category to PX Action category.
|
|
|
|
Args:
|
|
complaint_category: Complaint category name
|
|
|
|
Returns:
|
|
PX Action category name
|
|
"""
|
|
# Normalize category name (lowercase, remove spaces)
|
|
category_lower = complaint_category.lower().replace(" ", "_")
|
|
|
|
# Mapping dictionary
|
|
mapping = {
|
|
# Clinical categories
|
|
"clinical": "clinical_quality",
|
|
"medical": "clinical_quality",
|
|
"diagnosis": "clinical_quality",
|
|
"treatment": "clinical_quality",
|
|
"care": "clinical_quality",
|
|
# Safety categories
|
|
"safety": "patient_safety",
|
|
"infection": "patient_safety",
|
|
"risk": "patient_safety",
|
|
"dangerous": "patient_safety",
|
|
# Service quality
|
|
"service": "service_quality",
|
|
"wait": "service_quality",
|
|
"waiting": "service_quality",
|
|
"appointment": "service_quality",
|
|
"scheduling": "service_quality",
|
|
# Staff behavior
|
|
"staff": "staff_behavior",
|
|
"behavior": "staff_behavior",
|
|
"attitude": "staff_behavior",
|
|
"rude": "staff_behavior",
|
|
"communication": "staff_behavior",
|
|
# Facility
|
|
"facility": "facility",
|
|
"environment": "facility",
|
|
"clean": "facility",
|
|
"cleanliness": "facility",
|
|
"equipment": "facility",
|
|
"room": "facility",
|
|
"bathroom": "facility",
|
|
# Process
|
|
"process": "process_improvement",
|
|
"workflow": "process_improvement",
|
|
"procedure": "process_improvement",
|
|
"policy": "process_improvement",
|
|
}
|
|
|
|
# Check for partial matches
|
|
for key, value in mapping.items():
|
|
if key in category_lower:
|
|
return value
|
|
|
|
# Default to 'other' if no match found
|
|
return "other"
|
|
|
|
@classmethod
|
|
def _detect_complaint_type(cls, text: str) -> str:
|
|
"""
|
|
Detect if the text is a complaint or appreciation using sentiment and keywords.
|
|
|
|
Args:
|
|
text: Text to analyze
|
|
|
|
Returns:
|
|
'complaint' or 'appreciation'
|
|
"""
|
|
# Keywords for appreciation (English and Arabic)
|
|
appreciation_keywords_en = [
|
|
"thank",
|
|
"thanks",
|
|
"excellent",
|
|
"great",
|
|
"wonderful",
|
|
"amazing",
|
|
"appreciate",
|
|
"commend",
|
|
"outstanding",
|
|
"fantastic",
|
|
"brilliant",
|
|
"professional",
|
|
"caring",
|
|
"helpful",
|
|
"friendly",
|
|
"good",
|
|
"nice",
|
|
"impressive",
|
|
"exceptional",
|
|
"superb",
|
|
"pleased",
|
|
"satisfied",
|
|
]
|
|
appreciation_keywords_ar = [
|
|
"شكرا",
|
|
"ممتاز",
|
|
"رائع",
|
|
"بارك",
|
|
"مدهش",
|
|
"عظيم",
|
|
"أقدر",
|
|
"شكر",
|
|
"متميز",
|
|
"مهني",
|
|
"رعاية",
|
|
"مفيد",
|
|
"ودود",
|
|
"جيد",
|
|
"لطيف",
|
|
"مبهر",
|
|
"استثنائي",
|
|
"سعيد",
|
|
"رضا",
|
|
"احترافية",
|
|
"خدمة ممتازة",
|
|
]
|
|
|
|
# Keywords for complaints (English and Arabic)
|
|
complaint_keywords_en = [
|
|
"problem",
|
|
"issue",
|
|
"complaint",
|
|
"bad",
|
|
"terrible",
|
|
"awful",
|
|
"disappointed",
|
|
"unhappy",
|
|
"poor",
|
|
"worst",
|
|
"unacceptable",
|
|
"rude",
|
|
"slow",
|
|
"delay",
|
|
"wait",
|
|
"neglect",
|
|
"ignore",
|
|
"angry",
|
|
"frustrated",
|
|
"dissatisfied",
|
|
"concern",
|
|
"worried",
|
|
]
|
|
complaint_keywords_ar = [
|
|
"مشكلة",
|
|
"مشاكل",
|
|
"سيء",
|
|
"مخيب",
|
|
"سيء للغاية",
|
|
"تعيس",
|
|
"ضعيف",
|
|
"أسوأ",
|
|
"غير مقبول",
|
|
"فظ",
|
|
"بطيء",
|
|
"تأخير",
|
|
"انتظار",
|
|
"إهمال",
|
|
"تجاهل",
|
|
"غاضب",
|
|
"محبط",
|
|
"غير راضي",
|
|
"قلق",
|
|
]
|
|
|
|
text_lower = text.lower()
|
|
|
|
# Count keyword matches
|
|
appreciation_count = 0
|
|
complaint_count = 0
|
|
|
|
for keyword in appreciation_keywords_en + appreciation_keywords_ar:
|
|
if keyword in text_lower:
|
|
appreciation_count += 1
|
|
|
|
for keyword in complaint_keywords_en + complaint_keywords_ar:
|
|
if keyword in text_lower:
|
|
complaint_count += 1
|
|
|
|
# Get sentiment analysis
|
|
try:
|
|
sentiment_result = cls.classify_sentiment(text)
|
|
sentiment = sentiment_result.get("sentiment", "neutral")
|
|
sentiment_score = sentiment_result.get("score", 0.0)
|
|
|
|
logger.info(f"Sentiment analysis: sentiment={sentiment}, score={sentiment_score}")
|
|
|
|
# If sentiment is clearly positive and has appreciation keywords
|
|
if sentiment == "positive" and sentiment_score > 0.5:
|
|
if appreciation_count >= complaint_count:
|
|
return "appreciation"
|
|
|
|
# If sentiment is clearly negative
|
|
if sentiment == "negative" and sentiment_score < -0.3:
|
|
return "complaint"
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Sentiment analysis failed, using keyword-based detection: {e}")
|
|
|
|
# Fallback to keyword-based detection
|
|
if appreciation_count > complaint_count:
|
|
return "appreciation"
|
|
elif complaint_count > appreciation_count:
|
|
return "complaint"
|
|
else:
|
|
# No clear indicators, default to complaint
|
|
return "complaint"
|
|
|
|
|
|
# Convenience singleton instance
|
|
ai_service = AIService()
|