folkugat-web/folkugat_web/services/sessions.py

import dataclasses
import datetime
from datetime import date as Date

from folkugat_web.config import date as config
from folkugat_web.dal.sql import Connection, get_connection
from folkugat_web.dal.sql.playlists import write as playlists_write
from folkugat_web.dal.sql.sessions import playlists as session_playlists
from folkugat_web.dal.sql.sessions import query, write
from folkugat_web.model import sessions as model
from folkugat_web.model import temes as temes_model
from folkugat_web.model.playlists import Playlist, PlaylistType
from folkugat_web.model.sql import Order, OrderCol, Range
from folkugat_web.services import playlists as playlists_service


def get_date_names(date: Date) -> model.DateNames:
    return model.DateNames(
        day_name=config.DAY_NAMES_CAT[date.weekday()],
        day=str(date.day),
        month_name=config.MONTH_NAMES_CAT[date.month],
        year=str(date.year),
    )


def new_session() -> model.Session:
    sessions = get_sessions()
    if not sessions:
        venue = model.SessionVenue()
    else:
        venue = sessions[-1].venue
    return model.Session(venue=venue)


def get_sessions() -> list[model.Session]:
    return query.get_sessions(order_by=OrderCol(model.SessionCols.DATE, Order.ASCENDING))


def get_next_sessions(limit: int | None = None, offset: int | None = None) -> list[model.Session]:
    return query.get_sessions(
        date_range=Range(gte=datetime.date.today()),
        order_by=OrderCol(model.SessionCols.DATE, Order.ASCENDING),
        limit=limit,
        offset=offset,
    )


def get_sessions_history(limit: int | None = None, offset: int | None = None) -> list[model.Session]:
    return query.get_sessions(
        date_range=Range(lt=datetime.date.today()),
        order_by=OrderCol(model.SessionCols.DATE, Order.DESCENDING),
        limit=limit,
        offset=offset,
    )


def get_next_session() -> model.Session | None:
    return next(iter(get_next_sessions(limit=1,)), None)


def get_live_session() -> model.Session | None:
    return next(iter(query.get_sessions(is_live=True)), None)


def get_session(session_id: int) -> model.Session | None:
    return next(iter(query.get_sessions(session_id=session_id)), None)


def insert_session(session: model.Session):
    created_session = write.insert_session(session)
    if created_session.id is not None:
        _create_session_playlists(created_session.id)
    return created_session


def set_session(session: model.Session):
    write.update_session(session)
    _update_session_playlist_names(session)


def delete_session(session_id: int):
    _delete_session_playlists(session_id)
    write.delete_session_by_id(session_id)


def stop_live_sessions():
    write.stop_live_sessions()


def set_live_session(session_id: int):
    write.set_live_session(session_id=session_id)


def get_session_playlist_id(
    session_id: int,
    playlist_type: PlaylistType,
    con: Connection | None = None,
) -> int | None:
    with get_connection(con=con) as con:
        return session_playlists.get_playlist_id(
            session_id=session_id,
            playlist_type=playlist_type,
            con=con,
        )


def get_session_playlist(
    session_id: int,
    playlist_type: PlaylistType,
    con: Connection | None = None,
) -> Playlist | None:
    with get_connection(con=con) as con:
        playlist_id = get_session_playlist_id(
            session_id=session_id,
            playlist_type=playlist_type,
            con=con,
        )
        if playlist_id is None:
            return None
        return playlists_service.get_playlist(
            playlist_id=playlist_id,
            con=con,
        )


def get_session_setlist(
    session_id: int,
    con: Connection | None = None,
) -> Playlist | None:
    return get_session_playlist(
        session_id=session_id,
        playlist_type=PlaylistType.SESSION_SETLIST,
        con=con,
    )


def get_session_slowjam(
    session_id: int,
    con: Connection | None = None,
) -> Playlist | None:
    return get_session_playlist(
        session_id=session_id,
        playlist_type=PlaylistType.SESSION_SLOWJAM,
        con=con,
    )


