"""Repository for AI Insight database operations.""" from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, and_, or_, desc from sqlalchemy.orm import selectinload from typing import Optional, List, Dict, Any from uuid import UUID from datetime import datetime, timedelta from app.models.ai_insight import AIInsight from app.schemas.insight import AIInsightCreate, AIInsightUpdate, InsightFilters class InsightRepository: """Repository for AI Insight operations.""" def __init__(self, session: AsyncSession): self.session = session async def create(self, insight_data: AIInsightCreate) -> AIInsight: """Create a new AI Insight.""" # Calculate expiration date (default 7 days from now) from app.core.config import settings expired_at = datetime.utcnow() + timedelta(days=settings.DEFAULT_INSIGHT_TTL_DAYS) insight = AIInsight( **insight_data.model_dump(), expired_at=expired_at ) self.session.add(insight) await self.session.flush() await self.session.refresh(insight) return insight async def get_by_id(self, insight_id: UUID) -> Optional[AIInsight]: """Get insight by ID.""" query = select(AIInsight).where(AIInsight.id == insight_id) result = await self.session.execute(query) return result.scalar_one_or_none() async def get_by_tenant( self, tenant_id: UUID, filters: Optional[InsightFilters] = None, skip: int = 0, limit: int = 100 ) -> tuple[List[AIInsight], int]: """Get insights for a tenant with filters and pagination.""" # Build base query query = select(AIInsight).where(AIInsight.tenant_id == tenant_id) # Apply filters if filters: if filters.category and filters.category != 'all': query = query.where(AIInsight.category == filters.category) if filters.priority and filters.priority != 'all': query = query.where(AIInsight.priority == filters.priority) if filters.status and filters.status != 'all': query = query.where(AIInsight.status == filters.status) if filters.actionable_only: query = query.where(AIInsight.actionable == True) if filters.min_confidence > 0: query = query.where(AIInsight.confidence >= filters.min_confidence) if filters.source_service: query = query.where(AIInsight.source_service == filters.source_service) if filters.from_date: query = query.where(AIInsight.created_at >= filters.from_date) if filters.to_date: query = query.where(AIInsight.created_at <= filters.to_date) # Get total count count_query = select(func.count()).select_from(query.subquery()) total_result = await self.session.execute(count_query) total = total_result.scalar() or 0 # Apply ordering, pagination query = query.order_by(desc(AIInsight.confidence), desc(AIInsight.created_at)) query = query.offset(skip).limit(limit) # Execute query result = await self.session.execute(query) insights = result.scalars().all() return list(insights), total async def get_orchestration_ready_insights( self, tenant_id: UUID, target_date: datetime, min_confidence: int = 70 ) -> Dict[str, List[AIInsight]]: """Get actionable insights for orchestration.""" query = select(AIInsight).where( and_( AIInsight.tenant_id == tenant_id, AIInsight.actionable == True, AIInsight.confidence >= min_confidence, AIInsight.status.in_(['new', 'acknowledged']), or_( AIInsight.expired_at.is_(None), AIInsight.expired_at > datetime.utcnow() ) ) ).order_by(desc(AIInsight.confidence)) result = await self.session.execute(query) insights = result.scalars().all() # Categorize insights categorized = { 'forecast_adjustments': [], 'procurement_recommendations': [], 'production_optimizations': [], 'supplier_alerts': [], 'price_opportunities': [] } for insight in insights: if insight.category == 'forecasting': categorized['forecast_adjustments'].append(insight) elif insight.category == 'procurement': if 'supplier' in insight.title.lower(): categorized['supplier_alerts'].append(insight) elif 'price' in insight.title.lower(): categorized['price_opportunities'].append(insight) else: categorized['procurement_recommendations'].append(insight) elif insight.category == 'production': categorized['production_optimizations'].append(insight) return categorized async def update(self, insight_id: UUID, update_data: AIInsightUpdate) -> Optional[AIInsight]: """Update an insight.""" insight = await self.get_by_id(insight_id) if not insight: return None for field, value in update_data.model_dump(exclude_unset=True).items(): setattr(insight, field, value) await self.session.flush() await self.session.refresh(insight) return insight async def delete(self, insight_id: UUID) -> bool: """Delete (dismiss) an insight.""" insight = await self.get_by_id(insight_id) if not insight: return False insight.status = 'dismissed' await self.session.flush() return True async def get_metrics(self, tenant_id: UUID) -> Dict[str, Any]: """Get aggregate metrics for insights.""" query = select(AIInsight).where( and_( AIInsight.tenant_id == tenant_id, AIInsight.status != 'dismissed', or_( AIInsight.expired_at.is_(None), AIInsight.expired_at > datetime.utcnow() ) ) ) result = await self.session.execute(query) insights = result.scalars().all() if not insights: return { 'total_insights': 0, 'actionable_insights': 0, 'average_confidence': 0, 'high_priority_count': 0, 'medium_priority_count': 0, 'low_priority_count': 0, 'critical_priority_count': 0, 'by_category': {}, 'by_status': {}, 'total_potential_impact': 0 } # Calculate metrics total = len(insights) actionable = sum(1 for i in insights if i.actionable) avg_confidence = sum(i.confidence for i in insights) / total if total > 0 else 0 # Priority counts priority_counts = { 'high': sum(1 for i in insights if i.priority == 'high'), 'medium': sum(1 for i in insights if i.priority == 'medium'), 'low': sum(1 for i in insights if i.priority == 'low'), 'critical': sum(1 for i in insights if i.priority == 'critical') } # By category by_category = {} for insight in insights: by_category[insight.category] = by_category.get(insight.category, 0) + 1 # By status by_status = {} for insight in insights: by_status[insight.status] = by_status.get(insight.status, 0) + 1 # Total potential impact total_impact = sum( float(i.impact_value) for i in insights if i.impact_value and i.impact_type in ['cost_savings', 'revenue_increase'] ) return { 'total_insights': total, 'actionable_insights': actionable, 'average_confidence': round(avg_confidence, 1), 'high_priority_count': priority_counts['high'], 'medium_priority_count': priority_counts['medium'], 'low_priority_count': priority_counts['low'], 'critical_priority_count': priority_counts['critical'], 'by_category': by_category, 'by_status': by_status, 'total_potential_impact': round(total_impact, 2) } async def expire_old_insights(self) -> int: """Mark expired insights as expired.""" query = select(AIInsight).where( and_( AIInsight.expired_at.isnot(None), AIInsight.expired_at <= datetime.utcnow(), AIInsight.status.notin_(['applied', 'dismissed', 'expired']) ) ) result = await self.session.execute(query) insights = result.scalars().all() count = 0 for insight in insights: insight.status = 'expired' count += 1 await self.session.flush() return count