from __future__ import annotations import logging from ..tools.odoo_client import OdooClient logger = logging.getLogger(__name__) class ElearningTools: def __init__(self, odoo: OdooClient): self._o = odoo # ── Read ──────────────────────────────────────────────────────────────── async def get_courses(self, active: bool = True, limit: int = 50) -> list: domain = [('active', '=', active)] fields = ['name', 'description_short', 'website_published', 'total_slides', 'total_time', 'members_count', 'completion_rate', 'tag_ids'] return await self._o.search_read('slide.channel', domain, fields, limit=limit) async def get_course_stats(self, channel_id: int) -> dict: channels = await self._o.search_read( 'slide.channel', [('id', '=', channel_id)], ['name', 'total_slides', 'members_count', 'completion_rate', 'total_time'], limit=1, ) if not channels: return {} ch = channels[0] slides = await self._o.search_read( 'slide.slide', [('channel_id', '=', channel_id), ('active', '=', True)], ['name', 'slide_type', 'completion_rate', 'likes', 'dislikes', 'view_count'], limit=200, ) return { 'channel': ch, 'slide_count': len(slides), 'avg_slide_completion': sum(s.get('completion_rate', 0) for s in slides) / max(len(slides), 1), 'total_views': sum(s.get('view_count', 0) for s in slides), } async def get_enrolled_users(self, channel_id: int, limit: int = 100) -> list: domain = [('channel_id', '=', channel_id)] fields = ['partner_id', 'completion', 'last_activity_date', 'channel_completion'] return await self._o.search_read('slide.channel.partner', domain, fields, limit=limit) async def get_slide_completion(self, channel_id: int, min_completion: float = 0.0) -> list: return await self._o.search_read( 'slide.channel.partner', [('channel_id', '=', channel_id), ('channel_completion', '>=', min_completion)], ['partner_id', 'channel_completion', 'last_activity_date'], limit=200, ) async def get_learning_summary(self) -> dict: channels = await self._o.search_read( 'slide.channel', [('active', '=', True), ('website_published', '=', True)], ['name', 'members_count', 'completion_rate'], limit=50, ) low_completion = [c for c in channels if c.get('completion_rate', 100) < 30] return { 'total_courses': len(channels), 'total_enrollments': sum(c.get('members_count', 0) for c in channels), 'avg_completion': sum(c.get('completion_rate', 0) for c in channels) / max(len(channels), 1), 'low_completion_courses': low_completion, } # ── Create / Write ────────────────────────────────────────────────────── async def create_course(self, name: str, description: str = '', enroll_policy: str = 'public', website_published: bool = False) -> dict: """Create a new slide.channel (course). Returns {'id', 'name', 'success'}.""" vals: dict = { 'name': name, 'description_short': description, 'enroll_policy': enroll_policy, 'website_published': website_published, } result = await self._o.create('slide.channel', vals) if result.success: logger.info('elearning: created course id=%s name=%s', result.record_id, name) return {'id': result.record_id, 'name': name, 'success': True} return {'success': False, 'error': result.error} async def update_course(self, channel_id: int, name: str | None = None, description: str | None = None, enroll_policy: str | None = None) -> dict: """Update fields on an existing course.""" vals: dict = {} if name is not None: vals['name'] = name if description is not None: vals['description_short'] = description if enroll_policy is not None: vals['enroll_policy'] = enroll_policy if not vals: return {'success': False, 'error': 'No values to update'} result = await self._o.write('slide.channel', [channel_id], vals) return {'success': result.success, 'error': result.error} async def publish_course(self, channel_id: int) -> dict: """Set website_published=True on a course.""" result = await self._o.write('slide.channel', [channel_id], {'website_published': True}) return {'success': result.success, 'error': result.error} async def add_section(self, channel_id: int, name: str, sequence: int = 0) -> dict: """Add a section (category slide with is_category=True) to a course.""" result = await self._o.create('slide.slide', { 'channel_id': channel_id, 'name': name, 'is_category': True, 'sequence': sequence, }) if result.success: logger.info('elearning: added section id=%s name=%s channel=%s', result.record_id, name, channel_id) return {'id': result.record_id, 'name': name, 'success': True} return {'success': False, 'error': result.error} async def create_slide(self, channel_id: int, name: str, slide_type: str = 'document', sequence: int = 0, description: str = '', html_content: str = '', url: str = '') -> dict: """ Create a slide/lesson inside a course. slide_type choices: 'document', 'video', 'infographic', 'webpage', 'quiz' """ vals: dict = { 'channel_id': channel_id, 'name': name, 'slide_type': slide_type, 'sequence': sequence, } if description: vals['description'] = description if html_content: vals['html_content'] = html_content if url: vals['url'] = url result = await self._o.create('slide.slide', vals) if result.success: logger.info('elearning: created slide id=%s name=%s type=%s channel=%s', result.record_id, name, slide_type, channel_id) return {'id': result.record_id, 'name': name, 'slide_type': slide_type, 'success': True} return {'success': False, 'error': result.error} async def enroll_user(self, channel_id: int, partner_id: int) -> dict: """Enroll a partner in a course (creates slide.channel.partner if not already enrolled).""" existing = await self._o.search_read( 'slide.channel.partner', [('channel_id', '=', channel_id), ('partner_id', '=', partner_id)], ['id'], limit=1, ) if existing: return {'success': True, 'already_enrolled': True} result = await self._o.create('slide.channel.partner', { 'channel_id': channel_id, 'partner_id': partner_id, }) return {'success': result.success, 'error': result.error} # ── Notify ────────────────────────────────────────────────────────────── async def flag_low_completion(self, channel_id: int, reason: str) -> bool: msg = f'[AI FLAG] {reason}' await self._o.call('slide.channel', 'message_post', [[channel_id]], {'body': msg, 'message_type': 'comment'}) return True async def suggest_next_course(self, partner_id: int) -> list: completed = await self._o.search_read( 'slide.channel.partner', [('partner_id', '=', partner_id), ('channel_completion', '>=', 90)], ['channel_id'], limit=50, ) completed_ids = [ c['channel_id'][0] if isinstance(c['channel_id'], list) else c['channel_id'] for c in completed ] domain = [('active', '=', True), ('website_published', '=', True)] if completed_ids: domain.append(('id', 'not in', completed_ids)) return await self._o.search_read( 'slide.channel', domain, ['name', 'total_slides', 'completion_rate'], limit=5) async def post_chatter_note(self, model: str, record_id: int, note: str) -> bool: await self._o.call(model, 'message_post', [[record_id]], {'body': note, 'message_type': 'comment'}) return True