def add_playlists_to_session(session: model.Session) -> model.Session:
    if session.id is not None:
        with get_connection() as con:
            setlist = get_session_setlist(session_id=session.id, con=con)
            if setlist:
                setlist = playlists_service.add_temes_to_playlist(setlist)
            slowjam = get_session_slowjam(session_id=session.id, con=con)
            if slowjam:
                slowjam = playlists_service.add_temes_to_playlist(slowjam)
            session = dataclasses.replace(
                session,
                setlist=setlist,
                slowjam=slowjam,
            )
    return session


def get_commonly_played_temes(
    tema_id: int,
) -> list[temes_model.CommonlyPlayedTema]:
    return session_playlists.get_commonly_played_tunes(tema_id=tema_id)


def _get_playlist_names(session: model.Session) -> tuple[str, str]:
    date_names = get_date_names(session.date)

    setlist_name = f"Sessió del {date_names.day} {date_names.month_name} de {date_names.year}"
    slowjam_name = f"Slow Jam del {date_names.day} {date_names.month_name} de {date_names.year}"

    return setlist_name, slowjam_name


def _create_session_playlists(session_id: int):
    session = get_session(session_id=session_id)
    if not session:
        return

    setlist_name, slowjam_name = _get_playlist_names(session=session)

    setlist_playlist_id = playlists_write.create_playlist(name=setlist_name)
    slowjam_playlist_id = playlists_write.create_playlist(name=slowjam_name)

    with get_connection() as con:
        session_playlists.insert_playlist(
            session_id=session_id,
            playlist_type=PlaylistType.SESSION_SETLIST,
            playlist_id=setlist_playlist_id,
            con=con
        )
        session_playlists.insert_playlist(
            session_id=session_id,
            playlist_type=PlaylistType.SESSION_SLOWJAM,
            playlist_id=slowjam_playlist_id,
            con=con
        )


def _update_session_playlist_names(session: model.Session):
    if session.id is None:
        return

    setlist_name, slowjam_name = _get_playlist_names(session=session)

    with get_connection() as con:
        setlist_playlist_id = session_playlists.get_playlist_id(
            session_id=session.id,
            playlist_type=PlaylistType.SESSION_SETLIST,
            con=con
        )
        slowjam_playlist_id = session_playlists.get_playlist_id(
            session_id=session.id,
            playlist_type=PlaylistType.SESSION_SLOWJAM,
            con=con
        )

        if setlist_playlist_id is not None:
            playlists_write.update_playlist_name(
                playlist_id=setlist_playlist_id,
                name=setlist_name,
                con=con
            )
        if slowjam_playlist_id is not None:
            playlists_write.update_playlist_name(
                playlist_id=slowjam_playlist_id,
                name=slowjam_name,
                con=con
            )


def _delete_session_playlists(session_id: int):
    with get_connection() as con:
        setlist_playlist_id = session_playlists.get_playlist_id(
            session_id=session_id,
            playlist_type=PlaylistType.SESSION_SETLIST,
            con=con
        )
        slowjam_playlist_id = session_playlists.get_playlist_id(
            session_id=session_id,
            playlist_type=PlaylistType.SESSION_SLOWJAM,
            con=con
        )

        if setlist_playlist_id is not None:
            session_playlists.delete_playlist(
                session_id=session_id,
                playlist_type=PlaylistType.SESSION_SETLIST,
                con=con,
            )
            playlists_write.delete_playlist(
                playlist_id=setlist_playlist_id,
                con=con,
            )
        if slowjam_playlist_id is not None:
            session_playlists.delete_playlist(
                session_id=session_id,
                playlist_type=PlaylistType.SESSION_SLOWJAM,
                con=con,
            )
            playlists_write.delete_playlist(
                playlist_id=slowjam_playlist_id,
                con=con,
            )