| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188 |
- import logging
- from uuid import uuid4
- import requests
- from django.apps import apps
- from django.conf import settings
- from django.core.files.base import ContentFile
- from django.db import models
- from django.urls import reverse
- from django.utils.translation import gettext_lazy as _
- from django_extensions.db.models import TimeStampedModel
- from podcasts.sources.podcastindex import lookup_podcast_from_podcastindex
- from scrobbles.mixins import (
- ObjectWithGenres,
- ScrobblableConstants,
- ScrobblableMixin,
- )
- from taggit.managers import TaggableManager
- logger = logging.getLogger(__name__)
- BNULL = {"blank": True, "null": True}
- class Producer(TimeStampedModel):
- name = models.CharField(max_length=255)
- uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
- def __str__(self):
- return f"{self.name}"
- @classmethod
- def find_or_create(cls, name):
- producer = cls.objects.filter(name__iexact=name).first()
- if not producer:
- producer = cls.objects.create(name=name)
- return producer
- class Podcast(TimeStampedModel):
- name = models.CharField(max_length=255)
- uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
- producer = models.ForeignKey(
- Producer, on_delete=models.DO_NOTHING, **BNULL
- )
- podcastindex_id = models.CharField(max_length=100, **BNULL)
- owner = models.CharField(max_length=150, *BNULL)
- description = models.TextField(**BNULL)
- active = models.BooleanField(default=True)
- feed_url = models.URLField(**BNULL)
- site_link = models.URLField(**BNULL)
- description = models.TextField(**BNULL)
- cover_image = models.ImageField(upload_to="podcasts/covers/", **BNULL)
- itunes_id = models.TextField(max_length=15, **BNULL)
- dead_date = models.DateField(**BNULL)
- genre = TaggableManager(through=ObjectWithGenres)
- def __str__(self):
- return f"{self.name}"
- def get_absolute_url(self):
- return reverse("podcasts:podcast_detail", kwargs={"slug": self.uuid})
- def scrobbles(self, user):
- Scrobble = apps.get_model("scrobbles", "Scrobble")
- return Scrobble.objects.filter(
- user=user, podcast_episode__podcast=self
- ).order_by("-timestamp")
- @property
- def itunes_link(self) -> str:
- if not self.itunes_id:
- return ""
- return f"https://podcasts.apple.com/us/podcast/id{self.itunes_id}"
- def fix_metadata(self, force=False):
- if self.podcastindex_id and not force:
- logger.warning(
- "Podcast already has PodcastIndex ID, use force=True to overwrite"
- )
- return
- podcast_dict = lookup_podcast_from_podcastindex(self.name)
- if not podcast_dict:
- logger.info(
- "No podcast data found from PodcastIndex. Are credentials setup?"
- )
- return
- genres = podcast_dict.pop("genres")
- if genres:
- self.genre.add(*genres)
- cover_url = podcast_dict.pop("image_url")
- if (not self.cover_image or force) and cover_url:
- r = requests.get(cover_url)
- if r.status_code == 200:
- fname = f"{self.name}_{self.uuid}.jpg"
- self.cover_image.save(fname, ContentFile(r.content), save=True)
- for attr, value in podcast_dict.items():
- setattr(self, attr, value)
- self.save()
- class PodcastEpisode(ScrobblableMixin):
- COMPLETION_PERCENT = getattr(settings, "PODCAST_COMPLETION_PERCENT", 90)
- podcast = models.ForeignKey(Podcast, on_delete=models.DO_NOTHING)
- number = models.IntegerField(**BNULL)
- pub_date = models.DateField(**BNULL)
- mopidy_uri = models.CharField(max_length=255, **BNULL)
- def get_absolute_url(self):
- return reverse(
- "podcasts:podcast_detail", kwargs={"slug": self.podcast.uuid}
- )
- def __str__(self):
- return f"{self.title}"
- @property
- def subtitle(self):
- return self.podcast
- @property
- def strings(self) -> ScrobblableConstants:
- return ScrobblableConstants(verb="Listening", tags="microphone")
- @property
- def info_link(self):
- return ""
- @property
- def primary_image_url(self) -> str:
- url = ""
- if self.podcast.cover_image:
- url = self.podcast.cover_image.url
- return url
- @classmethod
- def find_or_create(
- cls,
- title: str,
- pub_date: str,
- episode_num: int = 0,
- base_run_time_seconds: int = 2400,
- mopidy_uri: str = "",
- podcast_name: str = "",
- podcast_producer: str = "",
- podcast_description: str = "",
- enrich: bool = True,
- ) -> "PodcastEpisode":
- """Given a data dict from Mopidy, finds or creates a podcast and
- producer before saving the epsiode so it can be scrobbled.
- """
- log_context={"mopidy_uri": mopidy_uri, "media_type": "Podcast"}
- producer = None
- if podcast_producer:
- producer = Producer.find_or_create(podcast_producer)
- podcast, created = Podcast.objects.get_or_create(name=podcast_name, defaults={"description": podcast_description})
- log_context["podcast_id"] = podcast.id
- log_context["podcast_name"] = podcast.name
- if created:
- logger.info("Created new podcast", extra=log_context)
- if enrich and created:
- logger.info("Enriching new podcast", extra=log_context)
- podcast.fix_metadata()
- episode, created = cls.objects.get_or_create(
- title=title,
- podcast=podcast,
- defaults={
- "base_run_time_seconds": base_run_time_seconds,
- "number": episode_num,
- "pub_date": pub_date,
- "mopidy_uri": mopidy_uri,
- }
- )
- if created:
- log_context["episode_id"] = episode.id
- log_context["episode_title"] = episode.title
- logger.info("Created new podcast episode", extra=log_context)
- return episode
|