vrobbler/vrobbler/apps/scrobbles/scrobblers.py

import logging
import re
from datetime import datetime
from typing import Optional

import pendulum
import pytz
from beers.models import Beer
from boardgames.models import BoardGame
from books.models import Book
from dateutil.parser import parse
from django.utils import timezone
from locations.constants import LOCATION_PROVIDERS
from locations.models import GeoLocation
from music.constants import JELLYFIN_POST_KEYS, MOPIDY_POST_KEYS
from music.models import Track
from podcasts.models import PodcastEpisode
from podcasts.utils import parse_mopidy_uri
from puzzles.models import Puzzle
from scrobbles.constants import (
    JELLYFIN_AUDIO_ITEM_TYPES,
    MANUAL_SCROBBLE_FNS,
    SCROBBLE_CONTENT_URLS,
)
from scrobbles.models import Scrobble
from scrobbles.utils import convert_to_seconds
from sports.models import SportEvent
from sports.thesportsdb import lookup_event_from_thesportsdb
from tasks.models import Task
from videogames.howlongtobeat import lookup_game_from_hltb
from videogames.models import VideoGame
from videos.models import Video
from webpages.models import WebPage

from vrobbler.apps.tasks.constants import (
    TODOIST_TITLE_PREFIX_LABELS,
    TODOIST_TITLE_SUFFIX_LABELS,
)

logger = logging.getLogger(__name__)


def mopidy_scrobble_media(post_data: dict, user_id: int) -> Scrobble:
    media_type = Scrobble.MediaType.TRACK
    if "podcast" in post_data.get("mopidy_uri", ""):
        media_type = Scrobble.MediaType.PODCAST_EPISODE

    logger.info(
        "[mopidy_webhook] called",
        extra={
            "user_id": user_id,
            "post_data": post_data,
            "media_type": media_type,
        },
    )

    if media_type == Scrobble.MediaType.PODCAST_EPISODE:
        parsed_data = parse_mopidy_uri(post_data.get("mopidy_uri", ""))
        podcast_name = post_data.get(
            "album", parsed_data.get("podcast_name", "")
        )

        media_obj = PodcastEpisode.find_or_create(
            title=parsed_data.get("episode_filename", ""),
            podcast_name=podcast_name,
            producer_name=post_data.get("artist", ""),
            number=parsed_data.get("episode_num", ""),
            pub_date=parsed_data.get("pub_date", ""),
            mopidy_uri=post_data.get("mopidy_uri", ""),
        )
    else:
        media_obj = Track.find_or_create(
            title=post_data.get("name", ""),
            artist_name=post_data.get("artist", ""),
            album_name=post_data.get("album", ""),
            run_time_seconds=post_data.get("run_time", 900000),
        )

    log = {}
    try:
        log = {"mopidy_source": post_data.get("mopidy_uri", "").split(":")[0]}
    except IndexError:
        pass

    return media_obj.scrobble_for_user(
        user_id,
        source="Mopidy",
        playback_position_seconds=int(
            post_data.get(MOPIDY_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1)
            / 1000
        ),
        status=post_data.get(MOPIDY_POST_KEYS.get("STATUS"), ""),
        log=log,
    )


def jellyfin_scrobble_media(
    post_data: dict, user_id: int
) -> Optional[Scrobble]:
    media_type = Scrobble.MediaType.VIDEO
    if post_data.pop("ItemType", "") in JELLYFIN_AUDIO_ITEM_TYPES:
        media_type = Scrobble.MediaType.TRACK

    null_position_on_progress = (
        post_data.get("PlaybackPosition") == "00:00:00"
        and post_data.get("NotificationType") == "PlaybackProgress"
    )

    # Jellyfin has some race conditions with it's webhooks, these hacks fix some of them
    if null_position_on_progress:
        logger.info(
            "[jellyfin_scrobble_media] no playback position tick, aborting",
            extra={"post_data": post_data},
        )
        return

    timestamp = parse(
        post_data.get(JELLYFIN_POST_KEYS.get("TIMESTAMP"), "")
    ).replace(tzinfo=pytz.utc)

    playback_position_seconds = int(
        post_data.get(JELLYFIN_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1)
        / 10000000
    )
    if media_type == Scrobble.MediaType.VIDEO:
        media_obj = Video.get_from_imdb_id(
            post_data.get("Provider_imdb", "").replace("tt", "")
        )
    else:
        media_obj = Track.find_or_create(
            title=post_data.get("Name", ""),
            artist_name=post_data.get("Artist", ""),
            album_name=post_data.get("Album", ""),
            run_time_seconds=convert_to_seconds(
                post_data.get("RunTime", 900000)
            ),
            musicbrainz_id=post_data.get("Provider_musicbrainztrack", ""),
        )
        # A hack because we don't worry about updating music ... we either finish it or we don't
        playback_position_seconds = 0

    if not media_obj:
        logger.info(
            "[jellyfin_scrobble_media] no video found from POST data",
            extra={"post_data": post_data},
        )
        return

    playback_status = "resumed"
    if post_data.get("IsPaused"):
        playback_status = "paused"
    elif post_data.get("NotificationType") == "PlaybackStop":
        playback_status = "stopped"

    return media_obj.scrobble_for_user(
        user_id,
        source=post_data.get(JELLYFIN_POST_KEYS.get("SOURCE")),
        playback_position_seconds=playback_position_seconds,
        status=playback_status,
    )


