imporve features
This commit is contained in:
342
services/external/app/utils/calendar_suggester.py
vendored
Normal file
342
services/external/app/utils/calendar_suggester.py
vendored
Normal file
@@ -0,0 +1,342 @@
|
||||
"""
|
||||
Calendar Suggester Utility
|
||||
|
||||
Provides intelligent school calendar suggestions based on POI detection data,
|
||||
tenant location, and heuristics optimized for bakery demand forecasting.
|
||||
"""
|
||||
|
||||
from typing import Optional, Dict, List, Any, Tuple
|
||||
from datetime import datetime, date, timezone
|
||||
import structlog
|
||||
|
||||
logger = structlog.get_logger()
|
||||
|
||||
|
||||
class CalendarSuggester:
|
||||
"""
|
||||
Suggests appropriate school calendars for tenants based on location context.
|
||||
|
||||
Uses POI detection data, proximity analysis, and bakery-specific heuristics
|
||||
to provide intelligent calendar recommendations with confidence scores.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.logger = logger
|
||||
|
||||
def suggest_calendar_for_tenant(
|
||||
self,
|
||||
city_id: str,
|
||||
available_calendars: List[Dict[str, Any]],
|
||||
poi_context: Optional[Dict[str, Any]] = None,
|
||||
tenant_data: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Suggest the most appropriate calendar for a tenant.
|
||||
|
||||
Args:
|
||||
city_id: Normalized city ID (e.g., "madrid")
|
||||
available_calendars: List of available school calendars for the city
|
||||
poi_context: Optional POI detection results including school data
|
||||
tenant_data: Optional tenant information (location, etc.)
|
||||
|
||||
Returns:
|
||||
Dict with:
|
||||
- suggested_calendar_id: UUID of suggested calendar or None
|
||||
- calendar_name: Name of suggested calendar
|
||||
- confidence: Float 0.0-1.0 confidence score
|
||||
- reasoning: List of reasoning steps
|
||||
- fallback_calendars: Alternative suggestions
|
||||
- should_assign: Boolean recommendation to auto-assign
|
||||
"""
|
||||
if not available_calendars:
|
||||
return self._no_calendars_available(city_id)
|
||||
|
||||
# Get current academic year
|
||||
academic_year = self._get_current_academic_year()
|
||||
|
||||
# Filter calendars for current academic year
|
||||
current_year_calendars = [
|
||||
cal for cal in available_calendars
|
||||
if cal.get("academic_year") == academic_year
|
||||
]
|
||||
|
||||
if not current_year_calendars:
|
||||
# Fallback to any calendar if current year not available
|
||||
current_year_calendars = available_calendars
|
||||
self.logger.warning(
|
||||
"No calendars for current academic year, using all available",
|
||||
city_id=city_id,
|
||||
academic_year=academic_year
|
||||
)
|
||||
|
||||
# Analyze POI context if available
|
||||
school_analysis = self._analyze_schools_from_poi(poi_context) if poi_context else None
|
||||
|
||||
# Apply bakery-specific heuristics
|
||||
suggestion = self._apply_suggestion_heuristics(
|
||||
current_year_calendars,
|
||||
school_analysis,
|
||||
city_id
|
||||
)
|
||||
|
||||
return suggestion
|
||||
|
||||
def _get_current_academic_year(self) -> str:
|
||||
"""
|
||||
Determine current academic year based on date.
|
||||
|
||||
Academic year runs September to June (Spain):
|
||||
- Jan-Aug: Previous year (e.g., 2024-2025)
|
||||
- Sep-Dec: Current year (e.g., 2025-2026)
|
||||
|
||||
Returns:
|
||||
Academic year string (e.g., "2024-2025")
|
||||
"""
|
||||
today = date.today()
|
||||
year = today.year
|
||||
|
||||
# Academic year starts in September
|
||||
if today.month >= 9: # September onwards
|
||||
return f"{year}-{year + 1}"
|
||||
else: # January-August
|
||||
return f"{year - 1}-{year}"
|
||||
|
||||
def _analyze_schools_from_poi(
|
||||
self,
|
||||
poi_context: Dict[str, Any]
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Analyze school POIs to infer school type preferences.
|
||||
|
||||
Args:
|
||||
poi_context: POI detection results
|
||||
|
||||
Returns:
|
||||
Dict with:
|
||||
- has_schools_nearby: Boolean
|
||||
- school_count: Int count of schools
|
||||
- nearest_distance: Float distance to nearest school (meters)
|
||||
- proximity_score: Float proximity score
|
||||
- school_names: List of detected school names
|
||||
"""
|
||||
try:
|
||||
poi_results = poi_context.get("poi_detection_results", {})
|
||||
schools_data = poi_results.get("schools", {})
|
||||
|
||||
if not schools_data:
|
||||
return None
|
||||
|
||||
school_pois = schools_data.get("pois", [])
|
||||
school_count = len(school_pois)
|
||||
|
||||
if school_count == 0:
|
||||
return None
|
||||
|
||||
# Extract school details
|
||||
school_names = [
|
||||
poi.get("name", "Unknown School")
|
||||
for poi in school_pois
|
||||
if poi.get("name")
|
||||
]
|
||||
|
||||
# Get proximity metrics
|
||||
features = schools_data.get("features", {})
|
||||
proximity_score = features.get("proximity_score", 0.0)
|
||||
|
||||
# Calculate nearest distance (approximate from POI data)
|
||||
nearest_distance = None
|
||||
if school_pois:
|
||||
# If we have POIs, estimate nearest distance
|
||||
# This is approximate - exact calculation would require tenant coords
|
||||
nearest_distance = 100.0 # Default assumption if schools detected
|
||||
|
||||
return {
|
||||
"has_schools_nearby": True,
|
||||
"school_count": school_count,
|
||||
"nearest_distance": nearest_distance,
|
||||
"proximity_score": proximity_score,
|
||||
"school_names": school_names
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self.logger.warning(
|
||||
"Failed to analyze schools from POI",
|
||||
error=str(e)
|
||||
)
|
||||
return None
|
||||
|
||||
def _apply_suggestion_heuristics(
|
||||
self,
|
||||
calendars: List[Dict[str, Any]],
|
||||
school_analysis: Optional[Dict[str, Any]],
|
||||
city_id: str
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Apply heuristics to suggest best calendar.
|
||||
|
||||
Bakery-specific heuristics:
|
||||
1. If schools detected nearby -> Prefer primary (stronger morning rush)
|
||||
2. If no schools detected -> Still suggest primary (more common, safer default)
|
||||
3. Primary schools have stronger impact on bakery traffic
|
||||
|
||||
Args:
|
||||
calendars: List of available calendars
|
||||
school_analysis: Analysis of nearby schools
|
||||
city_id: City identifier
|
||||
|
||||
Returns:
|
||||
Suggestion dict with confidence and reasoning
|
||||
"""
|
||||
reasoning = []
|
||||
confidence = 0.0
|
||||
|
||||
# Separate calendars by type
|
||||
primary_calendars = [c for c in calendars if c.get("school_type") == "primary"]
|
||||
secondary_calendars = [c for c in calendars if c.get("school_type") == "secondary"]
|
||||
other_calendars = [c for c in calendars if c.get("school_type") not in ["primary", "secondary"]]
|
||||
|
||||
# Heuristic 1: Schools detected nearby
|
||||
if school_analysis and school_analysis.get("has_schools_nearby"):
|
||||
school_count = school_analysis.get("school_count", 0)
|
||||
proximity_score = school_analysis.get("proximity_score", 0.0)
|
||||
|
||||
reasoning.append(f"Detected {school_count} schools nearby (proximity score: {proximity_score:.2f})")
|
||||
|
||||
if primary_calendars:
|
||||
suggested = primary_calendars[0]
|
||||
confidence = min(0.85, 0.65 + (proximity_score * 0.1)) # 65-85% confidence
|
||||
reasoning.append("Primary schools create strong morning rush (7:30-9am drop-off)")
|
||||
reasoning.append("Primary calendars recommended for bakeries near schools")
|
||||
elif secondary_calendars:
|
||||
suggested = secondary_calendars[0]
|
||||
confidence = 0.70
|
||||
reasoning.append("Secondary school calendars available (later morning start)")
|
||||
else:
|
||||
suggested = calendars[0]
|
||||
confidence = 0.50
|
||||
reasoning.append("Using available calendar (school type not specified)")
|
||||
|
||||
# Heuristic 2: No schools detected
|
||||
else:
|
||||
reasoning.append("No schools detected within 500m radius")
|
||||
|
||||
if primary_calendars:
|
||||
suggested = primary_calendars[0]
|
||||
confidence = 0.60 # Lower confidence without detected schools
|
||||
reasoning.append("Defaulting to primary calendar (more common, safer choice)")
|
||||
reasoning.append("Primary school holidays still affect general foot traffic")
|
||||
elif secondary_calendars:
|
||||
suggested = secondary_calendars[0]
|
||||
confidence = 0.55
|
||||
reasoning.append("Secondary calendar available as default")
|
||||
elif other_calendars:
|
||||
suggested = other_calendars[0]
|
||||
confidence = 0.50
|
||||
reasoning.append("Using available calendar")
|
||||
else:
|
||||
suggested = calendars[0]
|
||||
confidence = 0.45
|
||||
reasoning.append("No preferred calendar type available")
|
||||
|
||||
# Confidence adjustment based on school analysis quality
|
||||
if school_analysis:
|
||||
if school_analysis.get("school_count", 0) >= 3:
|
||||
confidence = min(1.0, confidence + 0.05) # Boost for multiple schools
|
||||
reasoning.append("High confidence: Multiple schools detected")
|
||||
|
||||
proximity = school_analysis.get("proximity_score", 0.0)
|
||||
if proximity > 2.0:
|
||||
confidence = min(1.0, confidence + 0.05) # Boost for close proximity
|
||||
reasoning.append("High confidence: Schools very close to bakery")
|
||||
|
||||
# Determine if we should auto-assign
|
||||
# Only auto-assign if confidence >= 75% AND schools detected
|
||||
should_auto_assign = (
|
||||
confidence >= 0.75 and
|
||||
school_analysis is not None and
|
||||
school_analysis.get("has_schools_nearby", False)
|
||||
)
|
||||
|
||||
# Build fallback suggestions
|
||||
fallback_calendars = []
|
||||
for cal in calendars:
|
||||
if cal.get("id") != suggested.get("id"):
|
||||
fallback_calendars.append({
|
||||
"calendar_id": str(cal.get("id")),
|
||||
"calendar_name": cal.get("name"),
|
||||
"school_type": cal.get("school_type"),
|
||||
"academic_year": cal.get("academic_year")
|
||||
})
|
||||
|
||||
return {
|
||||
"suggested_calendar_id": str(suggested.get("id")),
|
||||
"calendar_name": suggested.get("name"),
|
||||
"school_type": suggested.get("school_type"),
|
||||
"academic_year": suggested.get("academic_year"),
|
||||
"confidence": round(confidence, 2),
|
||||
"confidence_percentage": round(confidence * 100, 1),
|
||||
"reasoning": reasoning,
|
||||
"fallback_calendars": fallback_calendars[:2], # Top 2 alternatives
|
||||
"should_auto_assign": should_auto_assign,
|
||||
"school_analysis": school_analysis,
|
||||
"city_id": city_id
|
||||
}
|
||||
|
||||
def _no_calendars_available(self, city_id: str) -> Dict[str, Any]:
|
||||
"""Return response when no calendars available for city."""
|
||||
return {
|
||||
"suggested_calendar_id": None,
|
||||
"calendar_name": None,
|
||||
"school_type": None,
|
||||
"academic_year": None,
|
||||
"confidence": 0.0,
|
||||
"confidence_percentage": 0.0,
|
||||
"reasoning": [
|
||||
f"No school calendars configured for city: {city_id}",
|
||||
"Calendar assignment not possible at this time",
|
||||
"Location context created without calendar (can be added later)"
|
||||
],
|
||||
"fallback_calendars": [],
|
||||
"should_auto_assign": False,
|
||||
"school_analysis": None,
|
||||
"city_id": city_id
|
||||
}
|
||||
|
||||
def format_suggestion_for_admin(self, suggestion: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Format suggestion as human-readable text for admin UI.
|
||||
|
||||
Args:
|
||||
suggestion: Suggestion dict from suggest_calendar_for_tenant
|
||||
|
||||
Returns:
|
||||
Formatted string for display
|
||||
"""
|
||||
if not suggestion.get("suggested_calendar_id"):
|
||||
return f"⚠️ No calendars available for {suggestion.get('city_id', 'this city')}"
|
||||
|
||||
confidence_pct = suggestion.get("confidence_percentage", 0)
|
||||
calendar_name = suggestion.get("calendar_name", "Unknown")
|
||||
school_type = suggestion.get("school_type", "").capitalize()
|
||||
|
||||
# Confidence emoji
|
||||
if confidence_pct >= 80:
|
||||
emoji = "✅"
|
||||
elif confidence_pct >= 60:
|
||||
emoji = "📊"
|
||||
else:
|
||||
emoji = "💡"
|
||||
|
||||
text = f"{emoji} **Suggested**: {calendar_name}\n"
|
||||
text += f"**Type**: {school_type} | **Confidence**: {confidence_pct}%\n\n"
|
||||
text += "**Reasoning**:\n"
|
||||
|
||||
for reason in suggestion.get("reasoning", []):
|
||||
text += f"• {reason}\n"
|
||||
|
||||
if suggestion.get("fallback_calendars"):
|
||||
text += "\n**Alternatives**:\n"
|
||||
for alt in suggestion.get("fallback_calendars", [])[:2]:
|
||||
text += f"• {alt.get('calendar_name')} ({alt.get('school_type')})\n"
|
||||
|
||||
return text
|
||||
Reference in New Issue
Block a user