import json import mimetypes import os from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import urlparse import requests from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.logging_utils import get_logger from app.models import CalendarSyncState, Meeting, NonRecurringMeetingInvite, RecallWebhookEvent, Recording, RecurringMeetingCandidate, RecurringMeetingInvite, User logger = get_logger(__name__) GUEST_VISIBLE_MEETING_STATUSES = { "scheduled", "pending", "live", "processing", "bot.joining_call", "bot.in_waiting_room", "bot.in_call_not_recording", "bot.in_call_recording", "abort", "not_admitted", "completed", "bot.call_ended", "bot.done", "bot.fatal", "recording.done", "recording.failed", "transcript.done", "transcript.failed", "too_short", } def _storage_prefix() -> str: return (os.getenv("S3_FOLDER_PREFIX") or "local").strip() or "local" def _storage_client(): try: import boto3 except ImportError: logger.warning("boto3 is not installed; storage upload/presign is unavailable.") return None access_key_id = (os.getenv("AWS_ACCESS_KEY_ID") or "").strip() secret_access_key = (os.getenv("AWS_SECRET_ACCESS_KEY") or "").strip() bucket = (os.getenv("AWS_S3_BUCKET") or "").strip() if not access_key_id or not secret_access_key or not bucket: return None endpoint_url = (os.getenv("AWS_S3_ENDPOINT_URL") or "").strip() or None region_name = (os.getenv("AWS_DEFAULT_REGION") or os.getenv("AWS_REGION") or "").strip() or None if endpoint_url: parsed = urlparse(endpoint_url) if parsed.hostname and parsed.hostname.startswith(f"{bucket}."): netloc = parsed.hostname[len(f"{bucket}.") :] if parsed.port: netloc = f"{netloc}:{parsed.port}" endpoint_url = parsed._replace(netloc=netloc).geturl() return boto3.client( "s3", aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, endpoint_url=endpoint_url, region_name=region_name, ) def _presigned_url_expires_in_seconds() -> int: raw_value = (os.getenv("AWS_S3_PRESIGNED_URL_EXPIRES_SECONDS") or "3600").strip() try: return max(60, int(raw_value)) except ValueError: return 3600 def _guess_extension(source_url: str, content_type: str | None, fallback_kind: str) -> str: suffix = Path(urlparse(source_url).path).suffix if suffix: return suffix normalized_content_type = (content_type or "").split(";", 1)[0].strip().lower() guessed = mimetypes.guess_extension(normalized_content_type) if normalized_content_type else None if guessed: return guessed return ".mp3" if fallback_kind == "audio" else ".mp4" if fallback_kind == "video" else "" def _build_storage_key( *, user_id: int | None, meeting_id: int, kind: str, source_url: str, content_type: str | None, ) -> str: user_segment = str(user_id) if user_id is not None else "unknown-user" meeting_segment = str(meeting_id) extension = _guess_extension(source_url, content_type, kind) return f"{_storage_prefix()}/{user_segment}/{meeting_segment}/{kind}{extension}" def upload_recording_asset_to_storage( source_url: str | None, *, user_id: int | None, meeting_id: int, recording_id: str | None, kind: str, ) -> str | None: if not source_url: return None bucket = (os.getenv("AWS_S3_BUCKET") or "").strip() storage_client = _storage_client() if not bucket or not storage_client: return source_url try: with requests.get(source_url, stream=True, timeout=120) as response: response.raise_for_status() response.raw.decode_content = True content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip() key = _build_storage_key( user_id=user_id, meeting_id=meeting_id, kind=kind, source_url=source_url, content_type=content_type, ) extra_args: dict[str, str] = {} if content_type: extra_args["ContentType"] = content_type if extra_args: storage_client.upload_fileobj(response.raw, bucket, key, ExtraArgs=extra_args) else: storage_client.upload_fileobj(response.raw, bucket, key) return key except Exception as exc: logger.warning( "Storage upload failed for recording %s (%s). Error: %s", recording_id or "unknown-recording", kind, exc, ) return None def _is_recall_download_url(value: str | None) -> bool: if not value or not value.startswith(("http://", "https://")): return False host = (urlparse(value).hostname or "").lower() return "recall" in host def _has_bucket_backed_media(value: str | None) -> bool: if not value: return False return not _is_recall_download_url(value) def resolve_storage_download_url(stored_value: str | None) -> str | None: if not stored_value: return None if stored_value.startswith("http://") or stored_value.startswith("https://"): return stored_value bucket = (os.getenv("AWS_S3_BUCKET") or "").strip() storage_client = _storage_client() if not bucket or not storage_client: return stored_value try: return storage_client.generate_presigned_url( "get_object", Params={"Bucket": bucket, "Key": stored_value}, ExpiresIn=_presigned_url_expires_in_seconds(), ) except Exception as exc: logger.warning("Could not generate presigned URL for %s. Error: %s", stored_value, exc) return stored_value # # def create_or_update_user(db: Session, email, name, token, refresh_token: str | None = None): # user = db.query(User).filter(User.email == email).first() # if user: # user.access_token = token # if refresh_token: # user.refresh_token = refresh_token # else: # user = User( # email=email, # name=name, # access_token=token, # refresh_token=refresh_token, # calendar_sync_enabled=False, # ) # db.add(user) # # db.commit() # db.refresh(user) # return user def create_or_update_user( db: Session, email: str, name: str, token: str, refresh_token: str | None = None, calendar_scope_granted: bool = False, is_admin: bool = False, ): user = db.query(User).filter(User.email == email).first() if user: user.access_token = token user.name = name if refresh_token: user.refresh_token = refresh_token user.role = "admin" if is_admin else "user" if is_admin: user.approval_status = "approved" user.denied_at = None user.approved_at = user.approved_at or datetime.now(timezone.utc) else: user = User( email=email, name=name, access_token=token, refresh_token=refresh_token, role="admin" if is_admin else "user", approval_status="approved" if is_admin else "pending", approved_at=datetime.now(timezone.utc) if is_admin else None, calendar_sync_enabled=calendar_scope_granted, ) db.add(user) db.commit() db.refresh(user) return user def list_users(db: Session, approval_status: str | None = None): query = db.query(User).filter(User.role != "admin").order_by(User.id.asc()) if approval_status: query = query.filter(User.approval_status == approval_status) return query.all() def approve_user(db: Session, target_user_id: int, admin_user_id: int): user = db.query(User).filter(User.id == target_user_id).first() if not user: return None if user.role == "admin": return user user.approval_status = "approved" user.approved_at = datetime.now(timezone.utc) user.approved_by = admin_user_id user.denied_at = None db.commit() db.refresh(user) return user def deny_user(db: Session, target_user_id: int): user = db.query(User).filter(User.id == target_user_id).first() if not user: return None if user.role == "admin": return user user.approval_status = "denied" user.approved_at = None user.approved_by = None user.denied_at = datetime.now(timezone.utc) db.commit() db.refresh(user) return user def create_meeting_record( db: Session, user_id: int, meet_link: str, title: str = None, scheduled_at: datetime | None = None, platform: str = "google_meet", calendar_event_id: str | None = None, is_recurring: bool = False, recurring_series_id: str | None = None, source_event_id: str | None = None, host_rsvp_status: str | None = None, app_toggle_enabled: bool | None = None, ): meeting = Meeting( user_id=user_id, calendar_event_id=calendar_event_id, meet_link=meet_link, title=title, platform=platform, scheduled_at=scheduled_at, status=_initial_meeting_status(scheduled_at), is_recurring=is_recurring, recurring_series_id=recurring_series_id, source_event_id=source_event_id or calendar_event_id, host_rsvp_status=host_rsvp_status, app_toggle_enabled=app_toggle_enabled, ) db.add(meeting) db.commit() db.refresh(meeting) return meeting def create_recording(db: Session, meeting_id: int, file_path: str): recording = Recording(meeting_id=meeting_id, file_path=file_path, status="recording") db.add(recording) db.commit() db.refresh(recording) return recording def get_recording_by_meeting(db: Session, meeting_id: int): return db.query(Recording).filter(Recording.meeting_id == meeting_id).first() def get_meeting_by_recall_bot_id(db: Session, recall_bot_id: str): return db.query(Meeting).filter(Meeting.recall_bot_id == recall_bot_id).first() def get_meeting_by_id(db: Session, meeting_id: int): return db.query(Meeting).filter(Meeting.id == meeting_id).first() def get_meeting_by_meet_link_and_time(db: Session, meet_link: str, scheduled_at: datetime): # Ensure scheduled_at is in UTC for accurate comparison if scheduled_at.tzinfo is None: scheduled_at = scheduled_at.replace(tzinfo=timezone.utc) else: scheduled_at = scheduled_at.astimezone(timezone.utc) return db.query(Meeting).filter( Meeting.meet_link == meet_link, Meeting.scheduled_at == scheduled_at ).first() def get_meeting_by_calendar_event_id(db: Session, user_id: int, calendar_event_id: str): return db.query(Meeting).filter( Meeting.user_id == user_id, Meeting.calendar_event_id == calendar_event_id, ).first() def get_recording_by_recall_recording_id(db: Session, recall_recording_id: str): return db.query(Recording).filter(Recording.recall_recording_id == recall_recording_id).first() def save_calendar_meeting( db: Session, user_id: int, calendar_event_id: str, title: str | None, meet_link: str, scheduled_at: datetime, platform: str, is_recurring: bool = False, recurring_series_id: str | None = None, source_event_id: str | None = None, host_rsvp_status: str | None = None, app_toggle_enabled: bool | None = None, ): meeting = get_meeting_by_calendar_event_id(db, user_id, calendar_event_id) if not meeting: return create_meeting_record( db=db, user_id=user_id, meet_link=meet_link, title=title, scheduled_at=scheduled_at, platform=platform, calendar_event_id=calendar_event_id, is_recurring=is_recurring, recurring_series_id=recurring_series_id, source_event_id=source_event_id, host_rsvp_status=host_rsvp_status, app_toggle_enabled=app_toggle_enabled, ) meeting.title = title meeting.meet_link = meet_link meeting.scheduled_at = scheduled_at meeting.platform = platform meeting.is_recurring = is_recurring meeting.recurring_series_id = recurring_series_id meeting.source_event_id = source_event_id or calendar_event_id meeting.host_rsvp_status = host_rsvp_status if app_toggle_enabled is not None: meeting.app_toggle_enabled = app_toggle_enabled if meeting.status in {None, "pending", "scheduled"} and not meeting.recall_bot_id: meeting.status = _initial_meeting_status(scheduled_at) db.commit() db.refresh(meeting) return meeting def get_recurring_candidate_by_event_id(db: Session, user_id: int, google_event_id: str): return db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, RecurringMeetingCandidate.google_event_id == google_event_id, ).first() def get_recurring_candidate_by_linked_meeting_id(db: Session, user_id: int, meeting_id: int): return db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, RecurringMeetingCandidate.linked_meeting_id == meeting_id, ).first() def get_user_by_email(db: Session, email: str | None): if not email: return None normalized_email = email.strip().lower() return db.query(User).filter(User.email == normalized_email).first() def upsert_recurring_meeting_candidate( db: Session, user_id: int, recurring_series_id: str, google_event_id: str, title: str | None, meet_link: str, platform: str, scheduled_at: datetime, host_rsvp_status: str | None, organizer_email: str | None, app_toggle_enabled: bool | None, bot_permission_status: str, payload: dict, ): candidate = get_recurring_candidate_by_event_id(db, user_id, google_event_id) if not candidate: candidate = RecurringMeetingCandidate( user_id=user_id, recurring_series_id=recurring_series_id, google_event_id=google_event_id, ) db.add(candidate) candidate.title = title candidate.meet_link = meet_link candidate.platform = platform candidate.scheduled_at = scheduled_at candidate.host_rsvp_status = host_rsvp_status candidate.organizer_email = organizer_email if app_toggle_enabled is not None: candidate.app_toggle_enabled = app_toggle_enabled candidate.bot_permission_status = bot_permission_status candidate.payload_json = json.dumps(payload) candidate.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(candidate) return candidate def set_recurring_candidate_linked_meeting( db: Session, candidate_id: int, meeting_id: int, ): candidate = db.query(RecurringMeetingCandidate).filter(RecurringMeetingCandidate.id == candidate_id).first() if not candidate: return None candidate.linked_meeting_id = meeting_id candidate.bot_permission_status = "bot_created" db.commit() db.refresh(candidate) return candidate def update_recurring_candidate_toggle( db: Session, candidate_id: int, enabled: bool, ): candidate = db.query(RecurringMeetingCandidate).filter(RecurringMeetingCandidate.id == candidate_id).first() if not candidate: return None candidate.app_toggle_enabled = enabled candidate.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(candidate) return candidate def set_recurring_candidate_status( db: Session, candidate_id: int, bot_permission_status: str, ): candidate = db.query(RecurringMeetingCandidate).filter(RecurringMeetingCandidate.id == candidate_id).first() if not candidate: return None candidate.bot_permission_status = bot_permission_status candidate.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(candidate) return candidate def unlink_recurring_candidate_meeting( db: Session, candidate_id: int, ): candidate = db.query(RecurringMeetingCandidate).filter(RecurringMeetingCandidate.id == candidate_id).first() if not candidate: return None candidate.linked_meeting_id = None candidate.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(candidate) return candidate def delete_recurring_candidate(db: Session, user_id: int, google_event_id: str) -> bool: candidates = db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, ( (RecurringMeetingCandidate.google_event_id == google_event_id) | (RecurringMeetingCandidate.recurring_series_id == google_event_id) ), ).all() if not candidates: return False deleted_any = False for candidate in candidates: if candidate.linked_meeting_id: meeting = db.query(Meeting).filter(Meeting.id == candidate.linked_meeting_id).first() if meeting: recording = get_recording_by_meeting(db, meeting.id) transcript = get_transcript_by_meeting(db, meeting.id) completed_statuses = { "completed", "bot.call_ended", "bot.done", "recording.done", "transcript.done", "too_short", "abort", "not_admitted", } if recording or transcript or meeting.status in completed_statuses: continue db.delete(candidate) deleted_any = True if deleted_any: db.commit() return deleted_any def cleanup_recurring_candidates( db: Session, user_id: int, window_start: datetime, window_end: datetime, ): stale_candidates = db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, ( (RecurringMeetingCandidate.scheduled_at < window_start - timedelta(days=1)) | (RecurringMeetingCandidate.scheduled_at > window_end) ), ).all() removed = 0 for candidate in stale_candidates: if candidate.linked_meeting_id: continue db.delete(candidate) removed += 1 if removed: db.commit() return removed def refresh_recurring_candidate_visibility( db: Session, user_id: int, window_start: datetime, window_end: datetime, ): candidates = db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, RecurringMeetingCandidate.scheduled_at >= window_start, RecurringMeetingCandidate.scheduled_at <= window_end, ).order_by(RecurringMeetingCandidate.recurring_series_id.asc(), RecurringMeetingCandidate.scheduled_at.asc()).all() blocking_statuses = { "scheduled", "pending", "live", "processing", "bot.joining_call", "bot.in_waiting_room", "bot.in_call_not_recording", "bot.in_call_recording", "transcript.processing", } def candidate_should_block_series(candidate: RecurringMeetingCandidate) -> bool: if not candidate.linked_meeting_id: return False linked_meeting = get_meeting_by_id(db, candidate.linked_meeting_id) if not linked_meeting: return False scheduled_at = linked_meeting.scheduled_at or candidate.scheduled_at if scheduled_at and scheduled_at.tzinfo is None: scheduled_at = scheduled_at.replace(tzinfo=timezone.utc) if linked_meeting.status in blocking_statuses: return True return bool(scheduled_at and scheduled_at >= window_start) first_visible_per_series: set[str] = set() changed = False for candidate in candidates: if candidate.recurring_series_id in first_visible_per_series: should_show = False else: should_show = not candidate_should_block_series(candidate) if should_show or candidate_should_block_series(candidate): first_visible_per_series.add(candidate.recurring_series_id) if candidate.dashboard_visible != should_show: candidate.dashboard_visible = should_show changed = True if changed: db.commit() return candidates def list_visible_recurring_candidates(db: Session, user_id: int, now_utc: datetime, window_end: datetime): return db.query(RecurringMeetingCandidate).filter( RecurringMeetingCandidate.user_id == user_id, RecurringMeetingCandidate.dashboard_visible.is_(True), RecurringMeetingCandidate.scheduled_at >= now_utc, RecurringMeetingCandidate.scheduled_at <= window_end, ).order_by(RecurringMeetingCandidate.scheduled_at.asc()).all() def get_recurring_meeting_invite(db: Session, guest_user_id: int, google_event_id: str): return db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, RecurringMeetingInvite.google_event_id == google_event_id, ).first() def upsert_recurring_meeting_invite( db: Session, host_user_id: int | None, guest_user_id: int, recurring_series_id: str, google_event_id: str, title: str | None, meet_link: str, platform: str, scheduled_at: datetime, guest_rsvp_status: str | None, organizer_email: str | None, meeting_id: int | None = None, ): invite = get_recurring_meeting_invite(db, guest_user_id, google_event_id) if not invite: invite = RecurringMeetingInvite( guest_user_id=guest_user_id, google_event_id=google_event_id, ) db.add(invite) invite.host_user_id = host_user_id invite.recurring_series_id = recurring_series_id invite.title = title invite.meet_link = meet_link invite.platform = platform invite.scheduled_at = scheduled_at invite.guest_rsvp_status = guest_rsvp_status invite.organizer_email = organizer_email if meeting_id is not None: invite.meeting_id = meeting_id invite.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(invite) return invite def link_recurring_meeting_invites_to_meeting( db: Session, host_user_id: int, google_event_id: str, meeting_id: int, ): invites = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.host_user_id == host_user_id, RecurringMeetingInvite.google_event_id == google_event_id, ).all() updated = 0 for invite in invites: if invite.meeting_id == meeting_id: continue invite.meeting_id = meeting_id invite.updated_at = datetime.now(timezone.utc) updated += 1 if updated: db.commit() return updated def delete_recurring_meeting_invite(db: Session, guest_user_id: int, google_event_id: str) -> bool: invites = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, ( (RecurringMeetingInvite.google_event_id == google_event_id) | (RecurringMeetingInvite.recurring_series_id == google_event_id) ), ).all() if not invites: return False deleted_any = False for invite in invites: if recurring_invite_has_history(db, invite): logger.info( "Recurring meeting invite deletion was skipped for invite %s because meeting history already exists.", invite.id, ) continue db.delete(invite) deleted_any = True if deleted_any: db.commit() return deleted_any def cleanup_recurring_meeting_invites( db: Session, guest_user_id: int, window_start: datetime, window_end: datetime, ): stale_invites = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, ( (RecurringMeetingInvite.scheduled_at < window_start - timedelta(days=1)) | (RecurringMeetingInvite.scheduled_at > window_end) ), ).all() removed = 0 for invite in stale_invites: if recurring_invite_has_history(db, invite): continue db.delete(invite) removed += 1 if removed: db.commit() return removed def list_visible_recurring_meeting_invites(db: Session, guest_user_id: int, now_utc: datetime, window_end: datetime): invites = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, ).order_by( RecurringMeetingInvite.recurring_series_id.asc(), RecurringMeetingInvite.scheduled_at.asc(), ).all() visible: list[RecurringMeetingInvite] = [] first_visible_per_series: set[str] = set() for invite in invites: has_history = recurring_invite_has_history(db, invite) is_upcoming = now_utc <= invite.scheduled_at <= window_end if has_history: visible.append(invite) continue if not is_upcoming: continue if invite.recurring_series_id in first_visible_per_series: continue first_visible_per_series.add(invite.recurring_series_id) visible.append(invite) return visible def get_accessible_guest_recurring_meeting(db: Session, guest_user_id: int, meeting_id: int): invite = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, RecurringMeetingInvite.meeting_id == meeting_id, ).first() if not invite: return None return get_meeting_by_id(db, meeting_id) def get_non_recurring_meeting_invite(db: Session, guest_user_id: int, google_event_id: str): return db.query(NonRecurringMeetingInvite).filter( NonRecurringMeetingInvite.guest_user_id == guest_user_id, NonRecurringMeetingInvite.google_event_id == google_event_id, ).first() def upsert_non_recurring_meeting_invite( db: Session, host_user_id: int | None, guest_user_id: int, google_event_id: str, title: str | None, meet_link: str, platform: str, scheduled_at: datetime, guest_rsvp_status: str | None, organizer_email: str | None, meeting_id: int | None = None, ): invite = get_non_recurring_meeting_invite(db, guest_user_id, google_event_id) if not invite: invite = NonRecurringMeetingInvite( guest_user_id=guest_user_id, google_event_id=google_event_id, ) db.add(invite) invite.host_user_id = host_user_id invite.title = title invite.meet_link = meet_link invite.platform = platform invite.scheduled_at = scheduled_at invite.guest_rsvp_status = guest_rsvp_status invite.organizer_email = organizer_email if meeting_id is not None: invite.meeting_id = meeting_id invite.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(invite) return invite def link_non_recurring_meeting_invites_to_meeting( db: Session, host_user_id: int, google_event_id: str, meeting_id: int, ): invites = db.query(NonRecurringMeetingInvite).filter( NonRecurringMeetingInvite.host_user_id == host_user_id, NonRecurringMeetingInvite.google_event_id == google_event_id, ).all() updated = 0 for invite in invites: if invite.meeting_id == meeting_id: continue invite.meeting_id = meeting_id invite.updated_at = datetime.now(timezone.utc) updated += 1 if updated: db.commit() return updated def delete_non_recurring_meeting_invite(db: Session, guest_user_id: int, google_event_id: str) -> bool: invite = get_non_recurring_meeting_invite(db, guest_user_id, google_event_id) if not invite: return False if non_recurring_invite_has_history(db, invite): logger.info( "Non-recurring meeting invite deletion was skipped for invite %s because meeting history already exists.", invite.id, ) return False db.delete(invite) db.commit() return True def cleanup_non_recurring_meeting_invites( db: Session, guest_user_id: int, window_start: datetime, window_end: datetime, ): stale_invites = db.query(NonRecurringMeetingInvite).filter( NonRecurringMeetingInvite.guest_user_id == guest_user_id, ( (NonRecurringMeetingInvite.scheduled_at < window_start - timedelta(days=1)) | (NonRecurringMeetingInvite.scheduled_at > window_end) ), ).all() removed = 0 for invite in stale_invites: if non_recurring_invite_has_history(db, invite): continue db.delete(invite) removed += 1 if removed: db.commit() return removed def list_visible_non_recurring_meeting_invites(db: Session, guest_user_id: int, now_utc: datetime, window_end: datetime): invites = db.query(NonRecurringMeetingInvite).filter( NonRecurringMeetingInvite.guest_user_id == guest_user_id, ).order_by( NonRecurringMeetingInvite.scheduled_at.asc(), NonRecurringMeetingInvite.id.asc(), ).all() visible: list[NonRecurringMeetingInvite] = [] for invite in invites: has_history = non_recurring_invite_has_history(db, invite) is_upcoming = now_utc <= invite.scheduled_at <= window_end if has_history or is_upcoming: visible.append(invite) return visible def get_host_meeting_for_non_recurring_invite(db: Session, invite: NonRecurringMeetingInvite): if invite.meeting_id: return get_meeting_by_id(db, invite.meeting_id) meeting = db.query(Meeting).filter( Meeting.user_id == invite.host_user_id, ( (Meeting.calendar_event_id == invite.google_event_id) | (Meeting.source_event_id == invite.google_event_id) ), ).first() if meeting and invite.meeting_id != meeting.id: invite.meeting_id = meeting.id invite.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(invite) return meeting def non_recurring_invite_has_history(db: Session, invite: NonRecurringMeetingInvite) -> bool: meeting = get_host_meeting_for_non_recurring_invite(db, invite) if not meeting: return False recording = get_recording_by_meeting(db, meeting.id) transcript = get_transcript_by_meeting(db, meeting.id) return bool(recording or transcript or meeting.status in GUEST_VISIBLE_MEETING_STATUSES) def get_accessible_guest_meeting(db: Session, guest_user_id: int, meeting_id: int): recurring_invite = db.query(RecurringMeetingInvite).filter( RecurringMeetingInvite.guest_user_id == guest_user_id, RecurringMeetingInvite.meeting_id == meeting_id, ).first() if recurring_invite: return get_meeting_by_id(db, meeting_id) non_recurring_invite = db.query(NonRecurringMeetingInvite).filter( NonRecurringMeetingInvite.guest_user_id == guest_user_id, NonRecurringMeetingInvite.meeting_id == meeting_id, ).first() if non_recurring_invite: return get_meeting_by_id(db, meeting_id) return None def get_host_meeting_for_recurring_invite(db: Session, invite: RecurringMeetingInvite): if invite.meeting_id: return get_meeting_by_id(db, invite.meeting_id) meeting = db.query(Meeting).filter( Meeting.user_id == invite.host_user_id, Meeting.is_recurring.is_(True), ( (Meeting.calendar_event_id == invite.google_event_id) | (Meeting.source_event_id == invite.google_event_id) ), ).first() if meeting and invite.meeting_id != meeting.id: invite.meeting_id = meeting.id invite.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(invite) return meeting def recurring_invite_has_history(db: Session, invite: RecurringMeetingInvite) -> bool: meeting = get_host_meeting_for_recurring_invite(db, invite) if not meeting: return False recording = get_recording_by_meeting(db, meeting.id) transcript = get_transcript_by_meeting(db, meeting.id) return bool(recording or transcript or meeting.status in GUEST_VISIBLE_MEETING_STATUSES) def unlink_recurring_meeting_invites_from_meeting(db: Session, meeting_id: int): invites = ( db.query(RecurringMeetingInvite) .filter(RecurringMeetingInvite.meeting_id == meeting_id) .all() ) if not invites: return 0 for invite in invites: invite.meeting_id = None db.commit() return len(invites) def unlink_non_recurring_meeting_invites_from_meeting(db: Session, meeting_id: int): invites = ( db.query(NonRecurringMeetingInvite) .filter(NonRecurringMeetingInvite.meeting_id == meeting_id) .all() ) if not invites: return 0 for invite in invites: invite.meeting_id = None db.commit() return len(invites) def count_accepted_guests_for_meeting(db: Session, meeting_id: int) -> int: recurring_count = ( db.query(RecurringMeetingInvite) .filter( RecurringMeetingInvite.meeting_id == meeting_id, RecurringMeetingInvite.guest_rsvp_status == "accepted", ) .count() ) non_recurring_count = ( db.query(NonRecurringMeetingInvite) .filter( NonRecurringMeetingInvite.meeting_id == meeting_id, NonRecurringMeetingInvite.guest_rsvp_status == "accepted", ) .count() ) return recurring_count + non_recurring_count def delete_calendar_meeting(db: Session, user_id: int, calendar_event_id: str) -> bool: meetings = db.query(Meeting).filter( Meeting.user_id == user_id, ( (Meeting.calendar_event_id == calendar_event_id) | (Meeting.recurring_series_id == calendar_event_id) ), ).all() if not meetings: return False completed_statuses = { "completed", "bot.call_ended", "bot.done", "recording.done", "transcript.done", "too_short", "abort", "not_admitted" } deleted_any = False for meeting in meetings: recording = get_recording_by_meeting(db, meeting.id) transcript = get_transcript_by_meeting(db, meeting.id) if recording or transcript or meeting.status in completed_statuses: logger.info( "Calendar meeting deletion was skipped for meeting %s because history already exists. Status: %s. Recording exists: %s. Transcript exists: %s.", meeting.id, meeting.status, bool(recording), bool(transcript), ) continue detached_invites_count = unlink_recurring_meeting_invites_from_meeting(db, meeting.id) if detached_invites_count: logger.info( "Recurring meeting invites were detached from meeting %s before calendar meeting deletion. Detached invites: %s.", meeting.id, detached_invites_count, ) detached_non_recurring_invites_count = unlink_non_recurring_meeting_invites_from_meeting(db, meeting.id) if detached_non_recurring_invites_count: logger.info( "Non-recurring meeting invites were detached from meeting %s before calendar meeting deletion. Detached invites: %s.", meeting.id, detached_non_recurring_invites_count, ) # Detach from recurring meeting candidates to avoid foreign key violations candidates = db.query(RecurringMeetingCandidate).filter(RecurringMeetingCandidate.linked_meeting_id == meeting.id).all() for candidate in candidates: candidate.linked_meeting_id = None if candidates: logger.info( "Recurring meeting candidates were detached from meeting %s before calendar meeting deletion. Detached candidates: %s.", meeting.id, len(candidates), ) db.delete(meeting) deleted_any = True if deleted_any: db.commit() return deleted_any def get_calendar_sync_state(db: Session, user_id: int): return db.query(CalendarSyncState).filter(CalendarSyncState.user_id == user_id).first() def get_calendar_sync_state_by_channel(db: Session, channel_id: str): return db.query(CalendarSyncState).filter(CalendarSyncState.channel_id == channel_id).first() def set_calendar_sync_enabled(db: Session, user_id: int, enabled: bool): user = db.query(User).filter(User.id == user_id).first() if not user: return None user.calendar_sync_enabled = enabled db.commit() db.refresh(user) return user def save_calendar_sync_state( db: Session, user_id: int, calendar_id: str = "primary", sync_token: str | None = None, webhook_url: str | None = None, channel_id: str | None = None, channel_token: str | None = None, resource_id: str | None = None, resource_uri: str | None = None, expires_at: datetime | None = None, last_message_number: str | None = None, ): state = get_calendar_sync_state(db, user_id) if not state: state = CalendarSyncState(user_id=user_id, calendar_id=calendar_id) db.add(state) state.calendar_id = calendar_id if sync_token is not None: state.sync_token = sync_token if webhook_url is not None: state.webhook_url = webhook_url if channel_id is not None: state.channel_id = channel_id if channel_token is not None: state.channel_token = channel_token if resource_id is not None: state.resource_id = resource_id if resource_uri is not None: state.resource_uri = resource_uri if expires_at is not None: state.expires_at = expires_at if last_message_number is not None: state.last_message_number = last_message_number state.last_synced_at = datetime.now(timezone.utc) db.commit() db.refresh(state) return state TERMINAL_MEETING_STATUSES = {"not_admitted", "abort"} def mark_meeting_completed(db: Session, meeting_id: int): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting and meeting.status not in TERMINAL_MEETING_STATUSES: meeting.status = "processing" db.commit() db.refresh(meeting) return meeting def mark_meeting_abort(db: Session, meeting_id: int): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "abort" db.commit() db.refresh(meeting) return meeting def mark_recording_stopped(db: Session, meeting_id: int): recording = get_recording_by_meeting(db, meeting_id) if recording: recording.status = "stopped" recording.stopped_at = datetime.now(timezone.utc) meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting and meeting.status not in TERMINAL_MEETING_STATUSES: meeting.status = "processing" db.commit() if recording: db.refresh(recording) return recording def mark_meeting_live(db: Session, meeting_id: int): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "live" db.commit() db.refresh(meeting) return meeting def mark_meeting_processing(db: Session, meeting_id: int): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting and meeting.status not in TERMINAL_MEETING_STATUSES: meeting.status = "processing" db.commit() db.refresh(meeting) return meeting def mark_meeting_not_admitted(db: Session, meeting_id: int): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "not_admitted" db.commit() db.refresh(meeting) return meeting def create_transcript(db: Session, meeting_id: int): from app.models import Transcript existing = db.query(Transcript).filter(Transcript.meeting_id == meeting_id).first() if existing: return existing transcript = Transcript(meeting_id=meeting_id, status="pending") db.add(transcript) db.commit() db.refresh(transcript) return transcript def get_transcript_by_meeting(db: Session, meeting_id: int): from app.models import Transcript return db.query(Transcript).filter(Transcript.meeting_id == meeting_id).first() def get_transcript_by_recall_transcript_id(db: Session, recall_transcript_id: str): from app.models import Transcript return db.query(Transcript).filter(Transcript.recall_transcript_id == recall_transcript_id).first() def update_transcript_result(db: Session, meeting_id: int, result: dict): from app.models import Transcript transcript = db.query(Transcript).filter(Transcript.meeting_id == meeting_id).first() if not transcript: return None if result.get("error") == "too_short": transcript.status = "too_short" transcript.error_message = result["message"] transcript.duration_seconds = result.get("duration_seconds") meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "too_short" elif result.get("error"): transcript.status = "failed" transcript.error_message = str(result["error"]) meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "completed" else: transcript.status = "done" transcript.language = result.get("language") transcript.duration_seconds = result.get("duration_seconds") transcript.participants = json.dumps(result.get("participants", [])) transcript.speaker_map = json.dumps(result.get("speaker_map", {})) transcript.transcript_json = json.dumps(result.get("transcript", [])) transcript.full_text = result.get("full_text") transcript.completed_at = datetime.now(timezone.utc) meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "completed" db.commit() db.refresh(transcript) return transcript def set_meeting_recall_bot(db: Session, meeting_id: int, bot_data: dict): meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if not meeting: return None meeting.recall_bot_id = bot_data.get("id") meeting.bot_status = extract_recall_bot_status(bot_data) meeting.bot_payload = json.dumps(bot_data) join_at = bot_data.get("join_at") if join_at: meeting.recall_join_at = parse_iso_datetime(join_at) mapped_status = map_recall_bot_status_to_meeting_status(meeting.bot_status) if mapped_status: meeting.status = mapped_status elif meeting.recall_join_at and meeting.recall_join_at > datetime.now(timezone.utc): meeting.bot_status = "scheduled" meeting.status = "scheduled" db.commit() db.refresh(meeting) logger.info( "Recall bot details were saved for meeting %s. Bot id: %s. Bot status: %s. Meeting status: %s. Bot sub code: %s.", meeting.id, meeting.recall_bot_id, meeting.bot_status, meeting.status, meeting.bot_sub_code, ) return meeting def sync_meeting_from_recall_bot(db: Session, meeting: Meeting, bot_data: dict): meeting.bot_status = extract_recall_bot_status(bot_data) new_sub_code = _extract_recall_bot_sub_code(bot_data) if new_sub_code: meeting.bot_sub_code = new_sub_code meeting.bot_payload = json.dumps(bot_data) join_at = bot_data.get("join_at") if join_at: meeting.recall_join_at = parse_iso_datetime(join_at) mapped_status = map_recall_bot_status_to_meeting_status(meeting.bot_status, payload=bot_data) if mapped_status: if meeting.status in TERMINAL_MEETING_STATUSES and meeting.status != mapped_status: pass else: meeting.status = mapped_status elif meeting.recall_join_at and meeting.recall_join_at > datetime.now(timezone.utc): meeting.bot_status = "scheduled" meeting.status = "scheduled" db.commit() db.refresh(meeting) logger.info( "Meeting %s was synced from Recall AI. Bot id: %s. Bot status: %s. Meeting status: %s. Bot sub code: %s.", meeting.id, meeting.recall_bot_id, meeting.bot_status, meeting.status, meeting.bot_sub_code, ) return meeting def upsert_recording_from_recall( db: Session, meeting_id: int, recording_data: dict | None, mark_links_refreshed: bool = False, ): if not recording_data: logger.info("Recording sync was skipped for meeting %s because Recall AI did not return recording data.", meeting_id) return None shortcuts = recording_data.get("media_shortcuts") or {} audio_mixed = shortcuts.get("audio_mixed") or {} video_mixed = shortcuts.get("video_mixed") or {} transcript_shortcut = shortcuts.get("transcript") or {} participant_events = shortcuts.get("participant_events") or {} audio_download_url = ((audio_mixed.get("data") or {}).get("download_url")) video_download_url = ((video_mixed.get("data") or {}).get("download_url")) meeting = get_meeting_by_id(db, meeting_id) meeting_user_id = meeting.user_id if meeting else None if meeting_user_id is None and meeting and meeting.bot_payload: try: payload = json.loads(meeting.bot_payload) metadata = payload.get("metadata") or {} guest_user_id = metadata.get("guest_user_id") if guest_user_id: meeting_user_id = int(guest_user_id) except (ValueError, TypeError, json.JSONDecodeError): pass def apply_recording_fields(recording: Recording) -> Recording: recording.recall_recording_id = recording_data.get("id") recording.status = ((recording_data.get("status") or {}).get("code")) or recording.status or "pending" recording.started_at = parse_iso_datetime(recording_data.get("started_at")) or recording.started_at recording.stopped_at = parse_iso_datetime(recording_data.get("completed_at")) or recording.stopped_at if not _has_bucket_backed_media(recording.audio_download_url): recording.audio_download_url = upload_recording_asset_to_storage( audio_download_url, user_id=meeting_user_id, meeting_id=meeting_id, recording_id=recording_data.get("id"), kind="audio", ) if not _has_bucket_backed_media(recording.video_download_url): recording.video_download_url = upload_recording_asset_to_storage( video_download_url, user_id=meeting_user_id, meeting_id=meeting_id, recording_id=recording_data.get("id"), kind="video", ) recording.transcript_download_url = ((transcript_shortcut.get("data") or {}).get("download_url")) recording.transcript_provider_data_download_url = ((transcript_shortcut.get("data") or {}).get("provider_data_download_url")) recording.participant_events_download_url = ((participant_events.get("data") or {}).get("participant_events_download_url")) recording.speaker_timeline_download_url = ((participant_events.get("data") or {}).get("speaker_timeline_download_url")) recording.participants_download_url = ((participant_events.get("data") or {}).get("participants_download_url")) recording.recording_payload = json.dumps(recording_data) if mark_links_refreshed: recording.download_urls_refreshed_at = datetime.now(timezone.utc) return recording recording = get_recording_by_meeting(db, meeting_id) if not recording: recording = Recording(meeting_id=meeting_id) db.add(recording) apply_recording_fields(recording) try: db.commit() except IntegrityError: db.rollback() recording = get_recording_by_meeting(db, meeting_id) if not recording: raise apply_recording_fields(recording) db.commit() db.refresh(recording) logger.info( "Recording details were saved for meeting %s. Recording id: %s. Recording status: %s.", meeting_id, recording.recall_recording_id, recording.status, ) return recording def save_recall_transcript_request(db: Session, meeting_id: int, transcript_data: dict): from app.models import Transcript transcript = get_transcript_by_meeting(db, meeting_id) if not transcript: transcript = Transcript(meeting_id=meeting_id) db.add(transcript) transcript.recall_transcript_id = transcript_data.get("id") transcript.status = ((transcript_data.get("status") or {}).get("code")) or "processing" transcript.download_url = ((transcript_data.get("data") or {}).get("download_url")) transcript.provider_data_download_url = ((transcript_data.get("data") or {}).get("provider_data_download_url")) transcript.language = ((((transcript_data.get("provider") or {}).get("recallai_async") or {}).get("language_code"))) transcript.transcript_payload = json.dumps(transcript_data) transcript.error_message = None transcript.completed_at = None db.commit() db.refresh(transcript) logger.info( "Transcript creation request was saved for meeting %s. Transcript id: %s. Status: %s.", meeting_id, transcript.recall_transcript_id, transcript.status, ) return transcript def sync_recall_transcript_details(db: Session, meeting_id: int, transcript_data: dict): transcript = get_transcript_by_meeting(db, meeting_id) if not transcript: return save_recall_transcript_request(db, meeting_id, transcript_data) transcript.recall_transcript_id = transcript_data.get("id") transcript.status = ((transcript_data.get("status") or {}).get("code")) or transcript.status transcript.download_url = ((transcript_data.get("data") or {}).get("download_url")) transcript.provider_data_download_url = ((transcript_data.get("data") or {}).get("provider_data_download_url")) transcript.language = ((((transcript_data.get("provider") or {}).get("recallai_async") or {}).get("language_code"))) transcript.transcript_payload = json.dumps(transcript_data) if transcript.status == "done": transcript.completed_at = parse_iso_datetime((transcript_data.get("status") or {}).get("updated_at")) or datetime.now(timezone.utc) meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "transcript.done" elif transcript.status in {"failed", "error"}: transcript.error_message = (transcript_data.get("status") or {}).get("sub_code") or "Recall AI transcript failed" meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "transcript.failed" elif transcript.status == "processing": meeting = db.query(Meeting).filter(Meeting.id == meeting_id).first() if meeting: meeting.status = "transcript.processing" db.commit() db.refresh(transcript) logger.info( "Transcript details were synced for meeting %s. Transcript id: %s. Status: %s.", meeting_id, transcript.recall_transcript_id, transcript.status, ) return transcript def save_recall_transcript_content( db: Session, meeting_id: int, normalized_payload: dict, ): transcript = get_transcript_by_meeting(db, meeting_id) if not transcript: transcript = create_transcript(db, meeting_id) segments = normalized_payload.get("segments") or [] participants = normalized_payload.get("participants") or [] speaker_map = normalized_payload.get("speaker_map") or {} full_text = normalized_payload.get("full_text") or "" transcript.transcript_json = json.dumps(segments) transcript.participants = json.dumps(participants) transcript.speaker_map = json.dumps(speaker_map) transcript.full_text = full_text max_end = None for segment in segments: end_value = segment.get("end") if isinstance(end_value, (int, float)): max_end = max(max_end or 0, int(end_value)) if max_end is not None: transcript.duration_seconds = max_end db.commit() db.refresh(transcript) logger.info( "Transcript content was saved for meeting %s. Segments: %s. Participants: %s.", meeting_id, len(segments), len(participants), ) return transcript def recall_transcript_already_requested(db: Session, meeting_id: int) -> bool: transcript = get_transcript_by_meeting(db, meeting_id) if not transcript: return False if transcript.recall_transcript_id: return True return transcript.status in {"processing", "done"} def create_recall_webhook_event( db: Session, webhook_id: str, event: str, payload: dict, bot_id: str | None = None, recording_id: str | None = None, transcript_id: str | None = None, status_code: str | None = None, status_sub_code: str | None = None, ): existing = db.query(RecallWebhookEvent).filter(RecallWebhookEvent.webhook_id == webhook_id).first() if existing: logger.info("Recall webhook %s was ignored because it was already stored. Event: %s.", webhook_id, event) return existing, False webhook_event = RecallWebhookEvent( webhook_id=webhook_id, event=event, bot_id=bot_id, recording_id=recording_id, transcript_id=transcript_id, status_code=status_code, status_sub_code=status_sub_code, payload=json.dumps(payload), ) db.add(webhook_event) db.commit() db.refresh(webhook_event) logger.info( "Recall webhook %s was stored. Event: %s. Bot id: %s. Recording id: %s. Transcript id: %s. Status code: %s. Sub code: %s.", webhook_id, event, bot_id, recording_id, transcript_id, status_code, status_sub_code, ) return webhook_event, True def get_recall_webhook_event(db: Session, webhook_event_id: int): return db.query(RecallWebhookEvent).filter(RecallWebhookEvent.id == webhook_event_id).first() def get_latest_recall_status_sub_code(db: Session, bot_id: str | None): if not bot_id: return None webhook_event = ( db.query(RecallWebhookEvent) .filter(RecallWebhookEvent.bot_id == bot_id) .filter(RecallWebhookEvent.status_sub_code.isnot(None)) .order_by(RecallWebhookEvent.id.desc()) .first() ) if not webhook_event: return None return webhook_event.status_sub_code def mark_recall_webhook_processed(db: Session, webhook_event: RecallWebhookEvent): webhook_event.processed_at = datetime.now(timezone.utc) webhook_event.processing_error = None db.commit() db.refresh(webhook_event) logger.info("Recall webhook %s finished processing. Event: %s.", webhook_event.webhook_id, webhook_event.event) return webhook_event def mark_recall_webhook_failed(db: Session, webhook_event: RecallWebhookEvent, error_message: str): webhook_event.processing_error = error_message db.commit() db.refresh(webhook_event) logger.error( "Recall webhook %s failed during processing. Event: %s. Error: %s.", webhook_event.webhook_id, webhook_event.event, error_message, ) return webhook_event def save_meeting_bot_sub_code(db: Session, meeting: Meeting, sub_code: str | None): if not sub_code: return meeting meeting.bot_sub_code = sub_code db.commit() db.refresh(meeting) logger.info( "Meeting %s bot sub code was updated from Recall AI. Incoming sub code: %s. Result bot_sub_code: %s.", meeting.id, sub_code, meeting.bot_sub_code, ) return meeting def set_meeting_status_from_recall_code(db: Session, meeting: Meeting, status_code: str | None, payload: dict | None = None): if payload is not None: meeting.bot_payload = json.dumps(payload) meeting.bot_status = status_code mapped_status = map_recall_bot_status_to_meeting_status(status_code, payload=payload) if mapped_status: meeting.status = mapped_status db.commit() db.refresh(meeting) logger.info( "Meeting %s status was updated from Recall AI. Incoming status: %s. Meeting status: %s. Bot status: %s.", meeting.id, status_code, meeting.status, meeting.bot_status, ) return meeting def parse_iso_datetime(value: str | None): if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None def extract_recall_bot_status(bot_data: dict) -> str: status_changes = bot_data.get("status_changes") or [] if status_changes: return status_changes[-1].get("code") or "pending" return "pending" def _extract_recall_bot_sub_code(payload: dict | None) -> str | None: if not payload: return None status_changes = payload.get("status_changes") if isinstance(status_changes, list) and status_changes: return status_changes[-1].get("sub_code") data = payload.get("data") or {} if isinstance(data.get("data"), dict): return ((data.get("data") or {}).get("sub_code")) if isinstance(data.get("status"), dict): return ((data.get("status") or {}).get("sub_code")) return None def map_recall_bot_status_to_meeting_status(status_code: str | None, payload: dict | None = None) -> str | None: if status_code in {"call_ended", "bot.call_ended"}: sub_code = _extract_recall_bot_sub_code(payload) if sub_code == "bot_kicked_from_waiting_room": return "not_admitted" mapping = { "scheduled": "scheduled", "joining_call": "bot.joining_call", "bot.joining_call": "bot.joining_call", "in_waiting_room": "bot.in_waiting_room", "bot.in_waiting_room": "bot.in_waiting_room", "in_call_not_recording": "bot.in_call_not_recording", "bot.in_call_not_recording": "bot.in_call_not_recording", "in_call_recording": "bot.in_call_recording", "bot.in_call_recording": "bot.in_call_recording", "call_ended": "bot.call_ended", "bot.call_ended": "bot.call_ended", "recording_done": "recording.done", "recording.done": "recording.done", "recording.failed": "recording.failed", "done": "bot.done", "bot.done": "bot.done", "fatal": "bot.fatal", "bot.fatal": "bot.fatal", "error": "bot.fatal", "transcript.processing": "transcript.processing", "transcript.done": "transcript.done", "transcript.failed": "transcript.failed", } return mapping.get(status_code) def _initial_meeting_status(scheduled_at: datetime | None) -> str: if scheduled_at and scheduled_at > datetime.now(timezone.utc): return "scheduled" return "pending" def mark_recording_media_expired(db: Session, meeting_id: int): recording = get_recording_by_meeting(db, meeting_id) if recording: recording.is_media_expired = True recording.recall_recording_id = None recording.transcript_download_url = None recording.transcript_provider_data_download_url = None recording.participant_events_download_url = None recording.speaker_timeline_download_url = None recording.participants_download_url = None db.commit() db.refresh(recording) return recording