def web_scrobbler_scrobble_media(
    youtube_id: str, user_id: int, status: str = "started"
) -> Optional[Scrobble]:
    video = Video.get_from_youtube_id(youtube_id)
    return video.scrobble_for_user(user_id, status, source="Web Scrobbler")


def manual_scrobble_video(
    video_id: str, user_id: int, action: Optional[str] = None
):
    if "tt" in video_id:
        video = Video.get_from_imdb_id(video_id)

    else:
        video = Video.get_from_youtube_id(video_id)

    # When manually scrobbling, try finding a source from the series
    source = "Vrobbler"
    if video.tv_series:
        source = video.tv_series.preferred_source
    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": source,
    }

    logger.info(
        "[scrobblers] manual video scrobble request received",
        extra={
            "video_id": video.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.VIDEO,
        },
    )

    scrobble = Scrobble.create_or_update(video, user_id, scrobble_dict)

    if action == "stop":
        scrobble.stop(force_finish=True)

    return scrobble


def manual_scrobble_event(
    thesportsdb_id: str, user_id: int, action: Optional[str] = None
):
    data_dict = lookup_event_from_thesportsdb(thesportsdb_id)

    event = SportEvent.find_or_create(data_dict)
    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "TheSportsDB",
    }
    return Scrobble.create_or_update(event, user_id, scrobble_dict)


def manual_scrobble_video_game(
    hltb_id: str, user_id: int, action: Optional[str] = None
):
    game = VideoGame.objects.filter(hltb_id=hltb_id).first()
    if not game:
        data_dict = lookup_game_from_hltb(hltb_id)
        if not data_dict:
            logger.info(
                "[manual_scrobble_video_game] game not found on hltb",
                extra={
                    "hltb_id": hltb_id,
                    "user_id": user_id,
                    "media_type": Scrobble.MediaType.VIDEO_GAME,
                },
            )
            return

        game = VideoGame.find_or_create(data_dict)

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
        "long_play_complete": False,
    }

    logger.info(
        "[scrobblers] manual video game scrobble request received",
        extra={
            "videogame_id": game.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.VIDEO_GAME,
        },
    )

    return Scrobble.create_or_update(game, user_id, scrobble_dict)


def manual_scrobble_book(
    title: str, user_id: int, action: Optional[str] = None
):
    book = Book.get_from_google(title)

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
        "long_play_complete": False,
    }

    logger.info(
        "[scrobblers] manual book scrobble request received",
        extra={
            "book_id": book.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.BOOK,
        },
    )

    return Scrobble.create_or_update(book, user_id, scrobble_dict)


def manual_scrobble_board_game(
    bggeek_id: str, user_id: int, action: Optional[str] = None
):
    boardgame = BoardGame.find_or_create(bggeek_id)

    if not boardgame:
        logger.error(f"No board game found for ID {bggeek_id}")
        return

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
    }
    logger.info(
        "[vrobbler-scrobble] board game scrobble request received",
        extra={
            "boardgame_id": boardgame.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.BOARD_GAME,
        },
    )

    return Scrobble.create_or_update(boardgame, user_id, scrobble_dict)


def manual_scrobble_from_url(
    url: str, user_id: int, action: Optional[str] = None
) -> Scrobble:
    """We have scrobblable media URLs, and then any other webpages that
    we want to scrobble as a media type in and of itself. This checks whether
    we know about the content type, and routes it to the appropriate media
    scrobbler. Otherwise, return nothing."""
    content_key = ""
    try:
        domain = url.split("//")[-1].split("/")[0]
    except IndexError:
        domain = None

    for key, content_url in SCROBBLE_CONTENT_URLS.items():
        if domain in content_url:
            content_key = key

    item_id = None
    if not content_key:
        content_key = "-w"
        item_id = url

    # Try generic search for any URL with digit-based IDs
    if not item_id:
        try:
            item_id = re.findall(r"\d+", url)[0]
        except IndexError:
            pass

    if content_key == "-i":
        item_id = url.split("v=")[1].split("&")[0]

    scrobble_fn = MANUAL_SCROBBLE_FNS[content_key]
    return eval(scrobble_fn)(item_id, user_id, action=action)


