from __future__ import annotations import json from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session from app import crud from app.models import Meeting, RecallWebhookEvent, Recording, Transcript, User, RecurringMeetingInvite, NonRecurringMeetingInvite from app.services.meeting_list_service import build_meeting_list ACTIVE_MEETING_STATUSES = { "pending", "live", "processing", "bot.joining_call", "bot.in_waiting_room", "bot.in_call_not_recording", "bot.in_call_recording", "recording.done", "transcript.processing", } FAILED_MEETING_STATUSES = { "abort", "not_admitted", "too_short", "bot.fatal", "recording.failed", "transcript.failed", } COMPLETED_MEETING_STATUSES = { "completed", "bot.call_ended", "bot.done", "transcript.done", } SCHEDULED_MEETING_STATUSES = {"scheduled"} STATUS_FILTER_GROUPS = { "active": ACTIVE_MEETING_STATUSES, "failed": FAILED_MEETING_STATUSES, "completed": COMPLETED_MEETING_STATUSES, "scheduled": SCHEDULED_MEETING_STATUSES, } def _safe_json_loads(value: str | None, fallback: Any) -> Any: if not value: return fallback try: return json.loads(value) except (TypeError, json.JSONDecodeError): return fallback def _expand_status_filters(status: str | None) -> list[str]: if not status: return [] expanded_statuses: list[str] = [] for raw_value in status.split(","): normalized_value = raw_value.strip() if not normalized_value: continue grouped_statuses = STATUS_FILTER_GROUPS.get(normalized_value.lower()) if grouped_statuses: expanded_statuses.extend(sorted(grouped_statuses)) else: expanded_statuses.append(normalized_value) return list(dict.fromkeys(expanded_statuses)) def _extract_participant_names(transcript: Transcript | None) -> list[str]: if not transcript or not transcript.participants: return [] participants = _safe_json_loads(transcript.participants, []) if not isinstance(participants, list): return [] return [str(name) for name in participants] def _matches_search_value( search_value: str | None, *, title: str | None, participant_names: list[str], extra_values: list[str | None] | None = None, ) -> bool: if not search_value: return True normalized_search = search_value.strip().lower() if not normalized_search: return True haystacks = [title or ""] + participant_names if extra_values: haystacks.extend(value or "" for value in extra_values) return any(normalized_search in value.lower() for value in haystacks if value) def _extract_audio_download_url(recording: Recording | None) -> str | None: if not recording: return None if recording.audio_download_url: return crud.resolve_storage_download_url(recording.audio_download_url) if not recording.recording_payload: return None payload = _safe_json_loads(recording.recording_payload, {}) if not isinstance(payload, dict): return None shortcuts = payload.get("media_shortcuts") or {} if not isinstance(shortcuts, dict): return None audio_shortcut = shortcuts.get("audio_mixed") or {} if not isinstance(audio_shortcut, dict): return None data = audio_shortcut.get("data") or {} if not isinstance(data, dict): return None return crud.resolve_storage_download_url(data.get("download_url")) def get_admin_dashboard_stats(db: Session) -> dict: total_users = db.query(func.count(User.id)).filter(User.role != "admin").scalar() or 0 approved_users = ( db.query(func.count(User.id)) .filter(User.role != "admin", User.approval_status == "approved") .scalar() or 0 ) pending_approval = ( db.query(func.count(User.id)) .filter(User.role != "admin", User.approval_status == "pending") .scalar() or 0 ) total_meetings = db.query(func.count(Meeting.id)).scalar() or 0 completed_meetings = ( db.query(func.count(Meeting.id)) .filter(Meeting.status.in_(COMPLETED_MEETING_STATUSES)) .scalar() or 0 ) active_now = ( db.query(func.count(Meeting.id)) .filter(Meeting.status.in_(ACTIVE_MEETING_STATUSES)) .scalar() or 0 ) platform_rows = ( db.query( func.coalesce(Meeting.platform, "unknown"), func.count(Meeting.id), ) .group_by(Meeting.platform) .all() ) transcript_stats = ( db.query( func.count(Transcript.id), func.coalesce(func.sum(Transcript.duration_seconds), 0), ) .filter(Transcript.duration_seconds.isnot(None), Transcript.duration_seconds > 0) .one() ) recorded_meetings_count = transcript_stats[0] or 0 total_recorded_duration_seconds = transcript_stats[1] or 0 average_duration_seconds = ( int(total_recorded_duration_seconds / recorded_meetings_count) if recorded_meetings_count else 0 ) platform_breakdown = {} for platform_name, platform_count in platform_rows: normalized_platform = platform_name or "unknown" percentage = round((platform_count / total_meetings) * 100, 2) if total_meetings else 0 platform_breakdown[normalized_platform] = percentage return { "total_users": total_users, "approved_users": approved_users, "pending_approval": pending_approval, "total_meetings": total_meetings, "completed_meetings": completed_meetings, "active_now": active_now, "platform_statistics": { "platforms": platform_breakdown, }, "recording_statistics": { "recorded_meetings_count": recorded_meetings_count, "total_recorded_duration_seconds": total_recorded_duration_seconds, "total_recorded_duration_formatted": str(timedelta(seconds=total_recorded_duration_seconds)) if total_recorded_duration_seconds else "0:00:00", "average_duration_seconds": average_duration_seconds, "average_duration_formatted": str(timedelta(seconds=average_duration_seconds)) if average_duration_seconds else "0:00:00", }, } def get_admin_recording_usage(db: Session, *, frequency: str = "daily") -> dict: normalized_frequency = (frequency or "daily").strip().lower() allowed_frequencies = {"daily", "weekly", "monthly", "yearly"} if normalized_frequency not in allowed_frequencies: raise ValueError("Invalid frequency. Use one of: daily, weekly, monthly, yearly") transcripts = ( db.query(Transcript) .filter(Transcript.duration_seconds.isnot(None), Transcript.duration_seconds > 0) .order_by(Transcript.completed_at.asc(), Transcript.created_at.asc(), Transcript.id.asc()) .all() ) bucketed_usage: dict[tuple, dict[str, Any]] = {} total_duration_seconds = 0 for transcript in transcripts: reference_datetime = transcript.completed_at or transcript.created_at if not reference_datetime: continue if reference_datetime.tzinfo is None: reference_datetime = reference_datetime.replace(tzinfo=timezone.utc) bucket_key, label, period_start = _get_recording_usage_bucket(reference_datetime, normalized_frequency) entry = bucketed_usage.setdefault( bucket_key, { "label": label, "period_start": period_start, "duration_seconds": 0, }, ) entry["duration_seconds"] += transcript.duration_seconds or 0 total_duration_seconds += transcript.duration_seconds or 0 usage_rows = [ { "label": item["label"], "period_start": item["period_start"], "duration_seconds": item["duration_seconds"], "duration_hours": round(item["duration_seconds"] / 3600, 2), } for _, item in sorted(bucketed_usage.items(), key=lambda row: row[0]) ] if normalized_frequency == "weekly": for index, item in enumerate(usage_rows, start=1): item["label"] = f"Week {index}" return { "frequency": normalized_frequency, "total_duration_seconds": total_duration_seconds, "total_duration_hours": round(total_duration_seconds / 3600, 2), "data": usage_rows, } def list_admin_meetings( db: Session, *, page: int = 1, limit: int = 12, status: str | None = None, search: str | None = None, platform: str | None = None, meeting_type: str | None = None, time_filter: str | None = None, sort_by: str | None = None, ) -> dict: now_utc = datetime.now(timezone.utc) query = db.query(Meeting, User).outerjoin(User, User.id == Meeting.user_id) if status: requested_statuses = _expand_status_filters(status) if requested_statuses: query = query.filter(Meeting.status.in_(requested_statuses)) rows = query.all() meeting_ids = [meeting.id for meeting, _ in rows] recordings_by_meeting_id: dict[int, Recording] = {} transcripts_by_meeting_id: dict[int, Transcript] = {} if meeting_ids: recordings_by_meeting_id = { recording.meeting_id: recording for recording in db.query(Recording).filter(Recording.meeting_id.in_(meeting_ids)).all() } transcripts_by_meeting_id = { transcript.meeting_id: transcript for transcript in db.query(Transcript).filter(Transcript.meeting_id.in_(meeting_ids)).all() } if search: rows = [ (meeting, user) for meeting, user in rows if _matches_search_value( search, title=meeting.title, participant_names=_extract_participant_names(transcripts_by_meeting_id.get(meeting.id)), extra_values=[user.email if user else None, user.name if user else None], ) ] filtered_listing = build_meeting_list( rows, page=page, limit=limit, max_limit=50, platform=platform, meeting_type=meeting_type, time_filter=time_filter, sort_by=sort_by, reference_datetime_getter=lambda row: row[0].scheduled_at or row[0].recall_join_at or row[0].created_at, title_getter=lambda row: row[0].title, platform_getter=lambda row: row[0].platform, is_recurring_getter=lambda row: bool(row[0].is_recurring), now_utc=now_utc, duration_seconds_getter=lambda row: _get_meeting_duration_seconds( recordings_by_meeting_id.get(row[0].id), transcripts_by_meeting_id.get(row[0].id), ), ) page_rows = filtered_listing.items response_rows = [] missing_host_guest_ids = set() meetings_needing_invite_lookup = set() for meeting, user in page_rows: if not user: found_guest_id = False if meeting.bot_payload: try: payload = json.loads(meeting.bot_payload) metadata = payload.get("metadata") or {} guest_id_str = metadata.get("guest_user_id") if guest_id_str: missing_host_guest_ids.add(int(guest_id_str)) found_guest_id = True except (ValueError, TypeError, json.JSONDecodeError): pass if not found_guest_id: meetings_needing_invite_lookup.add(meeting.id) meeting_to_guest_id_from_invite = {} if meetings_needing_invite_lookup: recurring_invites = db.query(RecurringMeetingInvite).filter(RecurringMeetingInvite.meeting_id.in_(meetings_needing_invite_lookup)).all() for inv in recurring_invites: meeting_to_guest_id_from_invite[inv.meeting_id] = inv.guest_user_id missing_host_guest_ids.add(inv.guest_user_id) non_recurring_invites = db.query(NonRecurringMeetingInvite).filter(NonRecurringMeetingInvite.meeting_id.in_(meetings_needing_invite_lookup)).all() for inv in non_recurring_invites: meeting_to_guest_id_from_invite[inv.meeting_id] = inv.guest_user_id missing_host_guest_ids.add(inv.guest_user_id) guest_users_by_id = {} if missing_host_guest_ids: guest_users = db.query(User).filter(User.id.in_(missing_host_guest_ids)).all() guest_users_by_id = {u.id: u for u in guest_users} for meeting, user in page_rows: is_guest = user is None if not user: guest_id_to_lookup = None if meeting.bot_payload: try: payload = json.loads(meeting.bot_payload) metadata = payload.get("metadata") or {} guest_id_str = metadata.get("guest_user_id") if guest_id_str: guest_id_to_lookup = int(guest_id_str) except (ValueError, TypeError, json.JSONDecodeError): pass if not guest_id_to_lookup: guest_id_to_lookup = meeting_to_guest_id_from_invite.get(meeting.id) if guest_id_to_lookup: guest_user = guest_users_by_id.get(guest_id_to_lookup) if guest_user: user = guest_user recording = recordings_by_meeting_id.get(meeting.id) transcript = transcripts_by_meeting_id.get(meeting.id) duration_seconds = 0 started_at = None stopped_at = None if recording and recording.started_at and recording.stopped_at: started_at = recording.started_at stopped_at = recording.stopped_at duration_seconds = int((stopped_at - started_at).total_seconds()) if transcript and transcript.duration_seconds: duration_seconds = transcript.duration_seconds participant_names = _extract_participant_names(transcript) response_rows.append( { "meeting_id": meeting.id, "title": meeting.title, "status": meeting.status, "platform": meeting.platform, "meet_link": meeting.meet_link, "scheduled_at": meeting.scheduled_at, "recall_join_at": meeting.recall_join_at, "created_at": meeting.created_at, "user_id": user.id if user else None, "user_email": user.email if user else None, "user_name": user.name if user else None, "is_guest": is_guest, "is_recurring": meeting.is_recurring, "participants_details": { "count": len(participant_names), "names": participant_names, }, "recording_details": { "started_at": started_at, "stopped_at": stopped_at, "duration_seconds": duration_seconds, "duration_formatted": str(timedelta(seconds=duration_seconds)) if duration_seconds else "0:00:00", }, } ) return { "pagination": filtered_listing.pagination, "data": response_rows, } def _get_meeting_duration_seconds(recording: Recording | None, transcript: Transcript | None) -> int: if transcript and transcript.duration_seconds: return transcript.duration_seconds if recording and recording.started_at and recording.stopped_at: return int((recording.stopped_at - recording.started_at).total_seconds()) return 0 def _get_recording_usage_bucket(reference_datetime: datetime, frequency: str) -> tuple[tuple, str, str]: if frequency == "weekly": iso_year, iso_week, _ = reference_datetime.isocalendar() period_start = datetime.fromisocalendar(iso_year, iso_week, 1).date().isoformat() return (iso_year, iso_week), f"{iso_year}-W{iso_week:02d}", period_start if frequency == "monthly": period_start = f"{reference_datetime.year:04d}-{reference_datetime.month:02d}-01" return (reference_datetime.year, reference_datetime.month), period_start[:7], period_start if frequency == "yearly": period_start = f"{reference_datetime.year:04d}-01-01" return (reference_datetime.year,), f"{reference_datetime.year:04d}", period_start period_start = reference_datetime.date().isoformat() return (reference_datetime.year, reference_datetime.month, reference_datetime.day), period_start, period_start def get_admin_meeting_details(db: Session, meeting_id: int) -> dict | None: meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if not meeting: return None user = db.query(User).filter(User.id == meeting.user_id).first() recording = db.query(Recording).filter(Recording.meeting_id == meeting.id).first() transcript = db.query(Transcript).filter(Transcript.meeting_id == meeting.id).first() duration_seconds = 0 if recording and recording.started_at and recording.stopped_at: duration_seconds = int((recording.stopped_at - recording.started_at).total_seconds()) if transcript and transcript.duration_seconds: duration_seconds = transcript.duration_seconds participants = _safe_json_loads(transcript.participants if transcript else None, []) if not isinstance(participants, list): participants = [] participant_names = [str(name) for name in participants] speaker_map = _safe_json_loads(transcript.speaker_map if transcript else None, {}) if not isinstance(speaker_map, dict): speaker_map = {} segments = _safe_json_loads(transcript.transcript_json if transcript else None, []) if not isinstance(segments, list): segments = [] webhook_rows = [] if meeting.recall_bot_id: webhook_rows = ( db.query(RecallWebhookEvent) .filter(RecallWebhookEvent.bot_id == meeting.recall_bot_id) .order_by(RecallWebhookEvent.created_at.desc(), RecallWebhookEvent.id.desc()) .all() ) return { "meeting": { "id": meeting.id, "title": meeting.title, "status": meeting.status, "platform": meeting.platform, "duration_seconds": duration_seconds, "duration_formatted": str(timedelta(seconds=duration_seconds)) if duration_seconds else "0:00:00", "participants_count": len(participant_names), }, "metadata": { "owner": { "id": user.id if user else None, "name": user.name if user else None, "email": user.email if user else None, }, "user_id": meeting.user_id, "created_at": meeting.created_at, "scheduled_at": meeting.scheduled_at, "bot_id": meeting.recall_bot_id, "calendar_event_id": meeting.calendar_event_id, "bot_status": meeting.bot_status, "bot_sub_code": meeting.bot_sub_code, }, "assets": { "meet_link": meeting.meet_link, "audio_recording_url": _extract_audio_download_url(recording), "video_recording_url": crud.resolve_storage_download_url(recording.video_download_url) if recording else None, "transcript": { "id": transcript.id if transcript else None, "recall_transcript_id": transcript.recall_transcript_id if transcript else None, "status": transcript.status if transcript else None, "language": transcript.language if transcript else None, "duration_seconds": transcript.duration_seconds if transcript else None, "download_url": f"/transcript-pdf/{meeting.id}" if transcript else None, "provider_data_download_url": transcript.provider_data_download_url if transcript else None, "full_text": transcript.full_text if transcript else None, "participants": participant_names, "speaker_map": speaker_map, "segments": segments, "created_at": transcript.created_at if transcript else None, "completed_at": transcript.completed_at if transcript else None, }, }, "recording": { "id": recording.id if recording else None, "status": recording.status if recording else None, "recall_recording_id": recording.recall_recording_id if recording else None, "audio_download_url": crud.resolve_storage_download_url(recording.audio_download_url) if recording else None, "video_download_url": crud.resolve_storage_download_url(recording.video_download_url) if recording else None, "started_at": recording.started_at if recording else None, "stopped_at": recording.stopped_at if recording else None, "duration_seconds": duration_seconds, "transcript_download_url": recording.transcript_download_url if recording else None, "transcript_provider_data_download_url": recording.transcript_provider_data_download_url if recording else None, "participant_events_download_url": recording.participant_events_download_url if recording else None, "speaker_timeline_download_url": recording.speaker_timeline_download_url if recording else None, "participants_download_url": recording.participants_download_url if recording else None, }, "webhook_events": [ { "id": event.id, "webhook_id": event.webhook_id, "event": event.event, "bot_id": event.bot_id, "recording_id": event.recording_id, "transcript_id": event.transcript_id, "status_code": event.status_code, "status_sub_code": event.status_sub_code, "processed_at": event.processed_at, "processing_error": event.processing_error, "created_at": event.created_at, } for event in webhook_rows ], }