[videos] Clean up video tools
This commit is contained in:
@ -78,11 +78,13 @@ class ScrobblableMixin(TimeStampedModel):
|
||||
"user_id": user_id,
|
||||
"timestamp": timezone.now(),
|
||||
"source": source,
|
||||
"playback_position_seconds": playback_position_seconds,
|
||||
"status": status,
|
||||
"log": log,
|
||||
"playback_position_seconds": playback_position_seconds,
|
||||
}
|
||||
|
||||
if log:
|
||||
scrobble_data["log"] = log
|
||||
|
||||
logger.info(
|
||||
"[scrobble_for_user] called",
|
||||
extra={
|
||||
|
||||
@ -135,53 +135,10 @@ def jellyfin_scrobble_media(
|
||||
|
||||
|
||||
def web_scrobbler_scrobble_media(
|
||||
post_data: dict, user_id: int
|
||||
youtube_id: str, user_id: int, status: str = "started"
|
||||
) -> Optional[Scrobble]:
|
||||
media_type = Scrobble.MediaType.VIDEO
|
||||
|
||||
event_name = post_data.get("eventName")
|
||||
parsed = post_data.get("data").get("song").get("parsed")
|
||||
processed = post_data.get("data").get("song").get("processed")
|
||||
video, created = Video.objects.get_or_create(
|
||||
video_type=Video.VideoType.YOUTUBE,
|
||||
youtube_url=parsed.get("originUrl"),
|
||||
)
|
||||
timestamp = datetime.utcfromtimestamp(
|
||||
post_data.get("time", 0) / 1000
|
||||
).replace(tzinfo=pytz.utc)
|
||||
|
||||
if created or event_name == "nowplaying":
|
||||
processed = post_data.get("data").get("song").get("processed")
|
||||
video.run_time_seconds = processed.get("duration", 1500)
|
||||
if not processed.get("duration"):
|
||||
video.run_time_seconds = 1500
|
||||
# TODO maybe artist could be the series?
|
||||
video.title = " - ".join(
|
||||
[processed.get("artist"), processed.get("track")]
|
||||
)
|
||||
video.save()
|
||||
return video.scrobble_for_user(
|
||||
user_id,
|
||||
source="YouTube",
|
||||
playback_position_seconds=0,
|
||||
status=event_name,
|
||||
)
|
||||
|
||||
scrobble = Scrobble.objects.filter(
|
||||
user_id=user_id, video=video, in_progress=True
|
||||
).first()
|
||||
if not scrobble:
|
||||
return video.scrobble_for_user(
|
||||
user_id,
|
||||
source="YouTube",
|
||||
playback_position_seconds=0,
|
||||
status=event_name,
|
||||
)
|
||||
if event_name == "paused":
|
||||
scrobble.pause()
|
||||
if event_name == "resumedplaying":
|
||||
scrobble.resume()
|
||||
return 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):
|
||||
@ -620,7 +577,7 @@ def web_scrobbler_scrobble_video_or_song(
|
||||
|
||||
# No track found, create a Video
|
||||
if not track:
|
||||
Video.find_or_create(data_dict)
|
||||
Video.get_from_youtube_id()
|
||||
|
||||
# Now we run off a scrobble
|
||||
mopidy_data = {
|
||||
|
||||
65
vrobbler/apps/videos/metadata.py
Normal file
65
vrobbler/apps/videos/metadata.py
Normal file
@ -0,0 +1,65 @@
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
import pendulum
|
||||
from meta_yt import YouTube
|
||||
|
||||
|
||||
YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v="
|
||||
IMDB_VIDEO_URL = "https://www.imdb.com/title/tt"
|
||||
|
||||
|
||||
class VideoType(Enum):
|
||||
UNKNOWN = "U"
|
||||
TV_EPISODE = "E"
|
||||
MOVIE = "M"
|
||||
SKATE_VIDEO = "S"
|
||||
YOUTUBE = "Y"
|
||||
|
||||
@classmethod
|
||||
def as_choices(cls) -> tuple:
|
||||
return tuple((i.name, i.value) for i in cls)
|
||||
|
||||
|
||||
class VideoMetadata:
|
||||
title: str
|
||||
video_type: VideoType = VideoType.UNKNOWN
|
||||
run_time_seconds: int = (
|
||||
60 # Silly default, but things break if this is 0 or null
|
||||
)
|
||||
imdb_id: Optional[str]
|
||||
youtube_id: Optional[str]
|
||||
|
||||
# IMDB specific
|
||||
episode_number: Optional[str]
|
||||
season_number: Optional[str]
|
||||
next_imdb_id: Optional[str]
|
||||
year: Optional[int]
|
||||
tv_series_id: Optional[int]
|
||||
plot: Optional[str]
|
||||
imdb_rating: Optional[str]
|
||||
cover_url: Optional[str]
|
||||
overview: Optional[str]
|
||||
|
||||
# YouTube specific
|
||||
channel_id: Optional[int]
|
||||
|
||||
# General
|
||||
cover_url: Optional[str]
|
||||
genres: list[str]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
imdb_id: Optional[str] = "",
|
||||
youtube_id: Optional[str] = "",
|
||||
run_time_seconds: int = 900,
|
||||
):
|
||||
self.imdb_id = imdb_id
|
||||
self.youtube_id = youtube_id
|
||||
self.run_time_seconds = run_time_seconds
|
||||
|
||||
def as_dict_with_cover_and_genres(self) -> tuple:
|
||||
video_dict = vars(self)
|
||||
cover = video_dict.pop("cover_url")
|
||||
genres = video_dict.pop("genres")
|
||||
return video_dict, cover, genres
|
||||
@ -19,8 +19,9 @@ from scrobbles.mixins import (
|
||||
ScrobblableMixin,
|
||||
)
|
||||
from taggit.managers import TaggableManager
|
||||
from videos.services.metadata import VideoMetadata
|
||||
from videos.metadata import VideoMetadata
|
||||
from videos.sources.imdb import lookup_video_from_imdb
|
||||
|
||||
from vrobbler.apps.videos.sources.youtube import lookup_video_from_youtube
|
||||
|
||||
YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v="
|
||||
@ -329,6 +330,5 @@ class Video(ScrobblableMixin):
|
||||
cls, data_dict: dict, post_keys: dict = JELLYFIN_POST_KEYS
|
||||
) -> Optional["Video"]:
|
||||
"""Thes smallest of wrappers around our actual get or create utility."""
|
||||
from videos.utils import get_or_create_video
|
||||
|
||||
return get_or_create_video(data_dict, post_keys)
|
||||
imdb_key = post_keys.get("IMDB_ID")
|
||||
return cls.get_from_imdb_id(data_dict.get(imdb_key))
|
||||
|
||||
@ -1,65 +0,0 @@
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
import pendulum
|
||||
from meta_yt import YouTube
|
||||
|
||||
|
||||
YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v="
|
||||
IMDB_VIDEO_URL = "https://www.imdb.com/title/tt"
|
||||
|
||||
|
||||
class VideoType(Enum):
|
||||
UNKNOWN = "U"
|
||||
TV_EPISODE = "E"
|
||||
MOVIE = "M"
|
||||
SKATE_VIDEO = "S"
|
||||
YOUTUBE = "Y"
|
||||
|
||||
@classmethod
|
||||
def as_choices(cls) -> tuple:
|
||||
return tuple((i.name, i.value) for i in cls)
|
||||
|
||||
|
||||
class VideoMetadata:
|
||||
title: str
|
||||
video_type: VideoType = VideoType.UNKNOWN
|
||||
run_time_seconds: int = (
|
||||
60 # Silly default, but things break if this is 0 or null
|
||||
)
|
||||
imdb_id: Optional[str]
|
||||
youtube_id: Optional[str]
|
||||
|
||||
# IMDB specific
|
||||
episode_number: Optional[str]
|
||||
season_number: Optional[str]
|
||||
next_imdb_id: Optional[str]
|
||||
year: Optional[int]
|
||||
tv_series_id: Optional[int]
|
||||
plot: Optional[str]
|
||||
imdb_rating: Optional[str]
|
||||
cover_url: Optional[str]
|
||||
overview: Optional[str]
|
||||
|
||||
# YouTube specific
|
||||
channel_id: Optional[int]
|
||||
|
||||
# General
|
||||
cover_url: Optional[str]
|
||||
genres: list[str]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
imdb_id: Optional[str] = "",
|
||||
youtube_id: Optional[str] = "",
|
||||
run_time_seconds: int = 900,
|
||||
):
|
||||
self.imdb_id = imdb_id
|
||||
self.youtube_id = youtube_id
|
||||
self.run_time_seconds = run_time_seconds
|
||||
|
||||
def as_dict_with_cover_and_genres(self) -> tuple:
|
||||
video_dict = vars(self)
|
||||
cover = video_dict.pop("cover_url")
|
||||
genres = video_dict.pop("genres")
|
||||
return video_dict, cover, genres
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
import logging
|
||||
|
||||
from imdb import Cinemagoer, helpers
|
||||
from videos.services import metadata
|
||||
from videos.metadata import VideoMetadata, VideoType
|
||||
|
||||
imdb_client = Cinemagoer()
|
||||
|
||||
@ -10,7 +10,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
def lookup_video_from_imdb(
|
||||
name_or_id: str, kind: str = "movie"
|
||||
) -> metadata.VideoMetadata:
|
||||
) -> VideoMetadata:
|
||||
from videos.models import Series
|
||||
|
||||
# Very few video titles start with tt, but IMDB IDs often come in with it
|
||||
@ -24,7 +24,7 @@ def lookup_video_from_imdb(
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
video_metadata = metadata.VideoMetadata(imdb_id=imdb_id)
|
||||
video_metadata = VideoMetadata(imdb_id=imdb_id)
|
||||
imdb_result: dict = {}
|
||||
|
||||
imdb_result = imdb_client.get_movie(name_or_id)
|
||||
@ -69,14 +69,14 @@ def lookup_video_from_imdb(
|
||||
video_metadata.cover_url, width=800
|
||||
)
|
||||
|
||||
video_metadata.video_type = metadata.VideoType.MOVIE
|
||||
video_metadata.video_type = VideoType.MOVIE
|
||||
series_name = None
|
||||
if imdb_result.get("kind") == "episode":
|
||||
series_name = imdb_result.get("episode of", None).data.get(
|
||||
"title", None
|
||||
)
|
||||
series, _ = Series.objects.get_or_create(name=series_name)
|
||||
video_metadata.video_type = metadata.VideoType.TV_EPISODE
|
||||
video_metadata.video_type = VideoType.TV_EPISODE
|
||||
video_metadata.tv_series_id = series.id
|
||||
|
||||
if imdb_result.get("runtimes"):
|
||||
|
||||
@ -1,12 +1,12 @@
|
||||
import pendulum
|
||||
from meta_yt import Video, YouTube
|
||||
from videos.services import metadata
|
||||
from videos.metadata import VideoMetadata, VideoType
|
||||
|
||||
YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v="
|
||||
YOUTUBE_CHANNEL_URL = "https://www.youtube.com/channel/"
|
||||
|
||||
|
||||
def lookup_video_from_youtube(youtube_id: str) -> metadata.VideoMetadata:
|
||||
def lookup_video_from_youtube(youtube_id: str) -> VideoMetadata:
|
||||
from videos.models import Channel
|
||||
|
||||
yt_metadata: Optional[Video] = YouTube(
|
||||
@ -14,7 +14,7 @@ def lookup_video_from_youtube(youtube_id: str) -> metadata.VideoMetadata:
|
||||
).video
|
||||
|
||||
if yt_metadata:
|
||||
video_metadata = metadata.VideoMetadata(youtube_id=youtube_id)
|
||||
video_metadata = VideoMetadata(youtube_id=youtube_id)
|
||||
|
||||
if yt_metadata.channel:
|
||||
channel, _created = Channel.objects.get_or_create(
|
||||
@ -25,7 +25,7 @@ def lookup_video_from_youtube(youtube_id: str) -> metadata.VideoMetadata:
|
||||
|
||||
video_metadata.title = yt_metadata.title
|
||||
video_metadata.run_time_seconds = yt_metadata.duration
|
||||
video_metadata.video_type = metadata.VideoType.YOUTUBE
|
||||
video_metadata.video_type = VideoType.YOUTUBE
|
||||
video_metadata.youtube_id = yt_metadata.video_id
|
||||
video_metadata.cover_url = yt_metadata.thumbnail
|
||||
video_metadata.genres = yt_metadata.keywords
|
||||
|
||||
@ -29,48 +29,23 @@ def get_or_create_video(data_dict: dict, post_keys: dict, force_update=False):
|
||||
)
|
||||
return
|
||||
|
||||
video, video_created = Video.objects.get_or_create(
|
||||
imdb_id=video_dict.get("imdb_id"),
|
||||
title=video_dict.get("title"),
|
||||
)
|
||||
if video_created or force_update:
|
||||
if not "overview" in video_dict.keys():
|
||||
video_dict["overview"] = data_dict.get(
|
||||
post_keys.get("OVERVIEW"), None
|
||||
)
|
||||
if not "tagline" in video_dict.keys():
|
||||
video_dict["tagline"] = data_dict.get(
|
||||
post_keys.get("TAGLINE"), None
|
||||
)
|
||||
if not "tmdb_id" in video_dict.keys():
|
||||
video_dict["tmdb_id"] = data_dict.get(
|
||||
post_keys.get("TMDB_ID"), None
|
||||
)
|
||||
video = Video.get_from_imdb_id(video_dict.get("imdb_id")
|
||||
|
||||
series_name = video_dict.pop("series_name", None)
|
||||
if series_name:
|
||||
series, series_created = Series.objects.get_or_create(
|
||||
name=series_name
|
||||
)
|
||||
if series_created:
|
||||
series.fix_metadata()
|
||||
video_dict["tv_series_id"] = series.id
|
||||
|
||||
if genres := video_dict.pop("genres", None):
|
||||
video.genre.add(*genres)
|
||||
|
||||
if cover_url := video_dict.pop("cover_url", None):
|
||||
video.save_image_from_url(cover_url)
|
||||
|
||||
Video.objects.filter(pk=video.id).update(**video_dict)
|
||||
video.refresh_from_db()
|
||||
# TODO this is a hack so we don't spam scrobbles without a run time seconds
|
||||
if video.run_time_seconds == 0:
|
||||
video.run_time_seconds = 1800
|
||||
video.save(update_fields=["run_time_seconds"])
|
||||
if not "overview" in video_dict.keys():
|
||||
video_dict["overview"] = data_dict.get(
|
||||
post_keys.get("OVERVIEW"), None
|
||||
)
|
||||
if not "tagline" in video_dict.keys():
|
||||
video_dict["tagline"] = data_dict.get(
|
||||
post_keys.get("TAGLINE"), None
|
||||
)
|
||||
if not "tmdb_id" in video_dict.keys():
|
||||
video_dict["tmdb_id"] = data_dict.get(
|
||||
post_keys.get("TMDB_ID"), None
|
||||
)
|
||||
|
||||
return video
|
||||
|
||||
|
||||
def get_or_create_video_from_skatevideosite(title: str, force_update=True):
|
||||
...
|
||||
def get_or_create_video_from_skatevideosite(title: str, force_update: bool=True):
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user