def todoist_scrobble_task_finish(
    todoist_task: dict, user_id: int, timestamp: datetime
) -> Optional[Scrobble]:
    scrobble = Scrobble.objects.filter(
        user_id=user_id,
        log__todoist_id=todoist_task.get("todoist_id"),
        in_progress=True,
        played_to_completion=False,
    ).first()

    if not scrobble:
        logger.info(
            "[todoist_scrobble_task_finish] todoist webhook finish called on missing task"
        )
        return

    scrobble.stop(timestamp=timestamp, force_finish=True)

    return scrobble


def todoist_scrobble_update_task(
    todoist_note: dict, user_id: int
) -> Optional[Scrobble]:
    scrobble = Scrobble.objects.filter(
        in_progress=True,
        user_id=user_id,
        log__todoist_id=todoist_note.get("task_id"),
    ).first()

    if not scrobble:
        logger.info(
            "[todoist_scrobble_update_task] no task found",
            extra={
                "todoist_note": todoist_note,
                "user_id": user_id,
                "media_type": Scrobble.MediaType.TASK,
            },
        )
        return

    existing_notes = scrobble.log.get("notes", {})
    existing_notes[todoist_note.get("todoist_id")] = todoist_note.get("notes")
    scrobble.log["notes"] = existing_notes
    scrobble.save(update_fields=["log"])
    logger.info(
        "[todoist_scrobble_update_task] todoist note added",
        extra={
            "todoist_note": todoist_note,
            "user_id": user_id,
            "media_type": Scrobble.MediaType.TASK,
        },
    )

    return scrobble


def todoist_scrobble_task(
    todoist_task: dict,
    user_id: int,
    started: bool = False,
    stopped: bool = False,
) -> Scrobble:

    prefix = ""
    suffix = ""
    # TODO look up the user profile and instead of checking PREFIX/SUFFIX, check against
    # user.profile.task_context_tags which will result in context-based tag titles
    # We'd also have to migrate existing tasks to the new context based ones (maybe)
    for label in todoist_task["todoist_label_list"]:
        if label in TODOIST_TITLE_PREFIX_LABELS:
            prefix = label
        if label in TODOIST_TITLE_SUFFIX_LABELS:
            suffix = label

    if not prefix and suffix:
        logger.warning(
            "Missing a prefix and suffix tag for task",
            extra={"todoist_scrobble_task": todoist_task},
        )

    title = " ".join([prefix.capitalize(), suffix.capitalize()])

    task = Task.find_or_create(title)

    timestamp = pendulum.parse(todoist_task.get("updated_at", timezone.now()))
    in_progress_scrobble = Scrobble.objects.filter(
        user_id=user_id,
        in_progress=True,
        log__todoist_id=todoist_task.get("todoist_id"),
        task=task,
    ).last()

    if not in_progress_scrobble and stopped:
        logger.info(
            "[todoist_scrobble_task] cannot stop already stopped task",
            extra={
                "todoist_type": todoist_task["todoist_type"],
                "todoist_event": todoist_task["todoist_event"],
                "todoist_id": todoist_task["todoist_id"],
            },
        )
        return

    if in_progress_scrobble and started:
        logger.info(
            "[todoist_scrobble_task] cannot start already started task",
            extra={
                "todoist_type": todoist_task["todoist_type"],
                "todoist_event": todoist_task["todoist_event"],
                "todoist_id": todoist_task["todoist_id"],
            },
        )
        return in_progress_scrobble

    # Finish an in-progress scrobble
    if in_progress_scrobble and stopped:
        logger.info(
            "[todoist_scrobble_task] finishing",
            extra={
                "todoist_type": todoist_task["todoist_type"],
                "todoist_event": todoist_task["todoist_event"],
                "todoist_id": todoist_task["todoist_id"],
            },
        )
        return todoist_scrobble_task_finish(todoist_task, user_id, timestamp)

    # Default to create new scrobble "if not in_progress_scrobble and in_progress_in_todoist"
    # TODO Should use updated_at from TOdoist, but parsing isn't working
    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timestamp,
        "playback_position_seconds": 0,
        "source": "Todoist",
        "log": todoist_task,
    }

    logger.info(
        "[todoist_scrobble_task] creating",
        extra={
            "task_id": task.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.TASK,
        },
    )
    scrobble = Scrobble.create_or_update(task, user_id, scrobble_dict)
    return scrobble


