secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[format] Blacken everything
All checks were successful
build & deploy / test (push) Successful in 1m53s
Details
build & deploy / deploy (push) Successful in 1m12s
Details

Browse Source
This commit is contained in:
Colin Powell
2026-03-11 23:54:24 -04:00
parent 1e11679419
commit 5934dcdf8e
49 changed files with 480 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/scrobbles_tests/conftest.py
View File

@ -22,8 +22,18 @@ def boardgame_scrobble():
played_to_completion=True,
log={
"players": [
{"person_id": first.id, "win": True, "score": 30, "color": "Blue"},
{"person_id": second.id, "win": False, "score": 28, "color": "Red"},
{
"person_id": first.id,
"win": True,
"score": 30,
"color": "Blue",
},
{
"person_id": second.id,
"win": False,
"score": 28,
"color": "Red",
},
],
},
)
@ -58,7 +68,9 @@ class MopidyRequest:
"artist": kwargs.get("artist", self.artist),
"album": kwargs.get("album", self.album),
"track_number": int(kwargs.get("track_number", self.track_number)),
"run_time_ticks": int(kwargs.get("run_time_ticks", self.run_time_ticks)),
"run_time_ticks": int(
kwargs.get("run_time_ticks", self.run_time_ticks)
),
"run_time": int(kwargs.get("run_time", self.run_time)),
"playback_time_ticks": int(
kwargs.get("playback_time_ticks", self.playback_time_ticks)
@ -101,7 +113,9 @@ def mopidy_track():
@pytest.fixture
def mopidy_track_diff_album_request_data(**kwargs):
mb_album_id = "0c56c457-afe1-4679-baab-759ba8dd2a58"
return MopidyRequest(album="Gold", musicbrainz_album_id=mb_album_id).request_json
return MopidyRequest(
album="Gold", musicbrainz_album_id=mb_album_id
).request_json
@pytest.fixture

4
tests/scrobbles_tests/test_aggregators.py
View File

@ -40,7 +40,7 @@ def test_scrobble_counts_data(
mock_get_track,
mock_get_album,
client,
mopidy_track
mopidy_track,
):
build_scrobbles(client, mopidy_track.request_json)
user = get_user_model().objects.first()
@ -68,7 +68,7 @@ def test_live_charts(
mock_get_track,
mock_get_album,
client,
mopidy_track
mopidy_track,
):
build_scrobbles(client, mopidy_track.request_json, 7, 1)
user = get_user_model().objects.first()

6
tests/scrobbles_tests/test_metadata.py
View File

@ -1,6 +1,6 @@
import pytest
#from scrobbles.dataclasses import BoardGameLogData, BoardGameScoreLogData
# from scrobbles.dataclasses import BoardGameLogData, BoardGameScoreLogData
@pytest.mark.skip("Need to get local tests running working again")
@ -19,7 +19,7 @@ def test_boardgame_log_data(boardgame_scrobble):
new=None,
rank=None,
seat_order=None,
role=None
role=None,
),
BoardGameScoreLogData(
person_id=2,
@ -32,7 +32,7 @@ def test_boardgame_log_data(boardgame_scrobble):
new=None,
rank=None,
seat_order=None,
role=None
role=None,
),
],
difficulty=None,

20
tests/test_context_processors.py
View File