def emacs_scrobble_update_task(
    emacs_id: str, emacs_notes: dict, user_id: int
) -> Optional[Scrobble]:
    scrobble = Scrobble.objects.filter(
        in_progress=True,
        user_id=user_id,
        log__source_id=emacs_id,
        log__source="emacs",
    ).first()

    if not scrobble:
        logger.info(
            "[emacs_scrobble_update_task] no task found",
            extra={
                "emacs_notes": emacs_notes,
                "user_id": user_id,
                "media_type": Scrobble.MediaType.TASK,
            },
        )
        return

    notes_updated = False
    for note in emacs_notes:
        existing_note_ts = [
            n.get("timestamp") for n in scrobble.log.get("notes", [])
        ]
        if not scrobble.log.get('notes"'):
            scrobble.log["notes"] = []
        if note.get("timestamp") not in existing_note_ts:
            scrobble.log["notes"].append(
                {note.get("timestamp"): note.get("content")}
            )
            notes_updated = True

    if notes_updated:
        scrobble.save(update_fields=["log"])

        logger.info(
            "[emacs_scrobble_update_task] emacs note added",
            extra={
                "emacs_note": emacs_notes,
                "user_id": user_id,
                "media_type": Scrobble.MediaType.TASK,
            },
        )

    return scrobble


def emacs_scrobble_task(
    task_data: dict,
    user_id: int,
    started: bool = False,
    stopped: bool = False,
) -> Scrobble | None:

    prefix = ""
    suffix = ""

    source_id = task_data.get("source_id")
    for label in task_data.get("labels"):
        if label in TODOIST_TITLE_PREFIX_LABELS:
            prefix = label
        if label in TODOIST_TITLE_SUFFIX_LABELS:
            suffix = label

    if not prefix and not suffix:
        logger.warning(
            "Missing a prefix and suffix tag for task",
            extra={"emacs_scrobble_task": task_data},
        )

    title = " ".join([prefix.capitalize(), suffix.capitalize()])
    task = Task.find_or_create(title)

    timestamp = pendulum.parse(task_data.get("updated_at", timezone.now()))
    in_progress_scrobble = Scrobble.objects.filter(
        user_id=user_id,
        in_progress=True,
        log__source_id=source_id,
        log__source="emacs",
        task=task,
    ).last()

    if not in_progress_scrobble and stopped:
        logger.info(
            "[emacs_scrobble_task] cannot stop already stopped task",
            extra={
                "emacs_id": source_id,
            },
        )
        return

    if in_progress_scrobble and started:
        logger.info(
            "[emacs_scrobble_task] cannot start already started task",
            extra={
                "emacs_id": source_id,
            },
        )
        return in_progress_scrobble

    # Finish an in-progress scrobble
    if in_progress_scrobble and stopped:
        logger.info(
            "[emacs_scrobble_task] finishing",
            extra={
                "emacs_id": source_id,
            },
        )
        in_progress_scrobble.stop(timestamp=timestamp, force_finish=True)
        return in_progress_scrobble

    if in_progress_scrobble:
        return in_progress_scrobble

    notes = task_data.pop("notes")
    if notes:
        task_data["notes"] = []
        for note in notes:
            task_data["notes"].append(
                {note.get("timestamp"): note.get("content")}
            )

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timestamp,
        "playback_position_seconds": 0,
        "source": "Org-mode",
        "log": task_data,
    }

    logger.info(
        "[emacs_scrobble_task] creating",
        extra={
            "task_id": task.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.TASK,
        },
    )
    scrobble = Scrobble.create_or_update(task, user_id, scrobble_dict)
    return scrobble


def manual_scrobble_task(url: str, user_id: int, action: Optional[str] = None):
    source_id = re.findall(r"\d+", url)[0]

    if "todoist" in url:
        source = "Todoist"
        title = "Generic Todoist task"
        description = " ".join(url.split("/")[-1].split("-")[:-1]).capitalize()

    task = Task.find_or_create(title)

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": source,
        "log": {"description": description, "source_id": source_id},
    }
    logger.info(
        "[vrobbler-scrobble] webpage scrobble request received",
        extra={
            "task_id": task.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.WEBPAGE,
        },
    )
    scrobble = Scrobble.create_or_update(task, user_id, scrobble_dict)
    return scrobble