@ -14,7 +14,9 @@ def mock_request():
class TestVersionInfo:
def test_returns_version_and_commit(self, mock_request):
with (
patch("vrobbler.context_processors.get_version") as mock_get_version,
patch(
"vrobbler.context_processors.get_version"
) as mock_get_version,
patch(
"vrobbler.context_processors.subprocess.check_output"
) as mock_check_output,
@ -30,7 +32,9 @@ class TestVersionInfo:
def test_uses_env_commit_if_set(self, mock_request):
with (
patch.dict(os.environ, {"VROBBLER_COMMIT": "env_commit_hash"}),
patch("vrobbler.context_processors.get_version") as mock_get_version,
patch(
"vrobbler.context_processors.get_version"
) as mock_get_version,
):
mock_get_version.return_value = "1.0.0"
@ -40,7 +44,9 @@ class TestVersionInfo:
def test_returns_unknown_when_version_fails(self, mock_request):
with (
patch("vrobbler.context_processors.get_version") as mock_get_version,
patch(
"vrobbler.context_processors.get_version"
) as mock_get_version,
patch(
"vrobbler.context_processors.subprocess.check_output"
) as mock_check_output,
@ -56,7 +62,9 @@ class TestVersionInfo:
import subprocess
with (
patch("vrobbler.context_processors.get_version") as mock_get_version,
patch(
"vrobbler.context_processors.get_version"
) as mock_get_version,
patch(
"vrobbler.context_processors.subprocess.check_output"
) as mock_check_output,
@ -72,7 +80,9 @@ class TestVersionInfo:
import subprocess
with (
patch("vrobbler.context_processors.get_version") as mock_get_version,
patch(
"vrobbler.context_processors.get_version"
) as mock_get_version,
patch(
"vrobbler.context_processors.subprocess.check_output"
) as mock_check_output,

12
tests/videos_tests/test_api.py
View File

@ -48,8 +48,12 @@ class TestVideoAPI:
assert response.status_code == 200
assert response.data["title"] == "Test Video"
def test_filter_videos_by_channel(self, client, auth_headers, channel, video):
response = client.get(f"/api/v1/videos/?channel={channel.id}", **auth_headers)
def test_filter_videos_by_channel(
self, client, auth_headers, channel, video
):
response = client.get(
f"/api/v1/videos/?channel={channel.id}", **auth_headers
)
assert response.status_code == 200
assert len(response.data["results"]) == 1
assert response.data["results"][0]["channel"] == channel.id
@ -64,7 +68,9 @@ class TestChannelAPI:
assert response.data["results"][0]["name"] == "Test Channel"
def test_get_channel(self, client, auth_headers, channel):
response = client.get(f"/api/v1/channels/{channel.id}/", **auth_headers)
response = client.get(
f"/api/v1/channels/{channel.id}/", **auth_headers
)
assert response.status_code == 200
assert response.data["name"] == "Test Channel"

4
vrobbler/apps/beers/api/serializers.py
View File

@ -6,12 +6,14 @@ class BeerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = Beer
fields = "__all__"
class BeerProducerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = BeerProducer
fields = "__all__"
class BeerStyleSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = BeerStyle

2
vrobbler/apps/beers/api/views.py
View File

@ -8,11 +8,13 @@ class BeerViewSet(viewsets.ModelViewSet):
serializer_class = serializers.BeerSerializer
permission_classes = [permissions.IsAuthenticated]
class BeerProducerViewSet(viewsets.ModelViewSet):
queryset = models.BeerProducer.objects.all().order_by("-created")
serializer_class = serializers.BeerProducerSerializer
permission_classes = [permissions.IsAuthenticated]
class BeerStyleViewSet(viewsets.ModelViewSet):
queryset = models.BeerStyle.objects.all().order_by("-created")
serializer_class = serializers.BeerStyleSerializer

4
vrobbler/apps/boardgames/api/serializers.py
View File

@ -1,21 +1,25 @@
from boardgames import models
from rest_framework import serializers
class BoardGameDesignerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.BoardGameDesigner
fields = "__all__"
class BoardGamePublisherSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.BoardGamePublisher
fields = "__all__"
class BoardGameLocationSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.BoardGameLocation
fields = "__all__"
class BoardGameSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.BoardGame

5
vrobbler/apps/boardgames/bgg.py
View File

@ -19,7 +19,10 @@ SEARCH_ID_URL = (
)
GAME_ID_URL = "https://boardgamegeek.com/xmlapi/boardgame/{id}"
BGG_ACCESS_TOKEN = getattr(settings, "BGG_ACCESS_TOKEN", "")
BASE_HEADERS = {"User-Agent": "Vrobbler 31.0", "Authorization": f"Bearer {BGG_ACCESS_TOKEN}"}
BASE_HEADERS = {
"User-Agent": "Vrobbler 31.0",
"Authorization": f"Bearer {BGG_ACCESS_TOKEN}",
}
def take_first(thing: Optional[list]) -> str:

9
vrobbler/apps/boardgames/models.py
View File

@ -323,7 +323,10 @@ class BoardGame(ScrobblableMixin):
game = cls.objects.filter(bggeek_id=lookup_id).first()
if game:
logger.info("Board game exists in database.", extra={"lookup_id": lookup_id, "data": data})
logger.info(
"Board game exists in database.",
extra={"lookup_id": lookup_id, "data": data},
)
return game
bgg_data = lookup_boardgame_from_bgg(data.get("name"))
@ -334,9 +337,7 @@ class BoardGame(ScrobblableMixin):
publishers = bgg_data.pop("publishers", [])
cover_url = bgg_data.pop("cover_url")
game = cls.objects.create(
**bgg_data
)
game = cls.objects.create(**bgg_data)
game.save_image_from_url(cover_url)
game.cooperative = data.get("cooperative", False)

9
vrobbler/apps/books/management/commands/cleanup_book_scrobble_times.py
View File

@ -14,14 +14,13 @@ logger = logging.getLogger(__name__)
# Grace period between page reads for it to be a new scrobble
SESSION_GAP_SECONDS = 1800 # a half hour
def update_scrobble_from_page_data(scrobble, commit=True):
page_list = list(scrobble.book_page_data.items())
first_page_start_ts = datetime.fromtimestamp(page_list[0][1]["start_ts"])
last_page_end_ts = datetime.fromtimestamp(page_list[-1][1]["end_ts"])
if (
datetime(2023, 10, 15) <= first_page_start_ts <= datetime(2023, 12, 15)
):
if datetime(2023, 10, 15) <= first_page_start_ts <= datetime(2023, 12, 15):
first_page_start_ts.replace(tzinfo=pytz.timezone("Europe/Paris"))
last_page_end_ts.replace(tzinfo=pytz.timezone("Europe/Paris"))
else:
@ -38,5 +37,7 @@ class Command(BaseCommand):
def handle(self, *args, **options):
scrobbles_to_create = []
for scrobble in Scrobble.objects.filter(media_type="Book", source="KOReader"):
for scrobble in Scrobble.objects.filter(
media_type="Book", source="KOReader"
):
update_scrobble_from_page_data(scrobble)

19
vrobbler/apps/books/models.py
View File

@ -204,7 +204,9 @@ class Book(LongPlayScrobblableMixin):
return reverse("books:book_detail", kwargs={"slug": self.uuid})
@classmethod
def get_from_comicvine(cls, title: str, overwrite: bool = False, force_new: bool =False) -> "Book":
def get_from_comicvine(
cls, title: str, overwrite: bool = False, force_new: bool = False
) -> "Book":
book, created = cls.objects.get_or_create(title=title)
if not created:
@ -241,7 +243,11 @@ class Book(LongPlayScrobblableMixin):
@classmethod
def find_or_create(
cls, title: str, url: str = "", enrich: bool = False, commit: bool = True
cls,
title: str,
url: str = "",
enrich: bool = False,
commit: bool = True,
):
"""Given a title, get a Book instance.
@ -269,13 +275,18 @@ class Book(LongPlayScrobblableMixin):
if READCOMICSONLINE_URL in url:
book_dict = lookup_comic_from_comicvine(title)
book_dict["readcomics_url"] = get_comic_issue_url(url)
book_dict["next_readcomics_url"] = next_url_if_exists(book_dict["readcomics_url"])
book_dict["next_readcomics_url"] = next_url_if_exists(
book_dict["readcomics_url"]
)
if not book_dict:
book_dict = lookup_book_from_google(title)
if not book_dict:
logger.warning("No book found in any source, using data as is", extra={"title": title})
logger.warning(
"No book found in any source, using data as is",
extra={"title": title},
)
author_list = []
authors = book_dict.pop("authors", [])

10
vrobbler/apps/books/sources/comicvine.py
View File

@ -214,7 +214,9 @@ def lookup_comic_from_comicvine(title: str) -> dict:
api_key = getattr(settings, "COMICVINE_API_KEY", "")
if not api_key:
logger.warning("No ComicVine API key configured, not looking anything up")
logger.warning(
"No ComicVine API key configured, not looking anything up"
)
return {}
client = ComicVineClient(
@ -223,9 +225,7 @@ def lookup_comic_from_comicvine(title: str) -> dict:
raw_results = client.search(title).get("results")
results = [
r
for r in raw_results
if r.get("resource_type") == resource_type
r for r in raw_results if r.get("resource_type") == resource_type
]
if not results:
logger.warning("No comic found on ComicVine")
@ -264,7 +264,7 @@ def lookup_comic_from_comicvine(title: str) -> dict:
"comicvine_data": found_result,
"summary": found_result.get("description"),
"publish_date": found_result.get("cover_date"),
"first_publish_year": found_result.get("cover_date", "")[:4]
"first_publish_year": found_result.get("cover_date", "")[:4],
}
return data_dict

8
vrobbler/apps/books/sources/google.py
View File

@ -7,7 +7,7 @@ from django.conf import settings
API_KEY = settings.GOOGLE_API_KEY
GOOGLE_BOOKS_URL = (
"https://www.googleapis.com/books/v1/volumes?q=\"{title}\"&key={key}"
'https://www.googleapis.com/books/v1/volumes?q="{title}"&key={key}'
)
logger = logging.getLogger(__name__)
@ -69,8 +69,8 @@ def lookup_book_from_google(title: str) -> dict:
book_dict["base_run_time_seconds"] = 3600
if book_dict.get("pages"):
book_dict["base_run_time_seconds"] = book_dict.get("pages", 10) * getattr(
settings, "AVERAGE_PAGE_READING_SECONDS", 60
)
book_dict["base_run_time_seconds"] = book_dict.get(
"pages", 10
) * getattr(settings, "AVERAGE_PAGE_READING_SECONDS", 60)
return book_dict

6
vrobbler/apps/books/sources/semantic.py
View File

@ -67,9 +67,9 @@ def lookup_paper_from_semantic(title: str) -> dict:
paper_dict["openaccess_pdf_url"] = result.get("openAccessPdf", {}).get(
"url"
)
paper_dict["base_run_time_seconds"] = paper_dict.get("pages", 10) * getattr(
settings, "AVERAGE_PAGE_READING_SECONDS", 60
)
paper_dict["base_run_time_seconds"] = paper_dict.get(
"pages", 10
) * getattr(settings, "AVERAGE_PAGE_READING_SECONDS", 60)
paper_dict["author_dicts"] = result.get("authors")
paper_dict["genres"] = result.get("fieldsOfStudy")

10
vrobbler/apps/books/utils.py
View File

@ -10,7 +10,7 @@ def parse_readcomicsonline_uri(uri: str) -> tuple:
except IndexError:
return "", "", ""
parts = path.split('/')
parts = path.split("/")
title = ""
volume = 1
page = 1
@ -27,7 +27,7 @@ def parse_readcomicsonline_uri(uri: str) -> tuple:
def get_comic_issue_url(url: str) -> str:
parsed = urlparse(url)
parts = [p for p in parsed.path.strip('/').split('/') if p]
parts = [p for p in parsed.path.strip("/").split("/") if p]
# Find the index of "comic"
try:
@ -42,7 +42,7 @@ def get_comic_issue_url(url: str) -> str:
# Look for the first numeric segment after the title
number = None
for segment in parts[comic_index + 2:]:
for segment in parts[comic_index + 2 :]:
if segment.isdigit():
number = segment
break
@ -55,5 +55,7 @@ def get_comic_issue_url(url: str) -> str:
normalized_path = "/" + "/".join(new_parts)
# Rebuild full URL (same scheme and host)
simplified_url = urlunparse(parsed._replace(path=normalized_path, query='', fragment=''))
simplified_url = urlunparse(
parsed._replace(path=normalized_path, query="", fragment="")
)
return simplified_url

54
vrobbler/apps/foods/sources/usda.py
View File

@ -110,14 +110,22 @@ class USDAFoodAPI:
if "nutrientNumber" in nutrient:
nutrient_id = str(nutrient.get("nutrientNumber", ""))
value = nutrient.get("value")
elif "nutrient" in nutrient and isinstance(nutrient["nutrient"], dict):
nutrient_id = str(nutrient.get("nutrient", {}).get("id", ""))
elif "nutrient" in nutrient and isinstance(
nutrient["nutrient"], dict
):
nutrient_id = str(
nutrient.get("nutrient", {}).get("id", "")
)
value = nutrient.get("value")
elif "number" in nutrient:
nutrient_id = str(nutrient.get("number", ""))
value = nutrient.get("value")
if nutrient_id and nutrient_id in nutrient_map and value is not None:
if (
nutrient_id
and nutrient_id in nutrient_map
and value is not None
):
key = nutrient_map[nutrient_id]
try:
nutrients[key] = float(value)
@ -396,7 +404,11 @@ class NutritionCalculator:
# Strategy 1: Direct search
ingredient_name,
# Strategy 2: Singular form (remove trailing 's')
(ingredient_name.rstrip("s") if ingredient_name.endswith("s") else None),
(
ingredient_name.rstrip("s")
if ingredient_name.endswith("s")
else None
),
# Strategy 3: Remove common suffixes
re.sub(
r"\s+(fresh|dried|ground|chopped|sliced)$",
@ -405,7 +417,11 @@ class NutritionCalculator:
flags=re.IGNORECASE,
),
# Strategy 4: Just the last word (sometimes works for simple ingredients)
(ingredient_name.split()[-1] if len(ingredient_name.split()) > 1 else None),
(
ingredient_name.split()[-1]
if len(ingredient_name.split()) > 1
else None
),
]
for query in strategies:
@ -415,7 +431,9 @@ class NutritionCalculator:
try:
results = self.usda.search_foods(query, page_size=5)
if results:
logger.debug(f"Found {query}: {results[0].get('description')}")
logger.debug(
f"Found {query}: {results[0].get('description')}"
)
return results[0]
except Exception as e:
logger.warning(f"USDA search failed for '{query}': {e}")
@ -453,7 +471,9 @@ class NutritionCalculator:
}
# Get base weight in grams
base_grams = volume_to_grams.get(unit, 100) # Default to 100g if unknown
base_grams = volume_to_grams.get(
unit, 100
) # Default to 100g if unknown
# Adjust for ingredient type (very simplified)
if any(
@ -461,16 +481,22 @@ class NutritionCalculator:
for word in ["flour", "sugar", "rice", "grain"]
):
base_grams = volume_to_grams.get(unit, 125) # Dry ingredients
elif any(word in ingredient_name.lower() for word in ["oil", "butter", "fat"]):
elif any(
word in ingredient_name.lower()
for word in ["oil", "butter", "fat"]
):
base_grams = volume_to_grams.get(unit, 220) # Dense liquids
elif any(
word in ingredient_name.lower() for word in ["milk", "water", "broth"]
word in ingredient_name.lower()
for word in ["milk", "water", "broth"]
):
base_grams = volume_to_grams.get(unit, 240) # Liquids
return quantity * base_grams
def calculate_nutrition(self, ingredients: List[str], servings: int = 1) -> Dict:
def calculate_nutrition(
self, ingredients: List[str], servings: int = 1
) -> Dict:
"""
Calculate total nutrition for a recipe from ingredient list.
Returns nutrition per serving.
@ -517,8 +543,12 @@ class NutritionCalculator:
# USDA data is typically per 100g, so scale accordingly
multiplier = grams / 100
totals["calories"] += nutrients.get("calories", 0) * multiplier
totals["protein"] += nutrients.get("protein", 0) * multiplier
totals["calories"] += (
nutrients.get("calories", 0) * multiplier
)
totals["protein"] += (
nutrients.get("protein", 0) * multiplier
)
totals["fat"] += nutrients.get("fat", 0) * multiplier
totals["carbohydrates"] += (
nutrients.get("carbohydrates", 0) * multiplier

1
vrobbler/apps/locations/api/views.py
View File

@ -3,6 +3,7 @@ from rest_framework import permissions, viewsets
from locations.api import serializers
from locations import models
class GeoLocationViewSet(viewsets.ModelViewSet):
queryset = models.GeoLocation.objects.all().order_by("-created")
serializer_class = serializers.GeoLocationSerializer

1
vrobbler/apps/locations/apps.py
View File

@ -3,4 +3,3 @@ from django.apps import AppConfig
class LocationConfig(AppConfig):
name = "locations"

2
vrobbler/apps/locations/models.py
View File

@ -17,10 +17,12 @@ User = get_user_model()
GEOLOC_ACCURACY = int(getattr(settings, "GEOLOC_ACCURACY", 4))
GEOLOC_PROXIMITY = Decimal(getattr(settings, "GEOLOC_PROXIMITY", "0.0001"))
@dataclass
class GeoLocationLogData(BaseLogData, WithPeopleLogData):
pass
class GeoLocation(ScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, "LOCATION_COMPLETION_PERCENT", 100)

7
vrobbler/apps/locations/views.py
View File

@ -9,7 +9,12 @@ class GeoLocationListView(generic.ListView):
paginate_by = 75
def get_queryset(self):
return super().get_queryset().filter(scrobble__user_id=self.request.user.id).order_by("-scrobble__timestamp")
return (
super()
.get_queryset()
.filter(scrobble__user_id=self.request.user.id)
.order_by("-scrobble__timestamp")
)
def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)

4
vrobbler/apps/music/aggregators.py
View File

@ -175,7 +175,9 @@ def live_tv_charts(
seven_days_ago = now - timedelta(days=7)
thirty_days_ago = now - timedelta(days=30)
start_of_today = datetime.combine(now, datetime.min.time(), tzinfo)
start_day_of_week = start_of_today - timedelta(days=now.today().isoweekday() % 7)
start_day_of_week = start_of_today - timedelta(
days=now.today().isoweekday() % 7
)
start_day_of_month = now.replace(day=1)
start_day_of_year = now.replace(month=1, day=1)

4
vrobbler/apps/music/models.py
View File

@ -715,7 +715,9 @@ class Track(ScrobblableMixin):
except Exception:
print("No musicbrainz result found, cannot enrich")
return track
track.base_run_time_seconds = run_time_seconds or int(length / 1000)
track.base_run_time_seconds = run_time_seconds or int(
length / 1000
)
track.musicbrainz_id = mbid
if commit:
track.save()

3
vrobbler/apps/podcasts/api/serializers.py
View File

@ -1,16 +1,19 @@
from podcasts import models
from rest_framework import serializers
class ProducerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.Producer
fields = "__all__"
class PodcastEpisodeSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.PodcastEpisode
fields = "__all__"
class PodcastSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.Podcast

2
vrobbler/apps/podcasts/api/views.py
View File

@ -9,11 +9,13 @@ class ProducerViewSet(viewsets.ModelViewSet):
serializer_class = serializers.ProducerSerializer
permission_classes = [permissions.IsAuthenticated]
class PodcastViewSet(viewsets.ModelViewSet):
queryset = models.Podcast.objects.all().order_by("-created")
serializer_class = serializers.PodcastSerializer
permission_classes = [permissions.IsAuthenticated]
class PodcastEpisodeViewSet(viewsets.ModelViewSet):
queryset = models.PodcastEpisode.objects.all().order_by("-created")
serializer_class = serializers.PodcastEpisodeSerializer

8
vrobbler/apps/podcasts/models.py
View File

@ -156,12 +156,14 @@ class PodcastEpisode(ScrobblableMixin):
producer before saving the epsiode so it can be scrobbled.
"""
log_context={"mopidy_uri": mopidy_uri, "media_type": "Podcast"}
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})
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:
@ -178,7 +180,7 @@ class PodcastEpisode(ScrobblableMixin):
"number": episode_num,
"pub_date": pub_date,
"mopidy_uri": mopidy_uri,
}
},
)
if created:
log_context["episode_id"] = episode.id

36
vrobbler/apps/podcasts/utils.py
View File

@ -13,6 +13,7 @@ logger = logging.getLogger(__name__)
# TODO This should be configurable in settings or per deploy
PODCAST_DATE_FORMAT = "YYYY-MM-DD"
def parse_duration(d):
if not d:
return None
@ -24,6 +25,7 @@ def parse_duration(d):
h, m, s = parts
return h * 3600 + m * 60 + s
def fetch_metadata_from_rss(uri: str) -> dict[str, Any]:
log_context = {"mopidy_uri": uri, "media_type": "Podcast"}
podcast_data: dict[str, Any] = {}
@ -38,13 +40,22 @@ def fetch_metadata_from_rss(uri: str) -> dict[str, Any]:
resp = requests.get(rss_url, timeout=10)
feed = feedparser.parse(resp.text)
except IndexError:
logger.warning("Tried to parse uri as RSS feed, but no target found", extra=log_context)
logger.warning(
"Tried to parse uri as RSS feed, but no target found",
extra=log_context,
)
return podcast_data
podcast_publisher = getattr(feed.feed, "itunes_publisher", "")
try:
podcast_owner = feed.feed.itunes_owner.get("name") if isinstance(feed.feed.itunes_owner, dict) else feed.feed.itunes_owner
podcast_other = feed.feed.get("managingeditor") or feed.feed.get("copyright")
podcast_owner = (
feed.feed.itunes_owner.get("name")
if isinstance(feed.feed.itunes_owner, dict)
else feed.feed.itunes_owner
)
podcast_other = feed.feed.get("managingeditor") or feed.feed.get(
"copyright"
)
except AttributeError:
podcast_owner = None
podcast_other = None
@ -53,7 +64,9 @@ def fetch_metadata_from_rss(uri: str) -> dict[str, Any]:
"podcast_name": getattr(feed.feed, "title", ""),
# "podcast_description": getattr(feed.feed, "description", ""),
# "podcast_link": getattr(feed.feed, "link", ""),
"podcast_producer": podcast_publisher or podcast_owner or podcast_other
"podcast_producer": podcast_publisher
or podcast_owner
or podcast_other,
}
for entry in feed.entries:
@ -62,7 +75,9 @@ def fetch_metadata_from_rss(uri: str) -> dict[str, Any]:
podcast_data["title"] = entry.title
podcast_data["episode_num"] = int(entry.get("itunes_episode", 0))
podcast_data["pub_date"] = parse(entry.get("published", None))
podcast_data["base_run_time_seconds"] = parse_duration(entry.get("itunes_duration", None))
podcast_data["base_run_time_seconds"] = parse_duration(
entry.get("itunes_duration", None)
)
# podcast_data["description"] = entry.get("description", None)
# podcast_data["episode_url"] = entry.enclosures[0].href if entry.get("enclosures") else None
return podcast_data
@ -77,7 +92,6 @@ def parse_mopidy_uri(uri: str) -> dict[str, Any]:
if "podcast+https" in uri:
return fetch_metadata_from_rss(uri)
parsed_uri = os.path.splitext(unquote(uri))[0].split("/")
podcast_data = {
@ -87,7 +101,6 @@ def parse_mopidy_uri(uri: str) -> dict[str, Any]:
"pub_date": None,
}
episode_str = parsed_uri[-1]
episode_num = None
episode_num_pad = 0
@ -116,13 +129,18 @@ def parse_mopidy_uri(uri: str) -> dict[str, Any]:
if podcast_data["episode_num"]:
gap_to_strip += episode_num_pad
podcast_data["title"] = episode_str[gap_to_strip:].replace("-", " ").strip()
podcast_data["title"] = (
episode_str[gap_to_strip:].replace("-", " ").strip()
)
return podcast_data
def get_or_create_podcast(post_data: dict) -> PodcastEpisode:
logger.info("Looking up podcast", extra={"post_data": post_data, "media_type": "Podcast"})
logger.info(
"Looking up podcast",
extra={"post_data": post_data, "media_type": "Podcast"},
)
mopidy_uri = post_data.get("mopidy_uri", "")
parsed_data = parse_mopidy_uri(mopidy_uri)

1
vrobbler/apps/podcasts/views.py
View File

@ -3,6 +3,7 @@ from podcasts.models import Podcast
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class PodcastListView(generic.ListView):
model = Podcast
paginate_by = 20

2
vrobbler/apps/puzzles/api/serializers.py
View File

@ -1,11 +1,13 @@
from puzzles import models
from rest_framework import serializers
class PuzzleManufacturerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.PuzzleManufacturer
fields = "__all__"
class PuzzleSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = models.Puzzle

1
vrobbler/apps/puzzles/api/views.py
View File

@ -3,6 +3,7 @@ from rest_framework import permissions, viewsets
from puzzles.api import serializers
from puzzles import models
class PuzzleManufacturerViewSet(viewsets.ModelViewSet):
queryset = models.PuzzleManufacturer.objects.all().order_by("-created")
serializer_class = serializers.PuzzleManufacturerSerializer

6
vrobbler/apps/puzzles/models.py
View File

@ -11,7 +11,11 @@ from django_extensions.db.models import TimeStampedModel
from imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFit
from puzzles.sources import ipdb
from scrobbles.dataclasses import BaseLogData, WithPeopleLogData, LongPlayLogData
from scrobbles.dataclasses import (
BaseLogData,
WithPeopleLogData,
LongPlayLogData,
)
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
BNULL = {"blank": True, "null": True}

12
vrobbler/apps/scrobbles/admin.py
View File

@ -54,19 +54,23 @@ class ImportBaseAdmin(admin.ModelAdmin):
@admin.register(AudioScrobblerTSVImport)
class AudioScrobblerTSVImportAdmin(ImportBaseAdmin): ...
class AudioScrobblerTSVImportAdmin(ImportBaseAdmin):
...
@admin.register(LastFmImport)
class LastFmImportAdmin(ImportBaseAdmin): ...
class LastFmImportAdmin(ImportBaseAdmin):
...
@admin.register(KoReaderImport)
class KoReaderImportAdmin(ImportBaseAdmin): ...
class KoReaderImportAdmin(ImportBaseAdmin):
...
@admin.register(RetroarchImport)
class RetroarchImportAdmin(ImportBaseAdmin): ...
class RetroarchImportAdmin(ImportBaseAdmin):
...
@admin.register(Genre)

1
vrobbler/apps/scrobbles/context_processors.py
View File

@ -4,6 +4,7 @@ from django.utils import timezone
from scrobbles.constants import EXCLUDE_FROM_NOW_PLAYING
from scrobbles.models import Scrobble
def now_playing(request):
user = request.user
now = timezone.now()

4
vrobbler/apps/scrobbles/management/commands/send_mood_checkin.py
View File

@ -1,7 +1,5 @@
from django.core.management.base import BaseCommand
from vrobbler.apps.scrobbles.utils import (
send_mood_checkin_reminders
)
from vrobbler.apps.scrobbles.utils import send_mood_checkin_reminders
class Command(BaseCommand):

163
vrobbler/apps/scrobbles/models.py
View File

@ -58,7 +58,9 @@ logger = logging.getLogger(__name__)
User = get_user_model()
BNULL = {"blank": True, "null": True}
POINTS_FOR_MOVEMENT_HISTORY = int(getattr(settings, "POINTS_FOR_MOVEMENT_HISTORY", 3))
POINTS_FOR_MOVEMENT_HISTORY = int(
getattr(settings, "POINTS_FOR_MOVEMENT_HISTORY", 3)
)
class BaseFileImportMixin(TimeStampedModel):
@ -101,7 +103,9 @@ class BaseFileImportMixin(TimeStampedModel):
scrobble_id = line.split("\t")[0]
scrobble = Scrobble.objects.filter(id=scrobble_id).first()
if not scrobble:
logger.warning(f"Could not find scrobble {scrobble_id} to undo")
logger.warning(
f"Could not find scrobble {scrobble_id} to undo"
)
continue
logger.info(f"Removing scrobble {scrobble_id}")
if not dryrun:
@ -144,7 +148,9 @@ class BaseFileImportMixin(TimeStampedModel):
return
for count, scrobble in enumerate(scrobbles):
scrobble_str = f"{scrobble.id}\t{scrobble.timestamp}\t{scrobble.media_obj}"
scrobble_str = (
f"{scrobble.id}\t{scrobble.timestamp}\t{scrobble.media_obj}"
)
log_line = f"{scrobble_str}"
if count > 0:
log_line = "\n" + log_line
@ -166,7 +172,9 @@ class KoReaderImport(BaseFileImportMixin):
return "KOReader"
def get_absolute_url(self):
return reverse("scrobbles:koreader-import-detail", kwargs={"slug": self.uuid})
return reverse(
"scrobbles:koreader-import-detail", kwargs={"slug": self.uuid}
)
def get_path(instance, filename):
extension = filename.split(".")[-1]
@ -202,11 +210,15 @@ class KoReaderImport(BaseFileImportMixin):
fix_profile_historic_timezones(self.user.profile)
if self.processed_finished and not force:
logger.info(f"{self} already processed on {self.processed_finished}")
logger.info(
f"{self} already processed on {self.processed_finished}"
)
return
self.mark_started()
scrobbles = process_koreader_sqlite_file(self.upload_file_path, self.user.id)
scrobbles = process_koreader_sqlite_file(
self.upload_file_path, self.user.id
)
self.record_log(scrobbles)
self.mark_finished()
@ -220,7 +232,9 @@ class AudioScrobblerTSVImport(BaseFileImportMixin):
return "AudiosScrobbler"
def get_absolute_url(self):
return reverse("scrobbles:tsv-import-detail", kwargs={"slug": self.uuid})
return reverse(
"scrobbles:tsv-import-detail", kwargs={"slug": self.uuid}
)
def get_path(instance, filename):
extension = filename.split(".")[-1]
@ -244,12 +258,16 @@ class AudioScrobblerTSVImport(BaseFileImportMixin):
fix_profile_historic_timezones(self.user.profile)
if self.processed_finished and not force:
logger.info(f"{self} already processed on {self.processed_finished}")
logger.info(
f"{self} already processed on {self.processed_finished}"
)
return
self.mark_started()
scrobbles = import_audioscrobbler_tsv_file(self.upload_file_path, self.user.id)
scrobbles = import_audioscrobbler_tsv_file(
self.upload_file_path, self.user.id
)
self.record_log(scrobbles)
self.mark_finished()
@ -263,7 +281,9 @@ class LastFmImport(BaseFileImportMixin):
return "LastFM"
def get_absolute_url(self):
return reverse("scrobbles:lastfm-import-detail", kwargs={"slug": self.uuid})
return reverse(
"scrobbles:lastfm-import-detail", kwargs={"slug": self.uuid}
)
def process(self, import_all=False):
"""Import scrobbles found on LastFM"""
@ -272,7 +292,9 @@ class LastFmImport(BaseFileImportMixin):
fix_profile_historic_timezones(self.user.profile)
if self.processed_finished:
logger.info(f"{self} already processed on {self.processed_finished}")
logger.info(
f"{self} already processed on {self.processed_finished}"
)
return
last_import = None
@ -310,7 +332,9 @@ class RetroarchImport(BaseFileImportMixin):
return "Retroarch"
def get_absolute_url(self):
return reverse("scrobbles:retroarch-import-detail", kwargs={"slug": self.uuid})
return reverse(
"scrobbles:retroarch-import-detail", kwargs={"slug": self.uuid}
)
def process(self, import_all=False, force=False):
"""Import scrobbles found on Retroarch"""
@ -318,7 +342,9 @@ class RetroarchImport(BaseFileImportMixin):
fix_profile_historic_timezones(self.user.profile)
if self.processed_finished and not force:
logger.info(f"{self} already processed on {self.processed_finished}")
logger.info(
f"{self} already processed on {self.processed_finished}"
)
return
if force:
@ -509,25 +535,39 @@ class Scrobble(TimeStampedModel):
podcast_episode = models.ForeignKey(
PodcastEpisode, on_delete=models.DO_NOTHING, **BNULL
)
sport_event = models.ForeignKey(SportEvent, on_delete=models.DO_NOTHING, **BNULL)
sport_event = models.ForeignKey(
SportEvent, on_delete=models.DO_NOTHING, **BNULL
)
book = models.ForeignKey(Book, on_delete=models.DO_NOTHING, **BNULL)
paper = models.ForeignKey(Paper, on_delete=models.DO_NOTHING, **BNULL)
video_game = models.ForeignKey(VideoGame, on_delete=models.DO_NOTHING, **BNULL)
board_game = models.ForeignKey(BoardGame, on_delete=models.DO_NOTHING, **BNULL)
geo_location = models.ForeignKey(GeoLocation, on_delete=models.DO_NOTHING, **BNULL)
video_game = models.ForeignKey(
VideoGame, on_delete=models.DO_NOTHING, **BNULL
)
board_game = models.ForeignKey(
BoardGame, on_delete=models.DO_NOTHING, **BNULL
)
geo_location = models.ForeignKey(
GeoLocation, on_delete=models.DO_NOTHING, **BNULL
)
beer = models.ForeignKey(Beer, on_delete=models.DO_NOTHING, **BNULL)
puzzle = models.ForeignKey(Puzzle, on_delete=models.DO_NOTHING, **BNULL)
food = models.ForeignKey(Food, on_delete=models.DO_NOTHING, **BNULL)
trail = models.ForeignKey(Trail, on_delete=models.DO_NOTHING, **BNULL)
task = models.ForeignKey(Task, on_delete=models.DO_NOTHING, **BNULL)
web_page = models.ForeignKey(WebPage, on_delete=models.DO_NOTHING, **BNULL)
life_event = models.ForeignKey(LifeEvent, on_delete=models.DO_NOTHING, **BNULL)
life_event = models.ForeignKey(
LifeEvent, on_delete=models.DO_NOTHING, **BNULL
)
mood = models.ForeignKey(Mood, on_delete=models.DO_NOTHING, **BNULL)
brick_set = models.ForeignKey(BrickSet, on_delete=models.DO_NOTHING, **BNULL)
brick_set = models.ForeignKey(
BrickSet, on_delete=models.DO_NOTHING, **BNULL
)
media_type = models.CharField(
max_length=14, choices=MediaType.choices, default=MediaType.VIDEO
)
user = models.ForeignKey(User, blank=True, null=True, on_delete=models.DO_NOTHING)
user = models.ForeignKey(
User, blank=True, null=True, on_delete=models.DO_NOTHING
)
# Time keeping
timestamp = models.DateTimeField(**BNULL)
@ -555,7 +595,9 @@ class Scrobble(TimeStampedModel):
upload_to="scrobbles/videogame_save_data/", **BNULL
)
gpx_file = models.FileField(upload_to="scrobbles/gpx_file/", **BNULL)
screenshot = models.ImageField(upload_to="scrobbles/videogame_screenshot/", **BNULL)
screenshot = models.ImageField(
upload_to="scrobbles/videogame_screenshot/", **BNULL
)
screenshot_small = ImageSpecField(
source="screenshot",
processors=[ResizeToFit(100, 100)],
@ -577,7 +619,12 @@ class Scrobble(TimeStampedModel):
models.Index(fields=["media_type"]),
models.Index(fields=["source_id"]),
models.Index(
fields=["user", "in_progress", "played_to_completion", "is_paused"]
fields=[
"user",
"in_progress",
"played_to_completion",
"is_paused",
]
),
]
@ -646,7 +693,9 @@ class Scrobble(TimeStampedModel):
from scrobbles.models import Scrobble
if self.logdata and self.logdata.serial_scrobble_id:
return Scrobble.objects.filter(id=self.logdata.serial_scrobble_id).first()
return Scrobble.objects.filter(
id=self.logdata.serial_scrobble_id
).first()
@property
def finish_url(self) -> str:
@ -669,7 +718,8 @@ class Scrobble(TimeStampedModel):
self.media_type = self.MediaType(self.media_obj.__class__.__name__)
if (self.timestamp and self.stop_timestamp) and (
not self.playback_position_seconds or self.playback_position_seconds <= 0
not self.playback_position_seconds
or self.playback_position_seconds <= 0
):
self.playback_position_seconds = (
self.stop_timestamp - self.timestamp
@ -684,9 +734,9 @@ class Scrobble(TimeStampedModel):
return reverse("scrobbles:detail", kwargs={"uuid": self.uuid})
def push_to_archivebox(self):
pushable_media = hasattr(self.media_obj, "push_to_archivebox") and callable(
self.media_obj.push_to_archivebox
)
pushable_media = hasattr(
self.media_obj, "push_to_archivebox"
) and callable(self.media_obj.push_to_archivebox)
if pushable_media and self.user.profile.archivebox_url:
try:
@ -750,7 +800,10 @@ class Scrobble(TimeStampedModel):
logger.info(f"Redirecting to {self.media_obj.url}")
redirect_url = self.media_obj.url
if self.media_type == self.MediaType.VIDEO and self.media_obj.youtube_id:
if (
self.media_type == self.MediaType.VIDEO
and self.media_obj.youtube_id
):
redirect_url = self.media_obj.youtube_link
return redirect_url
@ -766,7 +819,9 @@ class Scrobble(TimeStampedModel):
@property
def local_stop_timestamp(self):
if self.stop_timestamp:
return timezone.localtime(self.stop_timestamp, timezone=self.tzinfo)
return timezone.localtime(
self.stop_timestamp, timezone=self.tzinfo
)
@property
def scrobble_media_key(self) -> str:
@ -894,7 +949,9 @@ class Scrobble(TimeStampedModel):
long_play_secs = 0
if self.previous and not self.previous.long_play_complete:
long_play_secs = self.previous.long_play_seconds or 0
percent = int(((playback_seconds + long_play_secs) / run_time_secs) * 100)
percent = int(
((playback_seconds + long_play_secs) / run_time_secs) * 100
)
return percent
@ -939,7 +996,9 @@ class Scrobble(TimeStampedModel):
expected_end = self.timestamp + datetime.timedelta(
seconds=self.media_obj.run_time_seconds
)
expected_end_padded = expected_end + datetime.timedelta(seconds=padding_seconds)
expected_end_padded = expected_end + datetime.timedelta(
seconds=padding_seconds
)
# Take our start time, add our media length and an extra 30 min (1800s) is it still in the future? keep going
is_in_progress = expected_end_padded > pendulum.now()
logger.info(
@ -1002,9 +1061,11 @@ class Scrobble(TimeStampedModel):
@classmethod
def by_date(cls, media_type: str = "Track"):
cls.objects.filter(media_type=media_type).values("timestamp__date").annotate(
count=models.Count("id")
).values("timestamp__date", "count").order_by(
cls.objects.filter(media_type=media_type).values(
"timestamp__date"
).annotate(count=models.Count("id")).values(
"timestamp__date", "count"
).order_by(
"-count",
)
@ -1116,7 +1177,9 @@ class Scrobble(TimeStampedModel):
"source_id": source_id,
},
)
scrobble_data["playback_status"] = scrobble_data.pop("status", None)
scrobble_data["playback_status"] = scrobble_data.pop(
"status", None
)
return existing_by_source_id.update(scrobble_data)
# Find our last scrobble of this media item (track, video, etc)
@ -1136,7 +1199,9 @@ class Scrobble(TimeStampedModel):
logger.warning(
f"[create_or_update] geoloc requires create_or_update_location"
)
scrobble = cls.create_or_update_location(media, scrobble_data, user_id)
scrobble = cls.create_or_update_location(
media, scrobble_data, user_id
)
return scrobble
if not skip_in_progress_check or read_log_page:
@ -1149,7 +1214,9 @@ class Scrobble(TimeStampedModel):
"scrobble_data": scrobble_data,
},
)
scrobble_data["playback_status"] = scrobble_data.pop("status", None)
scrobble_data["playback_status"] = scrobble_data.pop(
"status", None
)
# If it's marked as stopped, send it through our update mechanism, which will complete it
if scrobble and (
scrobble.can_be_updated
@ -1161,8 +1228,12 @@ class Scrobble(TimeStampedModel):
if page_list:
for page in page_list:
if not page.get("end_ts", None):
page["end_ts"] = int(timezone.now().timestamp())
page["duration"] = page["end_ts"] - page.get("start_ts")
page["end_ts"] = int(
timezone.now().timestamp()
)
page["duration"] = page["end_ts"] - page.get(
"start_ts"
)
page_list.append(
BookPageLogData(
@ -1198,9 +1269,9 @@ class Scrobble(TimeStampedModel):
"source": source,
},
)
if mtype == cls.MediaType.FOOD and not scrobble_data.get("log", {}).get(
"calories", None
):
if mtype == cls.MediaType.FOOD and not scrobble_data.get(
"log", {}
).get("calories", None):
if media.calories:
scrobble_data["log"] = FoodLogData(calories=media.calories)
@ -1287,7 +1358,9 @@ class Scrobble(TimeStampedModel):
if existing_locations := location.in_proximity(named=True):
existing_location = existing_locations.first()
ts = int(pendulum.now().timestamp())
scrobble.log[ts] = f"Location {location.id} too close to this scrobble"
scrobble.log[
ts
] = f"Location {location.id} too close to this scrobble"
scrobble.save(update_fields=["log"])
logger.info(
f"[scrobbling] finished - found existing named location",
@ -1466,7 +1539,9 @@ class Scrobble(TimeStampedModel):
"""Returns true if our media is beyond our completion percent, unless
our type is geolocation in which case we always return false
"""
beyond_completion = self.percent_played >= self.media_obj.COMPLETION_PERCENT
beyond_completion = (
self.percent_played >= self.media_obj.COMPLETION_PERCENT
)
if self.media_type == "GeoLocation":
logger.info(

22
vrobbler/apps/scrobbles/notifications.py
View File

@ -5,6 +5,7 @@ from django.conf import settings
from django.contrib.sites.models import Site
from django.urls import reverse
class BasicNtfyNotification(ABC):
ntfy_headers: dict = {}
ntfy_url: str = ""
@ -14,12 +15,13 @@ class BasicNtfyNotification(ABC):
self.profile = profile
protocol = "http" if settings.DEBUG else "https"
domain = Site.objects.get_current().domain
self.url_tmpl = f'{protocol}://{domain}' + '{path}'
self.url_tmpl = f"{protocol}://{domain}" + "{path}"
@abstractmethod
def send(self) -> None:
pass
class ScrobbleNotification(BasicNtfyNotification):
scrobble: "Scrobble"
@ -29,8 +31,7 @@ class ScrobbleNotification(BasicNtfyNotification):
self.media_obj = scrobble.media_obj
protocol = "http" if settings.DEBUG else "https"
domain = Site.objects.get_current().domain
self.url_tmpl = f'{protocol}://{domain}' + '{path}'
self.url_tmpl = f"{protocol}://{domain}" + "{path}"
@abstractmethod
def send(self) -> None:
@ -41,13 +42,21 @@ class ScrobbleNtfyNotification(ScrobbleNotification):
def __init__(self, scrobble, **kwargs):
super().__init__(scrobble)
self.ntfy_str: str = f"{self.scrobble.media_obj}"
self.click_url = self.url_tmpl.format(path=self.media_obj.get_absolute_url())
self.click_url = self.url_tmpl.format(
path=self.media_obj.get_absolute_url()
)
self.title = self.media_obj.strings.verb
if kwargs.get("end", False):
self.click_url = self.url_tmpl.format(path=self.scrobble.finish_url)
self.click_url = self.url_tmpl.format(
path=self.scrobble.finish_url
)
self.title = "Finish " + self.media_obj.strings.verb.lower() + "?"
if self.scrobble.log and isinstance(self.scrobble.log, dict) and self.scrobble.log.get("title"):
if (
self.scrobble.log
and isinstance(self.scrobble.log, dict)
and self.scrobble.log.get("title")
):
self.ntfy_str += f" - {self.scrobble.log.get('title')}"
def send(self):
@ -68,6 +77,7 @@ class ScrobbleNtfyNotification(ScrobbleNotification):
},
)
class MoodNtfyNotification(BasicNtfyNotification):
def __init__(self, profile, **kwargs):
super().__init__(profile)

51
vrobbler/apps/scrobbles/utils.py
View File

@ -334,6 +334,7 @@ def send_stop_notifications_for_in_progress_scrobbles() -> int:
return notifications_sent
def send_mood_checkin_reminders() -> int:
"""Get all profiles with mood check-ins enabled and checkin!"""
from profiles.models import UserProfile
@ -357,19 +358,32 @@ def extract_domain(url):
)
return domain
def fix_playback_position_seconds(*scrobbles: "Scrobble", commit=True) -> list["Scrobble"]:
def fix_playback_position_seconds(
*scrobbles: "Scrobble", commit=True
) -> list["Scrobble"]:
updated_scrobbles = list()
for scrobble in scrobbles:
if not scrobble.media_obj:
logger.info(f"No media object found for scrobble {scrobble.id}, cannot update elapsed time")
logger.info(
f"No media object found for scrobble {scrobble.id}, cannot update elapsed time"
)
continue
if scrobble.media_type == "Track" and scrobble.media_obj.run_time_seconds:
too_long = scrobble.playback_position_seconds > scrobble.media_obj.run_time_seconds
if (
scrobble.media_type == "Track"
and scrobble.media_obj.run_time_seconds
):
too_long = (
scrobble.playback_position_seconds
> scrobble.media_obj.run_time_seconds
)
zero = scrobble.playback_position_seconds == 0
null = not scrobble.playback_position_seconds
if too_long or zero or null:
scrobble.playback_position_seconds = scrobble.media_obj.run_time_seconds
scrobble.playback_position_seconds = (
scrobble.media_obj.run_time_seconds
)
updated_scrobbles.append(scrobble)
if commit:
scrobble.save(update_fields=["playback_position_seconds"])
@ -382,13 +396,17 @@ def base_scrobble_qs(user_id: int) -> models.QuerySet:
from scrobbles.models import Scrobble
return (
Scrobble.objects
.annotate(day=TruncDate("timestamp"))
.annotate(calories_int=Cast(KeyTextTransform("calories", "log"), models.IntegerField()))
Scrobble.objects.annotate(day=TruncDate("timestamp"))
.annotate(
calories_int=Cast(
KeyTextTransform("calories", "log"), models.IntegerField()
)
)
.filter(calories_int__isnull=False, user_id=user_id)
)
def get_daily_calories_for_user_by_day(user_id: int, date: date| str) -> int:
def get_daily_calories_for_user_by_day(user_id: int, date: date | str) -> int:
"""Return total calories for a user on a specific day."""
if isinstance(date, str):
@ -403,6 +421,7 @@ def get_daily_calories_for_user_by_day(user_id: int, date: date| str) -> int:
return agg.get("total_calories") or 0
def get_daily_calorie_dict_for_user(user_id: int) -> dict[date, int]:
"""Return {day: total_calories} for all days with scrobbles, in one query."""
qs = (
@ -414,18 +433,20 @@ def get_daily_calorie_dict_for_user(user_id: int) -> dict[date, int]:
return {entry["day"]: entry["total_calories"] for entry in qs}
def remove_last_part(url: str) -> str:
url = url.rstrip('/')
if '/' not in url:
url = url.rstrip("/")
if "/" not in url:
return url
return url.rsplit('/', 1)[0]
return url.rsplit("/", 1)[0]
def next_url_if_exists(url: str) -> str:
# Normalize (remove trailing slash)
url = url.rstrip('/')
url = url.rstrip("/")
# Find last number in the URL path
match = re.search(r'(\d+)(?:/?$)', url)
match = re.search(r"(\d+)(?:/?$)", url)
if not match:
logger.info("No numeric segment found in the URL", extra={"url": url})
return ""
@ -434,7 +455,7 @@ def next_url_if_exists(url: str) -> str:
new_number = number + 1
# Replace only the last occurrence of that number
new_url = re.sub(rf'{number}(?:/?$)', f'{new_number}/', url + '/', 1)
new_url = re.sub(rf"{number}(?:/?$)", f"{new_number}/", url + "/", 1)
# Check if the new URL exists
try:

42
vrobbler/apps/scrobbles/views.py
View File

@ -161,13 +161,13 @@ class RecentScrobbleList(ListView):
next_date = date + timedelta(weeks=+2)
prev_date = date + timedelta(weeks=-1)
if date.isocalendar()[1] < today.isocalendar()[1]:
data["next_link"] = (
f"?date={next_date.strftime('%Y-W%W')}"
)
data[
"next_link"
] = f"?date={next_date.strftime('%Y-W%W')}"
data["prev_link"] = f"?date={prev_date.strftime('%Y-W%W')}"
data["title"] = (
f"Week {date.strftime('%-W')} of {date.year}"
)
data[
"title"
] = f"Week {date.strftime('%-W')} of {date.year}"
data = data | Scrobble.as_dict_by_type(
Scrobble.for_week(
user_id, date.year, date.isocalendar()[1]
@ -177,9 +177,9 @@ class RecentScrobbleList(ListView):
next_date = date + relativedelta(months=1)
prev_date = date + relativedelta(months=-1)
if date.month < today.month:
data["next_link"] = (
f"?date={next_date.strftime('%Y-%m')}"
)
data[
"next_link"
] = f"?date={next_date.strftime('%Y-%m')}"
data["title"] = f"{date.strftime('%B %Y')}"
data["prev_link"] = f"?date={prev_date.strftime('%Y-%m')}"
data = data | Scrobble.as_dict_by_type(
@ -188,20 +188,20 @@ class RecentScrobbleList(ListView):
elif date_str == "today" or date_str.count("-") == 2:
next_date = date + timedelta(days=1)
prev_date = date - timedelta(days=1)
data["prev_link"] = (
f"?date={prev_date.strftime('%Y-%m-%d')}"
)
data[
"prev_link"
] = f"?date={prev_date.strftime('%Y-%m-%d')}"
if date < today:
data["next_link"] = (
f"?date={next_date.strftime('%Y-%m-%d')}"
)
data[
"next_link"
] = f"?date={next_date.strftime('%Y-%m-%d')}"
if date == today:
data["title"] = "Today"
else:
data["title"] = f"{date.strftime('%Y-%m-%d')}"
data["today_link"] = (
f"?date={today.strftime('%Y-%m-%d')}"
)
data[
"today_link"
] = f"?date={today.strftime('%Y-%m-%d')}"
data = data | Scrobble.as_dict_by_type(
Scrobble.for_day(
user_id, date.year, date.month, date.day
@ -228,9 +228,9 @@ class RecentScrobbleList(ListView):
data = data | Scrobble.as_dict_by_type(
Scrobble.for_day(user_id, date.year, date.month, date.day)
)
data["today_link"] = (
"" # f"?date={today.strftime('%Y-%m-%d')}"
)
data[
"today_link"
] = "" # f"?date={today.strftime('%Y-%m-%d')}"
data["active_imports"] = AudioScrobblerTSVImport.objects.filter(
processing_started__isnull=False,

1
vrobbler/apps/trails/api/views.py
View File

@ -1,4 +1,3 @@
from rest_framework import permissions, viewsets
from trails.api import serializers

5
vrobbler/apps/videos/admin.py
View File

@ -21,7 +21,10 @@ class SeriesAdmin(admin.ModelAdmin):
@admin.register(Video)
class VideoAdmin(admin.ModelAdmin):
date_hierarchy = "created"
raw_id_fields = ("tv_series","channel",)
raw_id_fields = (
"tv_series",
"channel",
)
list_display = (
"title",
"video_type",

8
vrobbler/apps/videos/api/serializers.py
View File

@ -15,8 +15,12 @@ class SeriesSerializer(serializers.HyperlinkedModelSerializer):
class VideoSerializer(serializers.HyperlinkedModelSerializer):
channel = serializers.PrimaryKeyRelatedField(queryset=Channel.objects.all())
tv_series = serializers.PrimaryKeyRelatedField(queryset=Series.objects.all(), required=False, allow_null=True)
channel = serializers.PrimaryKeyRelatedField(
queryset=Channel.objects.all()
)
tv_series = serializers.PrimaryKeyRelatedField(
queryset=Series.objects.all(), required=False, allow_null=True
)
class Meta:
model = Video

17
vrobbler/apps/videos/management/commands/find_video.py
View File

@ -3,25 +3,34 @@ from django.core import serializers
from videos.models import Video, Series, Channel
import json
class Command(BaseCommand):
help = "Find or create a Video by ID and output it as JSON"
def add_arguments(self, parser):
parser.add_argument("video_id", type=str, help="The video ID to find or create")
parser.add_argument(
"video_id", type=str, help="The video ID to find or create"
)
def handle(self, *args, **options):
instance = Video.find_or_create(options.get("video_id", ""), overwrite=True)
instance = Video.find_or_create(
options.get("video_id", ""), overwrite=True
)
data = json.loads(serializers.serialize("json", [instance]))[0]
# --- Enrich with series model ---
if instance.tv_series_id:
series_instance = instance.tv_series
series_json = json.loads(serializers.serialize("json", [series_instance]))[0]
series_json = json.loads(
serializers.serialize("json", [series_instance])
)[0]
data["series"] = series_json # new nested field
if instance.channel_id:
channel_instance = instance.channel
channel_json = json.loads(serializers.serialize("json", [channel_instance]))[0]
channel_json = json.loads(
serializers.serialize("json", [channel_instance])
)[0]
data["channel"] = channel_json # new nested field
self.stdout.write(json.dumps(data, indent=2))

5
vrobbler/apps/videos/sources/imdb.py
View File

@ -25,9 +25,7 @@ def lookup_video_from_imdb(imdb_id: str) -> VideoMetadata:
return None
if imdb_result.type_id.lower() in ["movie", "tvepisode"]:
video_metadata.base_run_time_seconds = (
imdb_result.runtime * 60
)
video_metadata.base_run_time_seconds = imdb_result.runtime * 60
video_metadata.imdb_id = imdb_id
video_metadata.title = imdb_result.title
video_metadata.year = imdb_result.year
@ -47,5 +45,4 @@ def lookup_video_from_imdb(imdb_id: str) -> VideoMetadata:
video_metadata.season_number = imdb_result.season
video_metadata.next_imdb_id = imdb_result.next_episode_id
return video_metadata

12
vrobbler/apps/videos/sources/tmdb.py
View File

@ -46,7 +46,9 @@ def lookup_video_from_tmdb(
media = Movie().details(tmdb_result.movie_results[0].id)
video_metadata.video_type = VideoType.MOVIE.value
video_metadata.title = media.title
video_metadata.cover_url = TMDB_IMAGE_URL + media.poster_path # TODO: enrich this with TMDB url
video_metadata.cover_url = (
TMDB_IMAGE_URL + media.poster_path
) # TODO: enrich this with TMDB url
video_metadata.year = pendulum.parse(media.release_date).year
video_metadata.genres = [g.get("name", "") for g in media.genres]
@ -54,7 +56,9 @@ def lookup_video_from_tmdb(
video_metadata.video_type = VideoType.TV_EPISODE.value
media = tmdb_result.tv_episode_results[0]
video_metadata.title = media.name
video_metadata.cover_url = TMDB_IMAGE_URL + media.still_path # TODO: enrich this with TMDB url
video_metadata.cover_url = (
TMDB_IMAGE_URL + media.still_path
) # TODO: enrich this with TMDB url
video_metadata.episode_number = media.episode_number
video_metadata.season_number = media.season_number
video_metadata.year = media.air_date.year
@ -68,7 +72,7 @@ def lookup_video_from_tmdb(
video_metadata.tv_series_id = series.id
if not media:
logger.warning("Video not found on TMDB", extra={"imdb_id":imdb_id})
logger.warning("Video not found on TMDB", extra={"imdb_id": imdb_id})
return video_metadata
video_metadata.tmdb_id = media.id
@ -76,6 +80,6 @@ def lookup_video_from_tmdb(
video_metadata.plot = media.overview
video_metadata.overview = media.overview
video_metadata.tmdb_rating = media.vote_average
#video_metadata.next_imdb_id = imdb_result.get("next episode", None)
# video_metadata.next_imdb_id = imdb_result.get("next episode", None)
return video_metadata

2
vrobbler/apps/videos/urls.py
View File

@ -12,7 +12,7 @@ urlpatterns = [
views.SeriesDetailView.as_view(),
name="series_detail",
),
path('videos/', views.VideoListView.as_view(), name='video_list'),
path("videos/", views.VideoListView.as_view(), name="video_list"),
path(
"video/<slug:slug>/",
views.VideoDetailView.as_view(),

11
vrobbler/apps/videos/utils.py
View File

@ -2,13 +2,16 @@ import logging
from videos.models import Video
from django.db import IntegrityError
#from videos.skatevideosite import lookup_video_from_skatevideosite
# from videos.skatevideosite import lookup_video_from_skatevideosite
logger = logging.getLogger(__name__)
def clean_up_videos():
videos = Video.objects.filter(imdb_id__isnull=False).exclude(imdb_id__icontains="tt")
videos = Video.objects.filter(imdb_id__isnull=False).exclude(
imdb_id__icontains="tt"
)
for video in videos:
logger.info(f"Fixing imdb_id for {video}")
@ -16,7 +19,9 @@ def clean_up_videos():
try:
video.save(update_fields=["imdb_id"])
except IntegrityError:
new_video = Video.objects.filter(imdb_id="tt" + video.imdb_id).first()
new_video = Video.objects.filter(
imdb_id="tt" + video.imdb_id
).first()
video.scrobble_set.all().update(video=new_video)
video.delete()

6
vrobbler/apps/videos/views.py
View File

@ -25,9 +25,9 @@ class SeriesDetailView(LoginRequiredMixin, generic.DetailView):
context_data = super().get_context_data(**kwargs)
context_data["scrobbles"] = self.object.scrobbles_for_user(user_id)
next_episode_id = self.object.last_scrobbled_episode(
user_id
).next_imdb_id or ""
next_episode_id = (
self.object.last_scrobbled_episode(user_id).next_imdb_id or ""
)
if self.object.is_episode_playing(user_id):
next_episode_id = ""
if next_episode_id:

1
vrobbler/apps/webpages/forms.py
View File

@ -1,6 +1,7 @@
from django import forms
from django.urls import reverse_lazy
class WebPageReadForm(forms.Form):
notes = forms.CharField(widget=forms.Textarea, required=False)
success_url = reverse_lazy("vrobbler-home")

36
vrobbler/settings.py
View File

@ -37,7 +37,9 @@ TESTING = len(sys.argv) > 1 and sys.argv[1] == "test"
TAGGIT_CASE_INSENSITIVE = True
KEEP_DETAILED_SCROBBLE_LOGS = os.getenv("VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS", False)
KEEP_DETAILED_SCROBBLE_LOGS = os.getenv(
"VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS", False
)
# Key must be 16, 24 or 32 bytes long and will be converted to a byte stream
ENCRYPTED_FIELD_KEY = os.getenv(
@ -52,7 +54,9 @@ DELETE_STALE_SCROBBLES = (
)
# Used to dump data coming from srobbling sources, helpful for building new inputs
DUMP_REQUEST_DATA = os.getenv("VROBBLER_DUMP_REQUEST_DATA", "false").lower() in TRUTHY
DUMP_REQUEST_DATA = (
os.getenv("VROBBLER_DUMP_REQUEST_DATA", "false").lower() in TRUTHY
)
USDA_API_KEY = os.getenv("VROBBLER_USDA_API_KEY")
THESPORTSDB_API_KEY = os.getenv("VROBBLER_THESPORTSDB_API_KEY", "2")
@ -68,7 +72,9 @@ COMICVINE_API_KEY = os.getenv("VROBBLER_COMICVINE_API_KEY")
BGG_ACCESS_TOKEN = os.getenv("VROBBLER_BGG_ACCESS_TOKEN", "")
GEOLOC_ACCURACY = os.getenv("VROBBLER_GEOLOC_ACCURACY", 3)
GEOLOC_PROXIMITY = os.getenv("VROBBLER_GEOLOC_PROXIMITY", "0.0001")
POINTS_FOR_MOVEMENT_HISTORY = os.getenv("VROBBLER_POINTS_FOR_MOVEMENT_HISTORY", 3)
POINTS_FOR_MOVEMENT_HISTORY = os.getenv(
"VROBBLER_POINTS_FOR_MOVEMENT_HISTORY", 3
)
TODOIST_CLIENT_ID = os.getenv("VROBBLER_TODOIST_CLIENT_ID", "")
TODOIST_CLIENT_SECRET = os.getenv("VROBBLER_TODOIST_CLIENT_SECRET", "")
@ -91,7 +97,9 @@ DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
TIME_ZONE = os.getenv("VROBBLER_TIME_ZONE", "America/New_York")
ALLOWED_HOSTS = ["*"]
CSRF_TRUSTED_ORIGINS = [os.getenv("VROBBLER_TRUSTED_ORIGINS", "http://localhost:8000")]
CSRF_TRUSTED_ORIGINS = [
os.getenv("VROBBLER_TRUSTED_ORIGINS", "http://localhost:8000")
]
X_FRAME_OPTIONS = "SAMEORIGIN"
REDIS_URL = os.getenv("VROBBLER_REDIS_URL", None)
@ -100,7 +108,9 @@ if REDIS_URL:
else:
print("Eagerly running all tasks")
CELERY_TASK_ALWAYS_EAGER = os.getenv("VROBBLER_SKIP_CELERY", "false").lower() in TRUTHY
CELERY_TASK_ALWAYS_EAGER = (
os.getenv("VROBBLER_SKIP_CELERY", "false").lower() in TRUTHY
)
CELERY_BROKER_URL = REDIS_URL if REDIS_URL else "memory://localhost/"
CELERY_RESULT_BACKEND = "django-db"
CELERY_TIMEZONE = os.getenv("VROBBLER_TIME_ZONE", "America/New_York")
@ -201,7 +211,9 @@ DATABASES = {
}
if TESTING:
DATABASES = {"default": dj_database_url.config(default="sqlite:///testdb.sqlite3")}
DATABASES = {
"default": dj_database_url.config(default="sqlite:///testdb.sqlite3")
}
db_str = ""
if "sqlite" in DATABASES["default"]["ENGINE"]:
@ -237,7 +249,9 @@ REST_FRAMEWORK = {
"rest_framework.authentication.SessionAuthentication",
],
"DEFAULT_CONTENT_NEGOTIATION_CLASS": "vrobbler.negotiation.IgnoreClientContentNegotiation",
"DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
"DEFAULT_FILTER_BACKENDS": [
"django_filters.rest_framework.DjangoFilterBackend"
],
"DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
"PAGE_SIZE": 200,
}
@ -297,7 +311,9 @@ else:
STATIC_ROOT = os.getenv(
"VROBBLER_STATIC_ROOT", os.path.join(PROJECT_ROOT, "static")
)
MEDIA_ROOT = os.getenv("VROBBLER_MEDIA_ROOT", os.path.join(PROJECT_ROOT, "media"))
MEDIA_ROOT = os.getenv(
"VROBBLER_MEDIA_ROOT", os.path.join(PROJECT_ROOT, "media")
)
STATIC_URL = os.getenv("VROBBLER_STATIC_URL", "/static/")
MEDIA_URL = os.getenv("VROBBLER_MEDIA_URL", "/media/")
@ -381,7 +397,9 @@ LOGGING = {
},
}
LOG_TO_CONSOLE = os.getenv("VROBBLER_LOG_TO_CONSOLE", "false").lower() in TRUTHY
LOG_TO_CONSOLE = (
os.getenv("VROBBLER_LOG_TO_CONSOLE", "false").lower() in TRUTHY
)
if LOG_TO_CONSOLE:
LOGGING["loggers"]["django"]["handlers"] = ["console"]
LOGGING["loggers"]["vrobbler"]["handlers"] = ["console"]