def manual_scrobble_webpage(
    url: str, user_id: int, action: Optional[str] = None
):
    webpage = WebPage.find_or_create({"url": url})

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
    }
    logger.info(
        "[vrobbler-scrobble] webpage scrobble request received",
        extra={
            "webpage_id": webpage.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.WEBPAGE,
        },
    )

    scrobble = Scrobble.create_or_update(webpage, user_id, scrobble_dict)
    # possibly async this?
    scrobble.push_to_archivebox()
    return scrobble


def gpslogger_scrobble_location(data_dict: dict, user_id: int) -> Scrobble:
    location = GeoLocation.find_or_create(data_dict)

    timestamp = pendulum.parse(data_dict.get("time", timezone.now()))
    extra_data = {
        "user_id": user_id,
        "timestamp": timestamp,
        "source": "GPSLogger",
        "media_type": Scrobble.MediaType.GEO_LOCATION,
    }

    scrobble = Scrobble.create_or_update_location(
        location,
        extra_data,
        user_id,
    )

    provider = LOCATION_PROVIDERS[data_dict.get("prov")]

    if "gps_updates" not in scrobble.log.keys():
        scrobble.log["gps_updates"] = []

    scrobble.log["gps_updates"].append(
        {
            "timestamp": data_dict.get("time"),
            "position_provider": provider,
        }
    )
    if scrobble.timestamp:
        scrobble.playback_position_seconds = (
            timezone.now() - scrobble.timestamp
        ).seconds

    scrobble.save(update_fields=["log", "playback_position_seconds"])
    logger.info(
        "[gpslogger_webhook] gpslogger scrobble request received",
        extra={
            "scrobble_id": scrobble.id,
            "provider": provider,
            "user_id": user_id,
            "timestamp": extra_data.get("timestamp"),
            "raw_timestamp": data_dict.get("time"),
            "media_type": Scrobble.MediaType.GEO_LOCATION,
        },
    )

    return scrobble


def web_scrobbler_scrobble_video_or_song(
    data_dict: dict, user_id: Optional[int]
) -> Scrobble:
    # We're not going to create music tracks, because the only time
    # we'd hit this is if we're listening to a concert or something.
    artist_name = data_dict.get("artist")
    track_name = data_dict.get("track")
    tracks = Track.objects.filter(
        artist__name=data_dict.get("artist"), title=data_dict.get("track")
    )
    if tracks.count() > 1:
        logger.warning(
            "Multiple tracks found for Web Scrobbler",
            extra={"artist": artist_name, "track": track_name},
        )
    track = tracks.first()

    # No track found, create a Video
    if not track:
        Video.get_from_youtube_id()

    # Now we run off a scrobble
    mopidy_data = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": data_dict.get("playback_time_ticks"),
        "source": "Mopidy",
        "mopidy_status": data_dict.get("status"),
    }

    logger.info(
        "[scrobblers] webhook mopidy scrobble request received",
        extra={
            "episode_id": episode.id if episode else None,
            "user_id": user_id,
            "scrobble_dict": mopidy_data,
            "media_type": Scrobble.MediaType.PODCAST_EPISODE,
        },
    )

    scrobble = None
    if episode:
        scrobble = Scrobble.create_or_update(episode, user_id, mopidy_data)
    return scrobble


def manual_scrobble_beer(
    untappd_id: str, user_id: int, action: Optional[str] = None
):
    beer = Beer.find_or_create(untappd_id)

    if not beer:
        logger.error(f"No beer found for Untappd ID {untappd_id}")
        return

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
    }
    logger.info(
        "[vrobbler-scrobble] beer scrobble request received",
        extra={
            "beer_id": beer.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.BEER,
        },
    )

    # TODO Kick out a process to enrich the media here, and in every scrobble event
    return Scrobble.create_or_update(beer, user_id, scrobble_dict)


def manual_scrobble_puzzle(
    ipdb_id: str, user_id: int, action: Optional[str] = None
):
    puzzle = Puzzle.find_or_create(ipdb_id)

    if not puzzle:
        logger.error(f"No puzzle found for IPDB ID {ipdb_id}")
        return

    scrobble_dict = {
        "user_id": user_id,
        "timestamp": timezone.now(),
        "playback_position_seconds": 0,
        "source": "Vrobbler",
    }
    logger.info(
        "[vrobbler-scrobble] puzzle scrobble request received",
        extra={
            "puzzle_id": puzzle.id,
            "user_id": user_id,
            "scrobble_dict": scrobble_dict,
            "media_type": Scrobble.MediaType.PUZZLE,
        },
    )

    # TODO Kick out a process to enrich the media here, and in every scrobble event
    return Scrobble.create_or_update(puzzle, user_id, scrobble_dict)