Compare commits

..

1 Commits

Author SHA1 Message Date
fd6e0f49b6 Bump version to 0.12.0
Version 1.0 approaches!
2023-03-02 15:10:58 -05:00
281 changed files with 3793 additions and 16128 deletions

View File

@ -4,12 +4,12 @@
################
kind: pipeline
name: build & deploy
name: run_tests
steps:
# Run tests against Python/Flask engine backend (with pytest)
- name: pytest with coverage
image: python:3.11.1
image: python:3.10.4
commands:
# Install dependencies
- cp vrobbler.conf.test vrobbler.conf
@ -23,48 +23,6 @@ steps:
# Mount pip cache from host
- name: pip_cache
path: /root/.cache/pip
- name: deploy
image: appleboy/drone-ssh
settings:
host:
- vrobbler.service
username: root
ssh_key:
from_secret: ssh_key
command_timeout: 2m
script:
- pip uninstall -y vrobbler
- pip install git+https://code.unbl.ink/secstate/vrobbler.git@main
- vrobbler migrate
- vrobbler collectstatic --noinput
- immortalctl restart celery && immortalctl restart vrobbler
when:
branch:
- main
- name: build success notification
image: parrazam/drone-ntfy
when:
status: [success]
settings:
url: https://ntfy.unbl.ink
topic: drone
priority: low
tags:
- cd
- failure
- vrobbler
- name: build failure notification
image: parrazam/drone-ntfy
when:
status: [failure]
settings:
url: https://ntfy.unbl.ink
topic: drone
priority: high
tags:
- cd
- success
- vrobbler
volumes:
- name: docker
host:

View File

@ -1,7 +1,7 @@
Vrobbler
========
[![Build Status](https://ci.lab.unbl.ink/api/badges/secstate/vrobbler/status.svg?ref=refs/heads/main)](https://ci.lab.unbl.ink/secstate/vrobbler)
[![Build Status](https://ci.unbl.ink/api/badges/secstate/vrobbler/status.svg?ref=refs/heads/main)](https://ci.unbl.ink/secstate/vrobbler)
Vrobbler is a pretty simple Django-powered web app for scrobbling video plays from you favorite Jellyfin installation.

View File

@ -6,7 +6,7 @@ import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "vrobbler.settings")
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'vrobbler.settings')
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
@ -18,5 +18,5 @@ def main():
execute_from_command_line(sys.argv)
if __name__ == "__main__":
if __name__ == '__main__':
main()

5120
poetry.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -1,11 +1,11 @@
[tool.poetry]
name = "vrobbler"
version = "0.11.12"
version = "0.11.5"
description = ""
authors = ["Colin Powell <colin@unbl.ink>"]
[tool.poetry.dependencies]
python = ">=3.9,<4.0"
python = "^3.8"
Django = "^4.0.3"
django-extensions = "^3.1.5"
python-dateutil = "^2.8.2"
@ -26,6 +26,7 @@ django-taggit = "^2.1.0"
django-markdownify = "^0.9.1"
gunicorn = "^20.1.0"
django-simple-history = "^3.1.1"
whitenoise = "^6.3.0"
musicbrainzngs = "^0.7.1"
cinemagoer = "^2022.12.27"
pysportsdb = "^0.1.0"
@ -35,15 +36,6 @@ pylast = "^5.1.0"
django-encrypted-field = "^1.0.5"
celery = "^5.2.7"
honcho = "^1.1.0"
howlongtobeatpy = "^1.0.5"
beautifulsoup4 = "^4.11.2"
django-storages = "^1.13.2"
boto3 = "^1.26.98"
stream-sqlite = "^0.0.41"
ipython = "^8.14.0"
pendulum = "^2.1.2"
trafilatura = "^1.6.3"
django-imagekit = "^5.0.0"
[tool.poetry.dev-dependencies]
Werkzeug = "2.0.3"
@ -67,10 +59,11 @@ bandit = "^1.7.4"
minversion = "6.0"
addopts = "-ra -q"
testpaths = ["tests"]
DJANGO_SETTINGS_MODULE='vrobbler.settings-testing'
DJANGO_SETTINGS_MODULE='vrobbler.settings'
[tool.black]
line-length = 79
skip-string-normalization = true
target-version = ["py39", "py310"]
include = ".py$"
exclude = "migrations"

View File

@ -1,27 +0,0 @@
from boardgames.bgg import (
take_first,
lookup_boardgame_id_from_bgg,
lookup_boardgame_from_bgg,
)
def test_take_first():
assert take_first([]) == ""
assert take_first(["a", "b"]) == "a"
def test_lookup_boardgame_id_from_bgg():
bgg_id = lookup_boardgame_id_from_bgg("Cosmic Encounter")
assert bgg_id == "15"
bgg_id = lookup_boardgame_id_from_bgg("Comedy Encounter")
assert bgg_id == None
def test_lookup_boardgame_from_bgg():
bgg_result = lookup_boardgame_from_bgg(15)
assert bgg_result.get("bggeek_id") == 15
bgg_result = lookup_boardgame_from_bgg("Cosmic Encounter")
assert bgg_result.get("bggeek_id") == "15"

View File

@ -1,19 +0,0 @@
from vrobbler.apps.podcasts.scrapers import scrape_data_from_google_podcasts
expected_desc_snippet = (
"NPR's Up First is the news you need to start your day. "
)
expected_img_url = "https://encrypted-tbn2.gstatic.com/images?q=tbn:ANd9GcR1F0CfR24RR6sme531yIkCrnK4zzmo97jeualO5drVPKG6oCk"
expected_google_url = "https://podcasts.google.com/feed/aHR0cHM6Ly9mZWVkcy5ucHIub3JnLzUxMDMxOC9wb2RjYXN0LnhtbA"
def test_get_not_allowed_from_mopidy():
query = "Up First"
result_dict = scrape_data_from_google_podcasts(query)
assert result_dict["title"] == query
assert expected_desc_snippet in result_dict["description"]
assert result_dict["image_url"] == expected_img_url
assert result_dict["producer"] == "NPR"
assert result_dict["google_url"] == expected_google_url

View File

@ -1,6 +1,7 @@
import json
import pytest
from scrobbles.models import Scrobble
from rest_framework.authtoken.models import Token
from django.contrib.auth import get_user_model
@ -18,12 +19,12 @@ class MopidyRequest:
musicbrainz_track_id = "54214d63-5adf-4909-87cd-c65c37a6d558"
musicbrainz_album_id = "03b864cd-7761-314c-a892-05a89ddff00d"
musicbrainz_artist_id = "95f5b748-d370-47fe-85bd-0af2dc450bc0"
mopidy_uri = "local:track:Sublime%20-%20Sublime/Disc%201%20-%2004%20-%20Same%20in%20the%20End.mp3" # noqa
mopidy_uri = "local:track:Sublime%20-%20Sublime/Disc%201%20-%2004%20-%20Same%20in%20the%20End.mp3"
status = "resumed"
def __init__(self, **kwargs):
self.request_data = {
"name": kwargs.get("name", self.name),
"name": kwargs.get('name', self.name),
"artist": kwargs.get("artist", self.artist),
"album": kwargs.get("album", self.album),
"track_number": int(kwargs.get("track_number", self.track_number)),
@ -60,7 +61,7 @@ class MopidyRequest:
@pytest.fixture
def valid_auth_token():
user = User.objects.create(email="test@exmaple.com")
user = User.objects.create(email='test@exmaple.com')
return Token.objects.create(user=user).key

View File

@ -1,6 +1,4 @@
from datetime import datetime, timedelta
from unittest import mock
from unittest.mock import patch
import pytest
import time_machine
@ -8,17 +6,16 @@ from django.contrib.auth import get_user_model
from django.urls import reverse
from django.utils import timezone
from music.aggregators import live_charts, scrobble_counts, week_of_scrobbles
from music.models import Album, Artist
from profiles.models import UserProfile
from scrobbles.models import Scrobble
def build_scrobbles(client, request_data, num=7, spacing=2):
url = reverse("scrobbles:mopidy-webhook")
user = get_user_model().objects.create(username="Test User")
UserProfile.objects.create(user=user, timezone="US/Eastern")
url = reverse('scrobbles:mopidy-webhook')
user = get_user_model().objects.create(username='Test User')
UserProfile.objects.create(user=user, timezone='US/Eastern')
for i in range(num):
client.post(url, request_data, content_type="application/json")
client.post(url, request_data, content_type='application/json')
s = Scrobble.objects.last()
s.user = user
s.timestamp = timezone.now() - timedelta(days=i * spacing)
@ -33,40 +30,73 @@ def test_scrobble_counts_data(client, mopidy_track_request_data):
user = get_user_model().objects.first()
count_dict = scrobble_counts(user)
assert count_dict == {
"alltime": 7,
"month": 2,
"today": 1,
"week": 3,
"year": 7,
'alltime': 7,
'month': 2,
'today': 1,
'week': 3,
'year': 7,
}
@pytest.mark.django_db
@time_machine.travel(datetime(2022, 3, 4, 1, 24))
def test_live_charts(client, mopidy_track_request_data):
def test_week_of_scrobbles_data(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
week = week_of_scrobbles(user)
assert list(week.values()) == [1, 1, 1, 1, 1, 1, 1]
@pytest.mark.django_db
def test_top_tracks_by_day(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user)
assert tops[0].title == "Same in the End"
tops = live_charts(user, chart_period="week")
@pytest.mark.django_db
def test_top_tracks_by_week(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='week')
assert tops[0].title == "Same in the End"
tops = live_charts(user, chart_period="month")
@pytest.mark.django_db
def test_top_tracks_by_month(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='month')
assert tops[0].title == "Same in the End"
tops = live_charts(user, chart_period="year")
@pytest.mark.django_db
def test_top_tracks_by_year(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='year')
assert tops[0].title == "Same in the End"
tops = live_charts(user, chart_period="week", media_type="Artist")
@pytest.mark.django_db
def test_top__artists_by_week(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='week', media_type="Artist")
assert tops[0].name == "Sublime"
tops = live_charts(user, chart_period="month", media_type="Artist")
@pytest.mark.django_db
def test_top__artists_by_month(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='month', media_type="Artist")
assert tops[0].name == "Sublime"
tops = live_charts(user, chart_period="year", media_type="Artist")
@pytest.mark.django_db
def test_top__artists_by_year(client, mopidy_track_request_data):
build_scrobbles(client, mopidy_track_request_data, 7, 1)
user = get_user_model().objects.first()
tops = live_charts(user, chart_period='year', media_type="Artist")
assert tops[0].name == "Sublime"

View File

@ -1,11 +1,11 @@
import pytest
from videos.imdb import lookup_video_from_imdb
from vrobbler.apps.scrobbles.imdb import lookup_video_from_imdb
@pytest.mark.skip(reason="Need to sort out third party API testing")
def test_lookup_imdb_bad_id(caplog):
data = lookup_video_from_imdb("3409324")
data = lookup_video_from_imdb('3409324')
assert data is None
assert caplog.records[0].levelname == "WARNING"
assert caplog.records[0].msg == "IMDB ID should begin with 'tt' 3409324"

View File

@ -1,12 +0,0 @@
from datetime import datetime
import pytz
from django.contrib.auth import get_user_model
from vrobbler.apps.scrobbles.utils import timestamp_user_tz_to_utc
def test_timestamp_user_tz_to_utc():
timestamp = timestamp_user_tz_to_utc(
1685561082, pytz.timezone("US/Eastern")
)
assert timestamp == datetime(2023, 5, 31, 23, 24, 42, tzinfo=pytz.utc)

View File

@ -1,27 +1,30 @@
import json
import pytest
from django.urls import reverse
from scrobbles.models import Scrobble
from music.models import Track
from podcasts.models import Episode
from scrobbles.models import Scrobble
@pytest.mark.django_db
def test_get_not_allowed_from_mopidy(client, valid_auth_token):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.get(url, headers=headers)
assert response.status_code == 405
@pytest.mark.django_db
def test_bad_mopidy_request_data(client, valid_auth_token):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.post(url, headers)
assert response.status_code == 400
assert (
response.data["detail"]
== "JSON parse error - Expecting value: line 1 column 1 (char 0)"
response.data['detail']
== 'JSON parse error - Expecting value: line 1 column 1 (char 0)'
)
@ -29,23 +32,22 @@ def test_bad_mopidy_request_data(client, valid_auth_token):
def test_scrobble_mopidy_track(
client, mopidy_track_request_data, valid_auth_token
):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.post(
url,
mopidy_track_request_data,
content_type="application/json",
content_type='application/json',
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 1}
assert response.data == {'scrobble_id': 1}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Same in the End"
@pytest.mark.skip(reason="API is unstable")
@pytest.mark.django_db
def test_scrobble_mopidy_same_track_different_album(
client,
@ -53,28 +55,26 @@ def test_scrobble_mopidy_same_track_different_album(
mopidy_track_diff_album_request_data,
valid_auth_token,
):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.post(
url,
mopidy_track_request_data,
content_type="application/json",
content_type='application/json',
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 1}
scrobble = Scrobble.objects.last()
assert response.data == {'scrobble_id': 1}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.album.name == "Sublime"
response = client.post(
url,
mopidy_track_diff_album_request_data,
content_type="application/json",
content_type='application/json',
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 2}
scrobble = Scrobble.objects.last()
scrobble = Scrobble.objects.get(id=2)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.album.name == "Gold"
assert scrobble.media_obj.title == "Same in the End"
@ -84,16 +84,16 @@ def test_scrobble_mopidy_same_track_different_album(
def test_scrobble_mopidy_podcast(
client, mopidy_podcast_request_data, valid_auth_token
):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.post(
url,
mopidy_podcast_request_data,
content_type="application/json",
content_type='application/json',
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 1}
assert response.data == {'scrobble_id': 1}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Episode

119
todos.org
View File

@ -2,85 +2,14 @@
A fun way to keep track of things in the project to fix or improve.
* Version 1.0.0
** TODO What to do with Youtube videos from LastFM and web-scrobbler :bug:source:lastfm:
** TODO Add a user profile page with ability to change settings :profiles:improvement:
** DONE [#C] Consider a purge command for duplicated and stuck in-progress scrobbles :utililty:improvement:
CLOSED: [2023-04-06 Thu 14:09]
** DONE Add a "stop_timestamp" so we don't rely on content length :improvement:scrobbling:
CLOSED: [2023-04-02 Sun 23:58]
Essentially, we currently have the timestamp as when the content began
scrobbling and then calculate the finish time from the length of the content.
This works pretty well because we know how long most things are.
But in some cases, sports events or long podcasts, we may start mid-way through
an event or finish halfway through but still want to mark it as done. In these
cases, knowing the finish time could be useful, especially when interfacing with
other scrobblers which may have different definitions of when a scrobble
finishes or started.
** DONE Fix bug with Various Artist albums being labeled with first artist as album artist :scrobbling:bug:music:
CLOSED: [2023-03-27 Mon 20:18]
:LOGBOOK:
CLOCK: [2023-03-26 Sun 22:01]--[2023-03-27 Mon 01:07] => 3:06
:END:
** DONE Fix bug with weekly aggregator being blank on Sundays :aggregators:music:bug:
CLOSED: [2023-03-26 Sun 13:52]
** DONE Fix KoReader scrobbling to use pages rather than time of last read :scrobbling:books:improvement:
CLOSED: [2023-03-26 Sun 13:51]
:LOGBOOK:
CLOCK: [2023-03-26 Sun 13:11]--[2023-03-26 Sun 13:51] => 0:40
:END:
** DONE [#A] Add django-storage to store files on S3 :settings:improvement:
CLOSED: [2023-03-24 Fri 14:46]
:LOGBOOK:
CLOCK: [2023-03-24 Fri 10:47]--[2023-03-24 Fri 14:46] => 3:59
CLOCK: [2023-03-24 Fri 10:36]--[2023-03-24 Fri 10:40] => 0:04
:END:
** DONE Fix vrobbler settings not using booleans :settings:bug:
CLOSED: [2023-03-24 Fri 10:45]
:LOGBOOK:
CLOCK: [2023-03-24 Fri 10:40]--[2023-03-24 Fri 10:46] => 0:06
:END:
** DONE Update weekly live chart to be 7-day continuous rather than weekly :views:bug:
CLOSED: [2023-03-24 Fri 00:31]
The live view will be blank every Monday, no reason to tie it to a day of the
week. It should be "the last 7 days"
** DONE [#B] Implement a detail view for TV shows :improvement:views:
CLOSED: [2023-03-22 Wed 17:05]
** DONE [#B] Implement a detail view for Movies :improvement:views:
CLOSED: [2023-03-22 Wed 17:05]
** DONE Add "service provider" to TV Series, and use that for source when available :bug:scrobbling:
CLOSED: [2023-03-22 Wed 17:04]
** DONE Add view for long-play content (books, video games) to restart them :views:improvement:
CLOSED: [2023-03-22 Wed 17:01]
** DONE Add live chart view like Maloja :improvement:views:
CLOSED: [2023-03-07 Tue 11:13]
** DONE [#C] Figure out how to add to web-scrobbler :improvement:scrobbling:
CLOSED: [2023-03-22 Wed 17:06]
An example:
https://github.com/web-scrobbler/web-scrobbler/blob/master/src/core/background/scrobbler/maloja-scrobbler.js
This is actually going to be moot because we can import from LastFM, and
web-scrobbler integrates well with LastFM. The only thing to think through here
now is what to do with all the garbage web-scrobbler sometimes pushes to LastFM
from Youtube (all videos get pushed, sigh).
* Version 0.11.4
** DONE Add rudimentary video game scrobbling :improvement:content:videogames:
CLOSED: [2023-03-07 Tue 11:11]
** DONE Add ability to scrobble from KOReader statistics files :improvement:books:content:
CLOSED: [2023-03-07 Tue 11:11]
** DONE [#A] Fix fetching artwork without release group :bug:
* DONE [#A] Fix fetching artwork without release group :bug:
CLOSED: [2023-01-29 Sun 14:27]
When we get artwork from Musicbrianz, and it's not found, we should check for
release groups as well. This will stop issues with missing artwork because of
obscure MB release matches.
** DONE [#A] Fix Jellyfin music scrobbling N+1 past 90 completion percent :bug:
* DONE [#A] Fix Jellyfin music scrobbling N+1 past 90 completion percent :bug:
CLOSED: [2023-01-30 Mon 18:31]
:LOGBOOK:
CLOCK: [2023-01-30 Mon 18:00]--[2023-01-30 Mon 18:31] => 0:31
@ -97,7 +26,7 @@ as complete for the following conditions:
But if we keep listening beyond 90, we should basically ignore updates (or just
update the existing scrobble)
** DONE [#A] Add support for Audioscrobbler tab-separated file uploads :improvement:
* DONE [#A] Add support for Audioscrobbler tab-separated file uploads :improvement:
CLOSED: [2023-02-03 Fri 16:52]
An example of the format:
@ -120,23 +49,25 @@ An example of the format:
311 311 Misdirected Hostility 7 179 S 1740496085 61ff2c1a-fc9c-44c3-8da1-5e50a44245af
,
#+end_src
** DONE [#B] Allow scrobbling music without MB IDs by grabbing them before scrobble :improvement:
* DONE [#B] Allow scrobbling music without MB IDs by grabbing them before scrobble :improvement:
CLOSED: [2023-02-17 Fri 00:10]
This would allow a few nice flows. One, you'd be able to record the play of an
entire album by just dropping the muscibrainz_id in. This could be helpful for
offline listening. It would also mean bad metadata from mopidy would not break
scrobbling.
** DONE When updating musicbrainz IDs, clear and run fetch artwrok :improvement:
* DONE When updating musicbrainz IDs, clear and run fetch artwrok :improvement:
CLOSED: [2023-02-17 Fri 00:11]
** DONE [#A] Add ability to manually scrobble albums or tracks from MB :improvement:
CLOSED: [2023-03-07 Tue 11:09]
* TODO [#A] Add ability to manually scrobble albums or tracks from MB :improvement:
Given a UUID from musicbrainz, we should be able to scrobble an album or
individual track.
** DONE [#C] Implement keeping track of week/month/year chart-toppers :improvement:
CLOSED: [2023-03-07 Tue 11:10]
* TODO [#A] Add django-storage to store files on S3 :improvement:
* TODO [#B] Adjust cancel/finish task to use javascript to submit :improvement:
* TODO [#B] Implement a detail view for TV shows :improvement:
* TODO [#B] Implement a detail view for Moviews :improvement:
* TODO [#C] Implement keeping track of week/month/year chart-toppers :improvement:
:LOGBOOK:
CLOCK: [2023-01-30 Mon 16:30]--[2023-01-30 Mon 18:00] => 1:30
:END:
@ -152,11 +83,9 @@ a period, along with a one, two, three instance.
Of course, it could also be a data model without a table, where it runs some fun
calculations, stores it's values in Redis as a long-term lookup table and just
has to re-populate when the server restarts.
* Backlog
** TODO Add Amazon scraper to look up books when OL fails :books:improvement:
** TODO [#C] Move to using more robust mopidy-webhooks pacakge form pypi :utility:improvement:
*** Example payloads from mopidy-webhooks
**** Podcast playback ended
* TODO [#C] Move to using more robust mopidy-webhooks pacakge form pypi :improvement:
** Example payloads from mopidy-webhooks
*** Podcast playback ended
#+begin_src json
{
"type": "event",
@ -190,7 +119,7 @@ has to re-populate when the server restarts.
}
}
#+end_src
**** Podcast playback state changes
*** Podcast playback state changes
#+begin_src json
{
"type": "event",
@ -212,7 +141,7 @@ has to re-populate when the server restarts.
}
}
#+end_src
**** Podcast playback started
*** Podcast playback started
#+begin_src json
{
"type": "event",
@ -245,7 +174,7 @@ has to re-populate when the server restarts.
}
}
#+end_src
**** Podcast playback paused
*** Podcast playback paused
#+begin_src json
{
"type": "status",
@ -276,7 +205,7 @@ has to re-populate when the server restarts.
}
#+end_src
**** Track playback started
*** Track playback started
#+begin_src json
{
"type": "event",
@ -333,7 +262,7 @@ has to re-populate when the server restarts.
}
}
#+end_src
**** Track playback in progress
*** Track playback in progress
#+begin_src json
{
"type": "status",
@ -387,7 +316,7 @@ has to re-populate when the server restarts.
}
}
#+end_src
**** Track event playback paused
*** Track event playback paused
#+begin_src json
{
"type": "event",
@ -445,5 +374,9 @@ has to re-populate when the server restarts.
}
}
#+end_src
** TODO Fix bug in Jellyfin scrobbles that spam more scrobbles after completion :scrobbling:videos:bug:
** TODO Fix bug in podcast scrobbling where a second scrobble is created after completion :scrobbling:podcasts:bug:
* TODO [#C] Consider a purge command for duplicated and stuck in-progress scrobbles :improvement:
* TODO [#C] Figure out how to add to web-scrobbler :imropvement:
An example:
https://github.com/web-scrobbler/web-scrobbler/blob/master/src/core/background/scrobbler/maloja-scrobbler.js

View File

@ -1,26 +1,11 @@
# You can use this file to set environment variables for your local setup
#
VROBBLER_DEBUG=True
VROBBLER_LOG_LEVEL="DEBUG"
VROBBLER_JSON_LOGGING=True
VROBBLER_LOG_LEVEL="DEBUG"
VROBBLER_MEDIA_ROOT = "/media/"
VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS=False
VROBBLER_TMDB_API_KEY = "KEY"
VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS=True
VROBBLER_USE_S3=False
# You may also need to set these in your environment
AWS_S3_ACCESS_KEY_ID=""
AWS_S3_SECRET_ACCESS_KEY=""
AWS_S3_CUSTOM_DOMAIN="https://minio.dev/"
# API keys
VROBBLER_TMDB_API_KEY = "<key>"
VROBBLER_LASTFM_API_KEY = "<key>"
VROBBLER_LASTFM_SECRET_KEY = "<key>"
VROBBLER_THESPORTSDB_API_KEY="<key>"
VROBBLER_THEAUDIODB_API_KEY="<key>"
VROBBLER_IGDB_CLIENT_ID="<id>"
VROBBLER_IGDB_CLIENT_SECRET="<key>"
# Storages
# VROBBLER_DATABASE_URL="postgres://USER:PASSWORD@HOST:PORT/NAME"
# VROBBLER_REDIS_URL="redis://:PASS@HOST:6379/0"
VROBBLER_DATABASE_URL="postgres://USER:PASSWORD@HOST:PORT/NAME"
VROBBLER_REDIS_URL="redis://:PASS@HOST:6379/0"

View File

@ -1,11 +1,8 @@
# Local configuration for Emus
VROBBLER_DUMP_REQUEST_DATA=False
VROBBLER_LOG_TO_CONSOLE=False
VROBBLER_DEBUG=False
VROBBLER_DUMP_REQUEST_DATA=True
VROBBLER_LOG_TO_CONSOLE=True
VROBBLER_DEBUG=True
VROBBLER_LOG_LEVEL="DEBUG"
VROBBLER_MEDIA_ROOT = "/tmp/media/"
VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS=False
VROBBLER_USE_S3="False"
VROBBLER_DATABASE_URL="sqlite:///testdb.sqlite3"
VROBBLER_KEEP_DETAILED_SCROBBLE_LOGS=True

View File

@ -2,4 +2,4 @@
# Django starts so that shared_task will use this app.
from .celery import app as celery_app
__all__ = ("celery_app",)
__all__ = ('celery_app',)

View File

@ -1,30 +0,0 @@
from django.contrib import admin
from boardgames.models import BoardGame, BoardGamePublisher
from scrobbles.admin import ScrobbleInline
@admin.register(BoardGamePublisher)
class BoardGamePublisherAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"uuid",
)
ordering = ("-created",)
@admin.register(BoardGame)
class GameAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"bggeek_id",
"title",
"published_date",
)
search_fields = ("title",)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]

View File

@ -1,92 +0,0 @@
import logging
from typing import Optional
import requests
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
SEARCH_ID_URL = (
"https://boardgamegeek.com/xmlapi/search?search={query}&exact=1"
)
GAME_ID_URL = "https://boardgamegeek.com/xmlapi/boardgame/{id}"
def take_first(thing: Optional[list]) -> str:
first = ""
try:
first = thing[0]
except IndexError:
pass
if first:
try:
first = first.get_text()
except:
pass
return first
def lookup_boardgame_id_from_bgg(title: str) -> Optional[int]:
soup = None
headers = {"User-Agent": "Vrobbler 0.11.12"}
game_id = None
url = SEARCH_ID_URL.format(query=title)
r = requests.get(url, headers=headers)
if r.status_code == 200:
soup = BeautifulSoup(r.text, "xml")
if soup:
result = soup.findAll("boardgame")
if not result:
return game_id
game_id = result[0].get("objectid", None)
return game_id
def lookup_boardgame_from_bgg(lookup_id: str) -> dict:
soup = None
game_dict = {}
headers = {"User-Agent": "Vrobbler 0.11.12"}
title = ""
bgg_id = None
try:
bgg_id = int(lookup_id)
logger.debug(f"Using BGG ID {bgg_id} to find board game")
except ValueError:
title = lookup_id
logger.debug(f"Using title {title} to find board game")
if not bgg_id:
bgg_id = lookup_boardgame_id_from_bgg(title)
url = GAME_ID_URL.format(id=bgg_id)
r = requests.get(url, headers=headers)
if r.status_code == 200:
soup = BeautifulSoup(r.text, "xml")
if soup:
seconds_to_play = None
minutes = take_first(soup.findAll("playingtime"))
if minutes:
seconds_to_play = int(minutes) * 60
game_dict = {
"bggeek_id": bgg_id,
"title": take_first(soup.findAll("name", primary="true")),
"description": take_first(soup.findAll("description")),
"year_published": take_first(soup.findAll("yearpublished")),
"publisher_name": take_first(soup.findAll("boardgamepublisher")),
"cover_url": take_first(soup.findAll("image")),
"min_players": take_first(soup.findAll("minplayers")),
"max_players": take_first(soup.findAll("maxplayers")),
"recommended_age": take_first(soup.findAll("age")),
"run_time_seconds": seconds_to_play,
}
return game_dict

View File

@ -1,168 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-17 22:11
from django.db import migrations, models
import django.db.models.deletion
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
("scrobbles", "0038_alter_objectwithgenres_tag"),
]
operations = [
migrations.CreateModel(
name="BoardGamePublisher",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"created",
django_extensions.db.fields.CreationDateTimeField(
auto_now_add=True, verbose_name="created"
),
),
(
"modified",
django_extensions.db.fields.ModificationDateTimeField(
auto_now=True, verbose_name="modified"
),
),
("name", models.CharField(max_length=255)),
(
"uuid",
models.UUIDField(
blank=True,
default=uuid.uuid4,
editable=False,
null=True,
),
),
(
"logo",
models.ImageField(
blank=True,
null=True,
upload_to="games/platform-logos/",
),
),
("igdb_id", models.IntegerField(blank=True, null=True)),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
migrations.CreateModel(
name="BoardGame",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"created",
django_extensions.db.fields.CreationDateTimeField(
auto_now_add=True, verbose_name="created"
),
),
(
"modified",
django_extensions.db.fields.ModificationDateTimeField(
auto_now=True, verbose_name="modified"
),
),
(
"run_time_seconds",
models.IntegerField(blank=True, null=True),
),
(
"run_time_ticks",
models.PositiveBigIntegerField(blank=True, null=True),
),
("title", models.CharField(max_length=255)),
(
"uuid",
models.UUIDField(
blank=True,
default=uuid.uuid4,
editable=False,
null=True,
),
),
(
"cover",
models.ImageField(
blank=True, null=True, upload_to="boardgames/covers/"
),
),
(
"layout_image",
models.ImageField(
blank=True, null=True, upload_to="boardgames/layouts/"
),
),
("summary", models.TextField(blank=True, null=True)),
("rating", models.FloatField(blank=True, null=True)),
(
"max_players",
models.PositiveSmallIntegerField(blank=True, null=True),
),
(
"min_players",
models.PositiveSmallIntegerField(blank=True, null=True),
),
("published_date", models.DateField(blank=True, null=True)),
(
"recommened_age",
models.PositiveSmallIntegerField(blank=True, null=True),
),
(
"seconds_to_play",
models.IntegerField(blank=True, null=True),
),
(
"bggeek_id",
models.CharField(blank=True, max_length=255, null=True),
),
(
"genre",
taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
(
"publisher",
models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
to="boardgames.boardgamepublisher",
),
),
],
options={
"abstract": False,
},
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-17 22:16
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="boardgame",
name="description",
field=models.TextField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-17 22:17
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0002_boardgame_description"),
]
operations = [
migrations.RenameField(
model_name="boardgame",
old_name="recommened_age",
new_name="recommended_age",
),
]

View File

@ -1,17 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-17 22:25
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0003_rename_recommened_age_boardgame_recommended_age"),
]
operations = [
migrations.RemoveField(
model_name="boardgame",
name="seconds_to_play",
),
]

View File

@ -1,17 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-17 22:29
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0004_remove_boardgame_seconds_to_play"),
]
operations = [
migrations.RemoveField(
model_name="boardgame",
name="summary",
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-18 02:33
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0040_alter_scrobble_media_type"),
("boardgames", "0005_remove_boardgame_summary"),
]
operations = [
migrations.AlterField(
model_name="boardgame",
name="genre",
field=taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,166 +0,0 @@
import logging
from datetime import datetime
from typing import Optional
from uuid import uuid4
import requests
from boardgames.bgg import lookup_boardgame_from_bgg
from django.conf import settings
from django.core.files.base import ContentFile
from django.db import models
from django.urls import reverse
from django_extensions.db.models import TimeStampedModel
from imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFit
from scrobbles.mixins import ScrobblableMixin
from vrobbler.apps.boardgames.bgg import lookup_boardgame_id_from_bgg
logger = logging.getLogger(__name__)
BNULL = {"blank": True, "null": True}
class BoardGamePublisher(TimeStampedModel):
name = models.CharField(max_length=255)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
logo = models.ImageField(upload_to="games/platform-logos/", **BNULL)
igdb_id = models.IntegerField(**BNULL)
def __str__(self):
return self.name
def get_absolute_url(self):
return reverse(
"boardgames:publisher_detail", kwargs={"slug": self.uuid}
)
class BoardGame(ScrobblableMixin):
COMPLETION_PERCENT = getattr(
settings, "BOARD_GAME_COMPLETION_PERCENT", 100
)
FIELDS_FROM_BGGEEK = [
"igdb_id",
"alternative_name",
"rating",
"rating_count",
"release_date",
"cover",
"screenshot",
]
title = models.CharField(max_length=255)
publisher = models.ForeignKey(
BoardGamePublisher, **BNULL, on_delete=models.DO_NOTHING
)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
description = models.TextField(**BNULL)
cover = models.ImageField(upload_to="boardgames/covers/", **BNULL)
cover_small = ImageSpecField(
source="cover",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
cover_medium = ImageSpecField(
source="cover",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
layout_image = models.ImageField(upload_to="boardgames/layouts/", **BNULL)
layout_image_small = ImageSpecField(
source="layout_image",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
layout_image_medium = ImageSpecField(
source="layout_image",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
rating = models.FloatField(**BNULL)
max_players = models.PositiveSmallIntegerField(**BNULL)
min_players = models.PositiveSmallIntegerField(**BNULL)
published_date = models.DateField(**BNULL)
recommended_age = models.PositiveSmallIntegerField(**BNULL)
bggeek_id = models.CharField(max_length=255, **BNULL)
def __str__(self):
return self.title
def get_absolute_url(self):
return reverse(
"boardgames:boardgame_detail", kwargs={"slug": self.uuid}
)
def primary_image_url(self) -> str:
url = ""
if self.cover:
url = self.cover.url
return url
def bggeek_link(self):
link = ""
if self.bggeek_id:
link = f"https://boardgamegeek.com/boardgame/{self.bggeek_id}"
return link
def get_start_url(self):
return reverse("scrobbles:start", kwargs={"uuid": self.uuid})
def fix_metadata(self, data: dict = {}, force_update=False) -> None:
if not self.published_date or force_update:
if not data:
data = lookup_boardgame_from_bgg(str(self.bggeek_id))
cover_url = data.pop("cover_url")
year = data.pop("year_published")
publisher_name = data.pop("publisher_name")
if year:
data["published_date"] = datetime(int(year), 1, 1)
# Fun trick for updating all fields at once
BoardGame.objects.filter(pk=self.id).update(**data)
self.refresh_from_db()
# Add publishers
(
self.publisher,
_created,
) = BoardGamePublisher.objects.get_or_create(name=publisher_name)
self.save()
# Go get cover image if the URL is present
if cover_url and not self.cover:
headers = {"User-Agent": "Vrobbler 0.11.12"}
r = requests.get(cover_url, headers=headers)
logger.debug(r.status_code)
if r.status_code == 200:
fname = f"{self.title}_cover_{self.uuid}.jpg"
self.cover.save(fname, ContentFile(r.content), save=True)
logger.debug("Loaded cover image from BGGeek")
@classmethod
def find_or_create(
cls, lookup_id: str, data: Optional[dict] = {}
) -> Optional["BoardGame"]:
"""Given a Lookup ID (either BGG or BGA ID), return a board game object"""
boardgame = None
if not data:
data = lookup_boardgame_from_bgg(lookup_id)
if data:
boardgame, created = cls.objects.get_or_create(
title=data["title"], bggeek_id=lookup_id
)
if created:
boardgame.fix_metadata(data=data)
return boardgame

View File

@ -1,21 +0,0 @@
from django.urls import path
from boardgames import views
app_name = "boardgames"
urlpatterns = [
path(
"board-game/", views.BoardGameListView.as_view(), name="boardgame_list"
),
path(
"board-game/<slug:slug>/",
views.BoardGameDetailView.as_view(),
name="boardgame_detail",
),
path(
"board-game-publisher/<slug:slug>/",
views.BoardGamePublisherDetailView.as_view(),
name="publisher_detail",
),
]

View File

@ -1,17 +0,0 @@
from django.views import generic
from boardgames.models import BoardGame, BoardGamePublisher
class BoardGameListView(generic.ListView):
model = BoardGame
paginate_by = 20
class BoardGameDetailView(generic.DetailView):
model = BoardGame
slug_field = "uuid"
class BoardGamePublisherDetailView(generic.DetailView):
model = BoardGamePublisher
slug_field = "uuid"

View File

@ -1,32 +1,19 @@
from django.contrib import admin
from books.models import Author, Book, Page
from books.models import Author, Book
from scrobbles.admin import ScrobbleInline
@admin.register(Author)
class AuthorAdmin(admin.ModelAdmin):
class AlbumAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"openlibrary_id",
"bio",
"wikipedia_url",
)
ordering = ("-created",)
search_fields = ("name",)
@admin.register(Page)
class PageAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_filter = ("book",)
ordering = ("book", "number")
list_display = ("name", "openlibrary_id")
ordering = ("name",)
@admin.register(Book)
class BookAdmin(admin.ModelAdmin):
class ArtistAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"title",
@ -35,8 +22,4 @@ class BookAdmin(admin.ModelAdmin):
"pages",
"openlibrary_id",
)
search_fields = ("name",)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]
ordering = ("title",)

View File

@ -1,144 +0,0 @@
from enum import Enum
from typing import Optional
from bs4 import BeautifulSoup
import requests
import logging
logger = logging.getLogger(__name__)
USER_AGENT = (
"Mozilla/5.0 (Android 4.4; Mobile; rv:41.0) Gecko/41.0 Firefox/41.0"
)
AMAZON_SEARCH_URL = "https://www.amazon.com/s?k={amazon_id}"
class AmazonAttribute(Enum):
SERIES = 0
PAGES = 1
LANGUAGE = 2
PUBLISHER = 3
PUB_DATE = 4
DIMENSIONS = 5
ISBN_10 = 6
ISBN_13 = 7
def strip_and_clean(text):
return text.strip("\n").rstrip().lstrip()
def get_rating_from_soup(soup) -> Optional[int]:
rating = None
try:
potential_rating = soup.find("div", class_="allmusic-rating")
if potential_rating:
rating = int(strip_and_clean(potential_rating.get_text()))
except ValueError:
pass
return rating
def get_review_from_soup(soup) -> str:
review = ""
try:
potential_text = soup.find("div", class_="text")
if potential_text:
review = strip_and_clean(potential_text.get_text())
except ValueError:
pass
return review
def scrape_data_from_amazon(url) -> dict:
data_dict = {}
headers = {"User-Agent": USER_AGENT}
r = requests.get(url, headers=headers)
if r.status_code == 200:
soup = BeautifulSoup(r.text, "html.parser")
import pdb
pdb.set_trace()
data_dict["rating"] = get_rating_from_soup(soup)
data_dict["review"] = get_review_from_soup(soup)
return data_dict
def get_amazon_product_dict(amazon_id: str) -> dict:
data_dict = {}
url = ""
search_url = AMAZON_SEARCH_URL.format(amazon_id=amazon_id)
headers = {
"user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
"accept-language": "en-GB,en;q=0.9",
}
response = requests.get(search_url, headers=headers)
if response.status_code != 200:
logger.info(f"Bad http response from Amazon {response}")
return data_dict
soup = BeautifulSoup(response.text, "html.parser")
results = soup.find("a", class_="a-link-normal")
if not results:
logger.info(f"No search results for {amazon_id}")
return data_dict
product_url = "https://www.amazon.com" + str(results.get("href", ""))
data_dict = {}
response = requests.get(product_url, headers=headers)
if response.status_code != 200:
logger.info(f"Bad http response from Amazon {response}")
return data_dict
soup = BeautifulSoup(response.text, "html.parser")
try:
data_dict["title"] = soup.findAll("span", class_="celwidget")[
1
].text.strip()
data_dict["cover_url"] = soup.find("img", class_="frontImage").get(
"src"
)
data_dict["summary"] = soup.findAll(
"div", class_="a-expander-content"
)[1].text
meta = soup.findAll("div", class_="rpi-attribute-value")
data_dict["isbn"] = meta[AmazonAttribute.ISBN_10.value].text.strip()
pages = meta[AmazonAttribute.PAGES.value].text
if "pages" in pages:
data_dict["pages"] = (
meta[AmazonAttribute.PAGES.value]
.text.split("pages")[0]
.strip()
)
except IndexError as e:
logger.error(
f"Amazon lookup is failing for this product {amazon_id}: {e}"
)
except AttributeError as e:
logger.error(
f"Amazon lookup is failing for this product {amazon_id}: {e}"
)
return data_dict
def lookup_book_from_amazon(amazon_id: str) -> dict:
top = {}
return {
"title": top.get("title"),
"isbn": isbn,
"openlibrary_id": ol_id,
"goodreads_id": get_first("id_goodreads", top),
"first_publish_year": top.get("first_publish_year"),
"first_sentence": first_sentence,
"pages": top.get("number_of_pages_median", None),
"cover_url": COVER_URL.format(id=ol_id),
"ol_author_id": ol_author_id,
"subject_key_list": top.get("subject_key", []),
}

View File

@ -8,12 +8,12 @@ from books.models import Author, Book
class AuthorViewSet(viewsets.ModelViewSet):
queryset = Author.objects.all().order_by("-created")
queryset = Author.objects.all().order_by('-created')
serializer_class = AuthorSerializer
permission_classes = [permissions.IsAuthenticated]
class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all().order_by("-created")
queryset = Book.objects.all().order_by('-created')
serializer_class = BookSerializer
permission_classes = [permissions.IsAuthenticated]

View File

@ -1,262 +0,0 @@
import codecs
import logging
import os
import re
import sqlite3
from datetime import datetime
from enum import Enum
from typing import Iterable, List
import pytz
import requests
from books.models import Author, Book, Page
from books.openlibrary import get_author_openlibrary_id
from django.db.models import Sum
from pylast import httpx, tempfile
from scrobbles.models import Scrobble
from stream_sqlite import stream_sqlite
logger = logging.getLogger(__name__)
class KoReaderBookColumn(Enum):
ID = 0
TITLE = 1
AUTHORS = 2
NOTES = 3
LAST_OPEN = 4
HIGHLIGHTS = 5
PAGES = 6
SERIES = 7
LANGUAGE = 8
MD5 = 9
TOTAL_READ_TIME = 10
TOTAL_READ_PAGES = 11
class KoReaderPageStatColumn(Enum):
ID_BOOK = 0
PAGE = 1
START_TIME = 2
DURATION = 3
TOTAL_PAGES = 4
def _sqlite_bytes(sqlite_url):
with httpx.stream("GET", sqlite_url) as r:
yield from r.iter_bytes(chunk_size=65_536)
def get_book_map_from_sqlite(rows: Iterable) -> dict:
"""Given an interable of sqlite rows from the books table, lookup existing
books, create ones that don't exist, and return a mapping of koreader IDs to
primary key IDs for page creation.
"""
book_id_map = {}
for book_row in rows:
book = Book.objects.filter(
koreader_md5=book_row[KoReaderBookColumn.MD5.value]
).first()
if not book:
book, created = Book.objects.get_or_create(
title=book_row[KoReaderBookColumn.TITLE.value]
)
if created:
total_pages = book_row[KoReaderBookColumn.PAGES.value]
run_time = total_pages * book.AVG_PAGE_READING_SECONDS
ko_authors = book_row[
KoReaderBookColumn.AUTHORS.value
].replace("\n", ", ")
# Strip middle initials, OpenLibrary often fails with these
ko_authors = re.sub(" [A-Z]. ", " ", ko_authors)
book_dict = {
"title": book_row[KoReaderBookColumn.TITLE.value],
"pages": total_pages,
"koreader_md5": book_row[KoReaderBookColumn.MD5.value],
"koreader_id": int(book_row[KoReaderBookColumn.ID.value]),
"koreader_authors": ko_authors,
"run_time_seconds": run_time,
}
Book.objects.filter(pk=book.id).update(**book_dict)
# Add authors
authors = ko_authors.split(", ")
author_list = []
for author_str in authors:
logger.debug(f"Looking up author {author_str}")
if author_str == "N/A":
continue
author = Author.objects.filter(name=author_str).first()
if not author:
author = Author.objects.create(
name=author_str,
openlibrary_id=get_author_openlibrary_id(
author_str
),
)
author.fix_metadata()
logger.debug(f"Created author {author}")
book.authors.add(author)
# This will try to fix metadata by looking it up on OL
book.fix_metadata()
book.refresh_from_db()
total_seconds = 0
if book_row[KoReaderBookColumn.TOTAL_READ_TIME.value]:
total_seconds = book_row[KoReaderBookColumn.TOTAL_READ_TIME.value]
book_id_map[book_row[KoReaderBookColumn.ID.value]] = (
book.id,
total_seconds,
)
return book_id_map
def build_scrobbles_from_pages(
rows: Iterable, book_id_map: dict, user_id: int
) -> List[Scrobble]:
new_scrobbles = []
new_scrobbles = []
pages_found = []
book_read_time_map = {}
for page_row in rows:
koreader_id = page_row[KoReaderPageStatColumn.ID_BOOK.value]
if koreader_id not in book_id_map.keys():
continue
page_number = page_row[KoReaderPageStatColumn.PAGE.value]
ts = page_row[KoReaderPageStatColumn.START_TIME.value]
book_id = book_id_map[koreader_id][0]
book_read_time_map[book_id] = book_id_map[koreader_id][1]
page, page_created = Page.objects.get_or_create(
book_id=book_id, number=page_number, user_id=user_id
)
if page_created:
page.start_time = datetime.utcfromtimestamp(ts).replace(
tzinfo=pytz.utc
)
page.duration_seconds = page_row[
KoReaderPageStatColumn.DURATION.value
]
page.save(update_fields=["start_time", "duration_seconds"])
pages_found.append(page)
playback_position_seconds = 0
for page in set(pages_found):
# Add up page seconds to set the aggregate time of all pages to reading time
playback_position_seconds = (
playback_position_seconds + page.duration_seconds
)
if page.is_scrobblable:
# Check to see if a scrobble with this timestamp, book and user already exists
scrobble = Scrobble.objects.filter(
timestamp=page.start_time,
book_id=page.book_id,
user_id=user_id,
).first()
if not scrobble:
logger.debug(
f"Queueing scrobble for {page.book}, page {page.number}"
)
new_scrobble = Scrobble(
book_id=page.book_id,
user_id=user_id,
source="KOReader",
media_type=Scrobble.MediaType.BOOK,
timestamp=page.start_time,
played_to_completion=True,
playback_position_seconds=playback_position_seconds,
in_progress=False,
book_pages_read=page.number,
long_play_complete=False,
)
new_scrobbles.append(new_scrobble)
# After setting a scrobblable page, reset our accumulator
playback_position_seconds = 0
return new_scrobbles
def enrich_koreader_scrobbles(scrobbles: list) -> None:
"""Given a list of scrobbles, update pages read, long play seconds and check
for media completion"""
for scrobble in scrobbles:
scrobble.book_pages_read = scrobble.book.page_set.last().number
# But if there's a next scrobble, set pages read to their starting page
#
if scrobble.next:
scrobble.book_pages_read = scrobble.next.book_pages_read - 1
scrobble.long_play_seconds = scrobble.book.page_set.filter(
number__lte=scrobble.book_pages_read
).aggregate(Sum("duration_seconds"))["duration_seconds__sum"]
scrobble.save(update_fields=["book_pages_read", "long_play_seconds"])
def process_koreader_sqlite_url(file_url, user_id) -> list:
book_id_map = {}
new_scrobbles = []
for table_name, pragma_table_info, rows in stream_sqlite(
_sqlite_bytes(file_url), max_buffer_size=1_048_576
):
logger.debug(f"Found table {table_name} - processing")
if table_name == "book":
book_id_map = get_book_map_from_sqlite(rows)
if table_name == "page_stat_data":
new_scrobbles = build_scrobbles_from_pages(
rows, book_id_map, user_id
)
logger.debug(f"Creating {len(new_scrobbles)} new scrobbles")
created = []
if new_scrobbles:
created = Scrobble.objects.bulk_create(new_scrobbles)
enrich_koreader_scrobbles(created)
logger.info(
f"Created {len(created)} scrobbles",
extra={"created_scrobbles": created},
)
return created
def process_koreader_sqlite_file(file_path, user_id) -> list:
"""Given a sqlite file from KoReader, open the book table, iterate
over rows creating scrobbles from each book found"""
# Create a SQL connection to our SQLite database
con = sqlite3.connect(file_path)
cur = con.cursor()
book_id_map = get_book_map_from_sqlite(cur.execute("SELECT * FROM book"))
new_scrobbles = build_scrobbles_from_pages(
cur.execute("SELECT * from page_stat_data"), book_id_map, user_id
)
created = []
if new_scrobbles:
created = Scrobble.objects.bulk_create(new_scrobbles)
enrich_koreader_scrobbles(created)
logger.info(
f"Created {len(created)} scrobbles",
extra={"created_scrobbles": created},
)
return created
def process_koreader_sqlite(file_path: str, user_id: int) -> list:
is_os_file = "https://" not in file_path
if is_os_file:
created = process_koreader_sqlite_file(file_path, user_id)
else:
created = process_koreader_sqlite_url(file_path, user_id)
return created

View File

@ -1,124 +0,0 @@
#!/usr/bin/env python3
from enum import Enum
from typing import Optional
from bs4 import BeautifulSoup
import requests
import logging
logger = logging.getLogger(__name__)
HEADERS = {
"user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
"accept-language": "en-GB,en;q=0.9",
}
LOCG_WRTIER_URL = ""
LOCG_WRITER_DETAIL_URL = "https://leagueofcomicgeeks.com/people/{slug}"
LOCG_SEARCH_URL = (
"https://leagueofcomicgeeks.com/search/ajax_issues?query={query}"
)
LOCG_DETAIL_URL = "https://leagueofcomicgeeks.com/comic/{locg_slug}"
def strip_and_clean(text):
return text.strip("\n").strip()
def get_rating_from_soup(soup) -> Optional[int]:
rating = None
try:
potential_rating = soup.find("div", class_="allmusic-rating")
if potential_rating:
rating = int(strip_and_clean(potential_rating.get_text()))
except ValueError:
pass
return rating
def lookup_comic_writer_by_locg_slug(slug: str) -> dict:
data_dict = {}
writer_url = LOCG_WRITER_DETAIL_URL.format(slug=slug)
response = requests.get(writer_url, headers=HEADERS)
if response.status_code != 200:
logger.info(f"Bad http response from LOCG {response}")
return data_dict
soup = BeautifulSoup(response.text, "html.parser")
data_dict["locg_slug"] = slug
data_dict["name"] = soup.find("h1").text.strip()
data_dict["photo_url"] = soup.find("div", class_="avatar").img.get("src")
return data_dict
def lookup_comic_by_locg_slug(slug: str) -> dict:
data_dict = {}
product_url = LOCG_DETAIL_URL.format(locg_slug=slug)
response = requests.get(product_url, headers=HEADERS)
if response.status_code != 200:
logger.info(f"Bad http response from LOCG {response}")
return data_dict
soup = BeautifulSoup(response.text, "html.parser")
try:
data_dict["title"] = soup.find("h1").text.strip()
data_dict["summary"] = soup.find("p").text.strip()
data_dict["cover_url"] = (
soup.find("div", class_="cover-art").find("img").get("src")
)
attrs = soup.findAll("div", class_="details-addtl-block")
try:
data_dict["pages"] = (
attrs[1]
.find("div", class_="value")
.text.split("pages")[0]
.strip()
)
except IndexError:
logger.warn(f"No ISBN field")
try:
data_dict["isbn"] = (
attrs[3].find("div", class_="value").text.strip()
)
except IndexError:
logger.warn(f"No ISBN field")
writer_slug = None
try:
writer_slug = (
soup.findAll("div", class_="name")[5]
.a.get("href")
.split("people/")[1]
)
except IndexError:
logger.warn(f"No wrtier found")
if writer_slug:
data_dict["locg_writer_slug"] = writer_slug
except AttributeError:
logger.warn(f"Trouble parsing HTML, elements missing")
return data_dict
def lookup_comic_from_locg(title: str) -> dict:
search_url = LOCG_SEARCH_URL.format(query=title)
response = requests.get(search_url, headers=HEADERS)
if response.status_code != 200:
logger.warn(f"Bad http response from LOCG {response}")
return {}
soup = BeautifulSoup(response.text, "html.parser")
try:
slug = soup.findAll("a")[1].get("href").split("comic/")[1]
except IndexError:
logger.warn(f"No comic found on LOCG for {title}")
return {}
return lookup_comic_by_locg_slug(slug)

View File

@ -1,23 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 05:29
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="book",
name="author_name",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="book",
name="author_openlibrary_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,20 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 05:31
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0002_book_author_name_book_author_openlibrary_id"),
]
operations = [
migrations.AddField(
model_name="book",
name="cover",
field=models.ImageField(
blank=True, null=True, upload_to="books/covers/"
),
),
]

View File

@ -1,69 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 16:31
from django.db import migrations, models
import django.db.models.deletion
import django_extensions.db.fields
class Migration(migrations.Migration):
dependencies = [
("books", "0003_book_cover"),
]
operations = [
migrations.RemoveField(
model_name="book",
name="author_name",
),
migrations.RemoveField(
model_name="book",
name="author_openlibrary_id",
),
migrations.AddField(
model_name="author",
name="headshot",
field=models.ImageField(
blank=True, null=True, upload_to="books/authors/"
),
),
migrations.CreateModel(
name="Page",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"created",
django_extensions.db.fields.CreationDateTimeField(
auto_now_add=True, verbose_name="created"
),
),
(
"modified",
django_extensions.db.fields.ModificationDateTimeField(
auto_now=True, verbose_name="modified"
),
),
("number", models.IntegerField()),
("start_time", models.DateTimeField()),
("duration_seconds", models.IntegerField()),
(
"book",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to="books.book",
),
),
],
options={
"unique_together": {("book", "number")},
},
),
]

View File

@ -1,21 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 16:34
from django.db import migrations, models
import uuid
class Migration(migrations.Migration):
dependencies = [
("books", "0004_remove_book_author_name_and_more"),
]
operations = [
migrations.AddField(
model_name="author",
name="uuid",
field=models.UUIDField(
blank=True, default=uuid.uuid4, editable=False, null=True
),
),
]

View File

@ -1,23 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 17:13
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0005_author_uuid"),
]
operations = [
migrations.AlterField(
model_name="page",
name="duration_seconds",
field=models.IntegerField(blank=True, null=True),
),
migrations.AlterField(
model_name="page",
name="start_time",
field=models.DateTimeField(blank=True, null=True),
),
]

View File

@ -1,48 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 17:44
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0006_alter_page_duration_seconds_alter_page_start_time"),
]
operations = [
migrations.AddField(
model_name="author",
name="amazon_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="author",
name="bio",
field=models.TextField(blank=True, null=True),
),
migrations.AddField(
model_name="author",
name="goodreads_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="author",
name="isni",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="author",
name="librarything_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="author",
name="wikidata_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="author",
name="wikipedia_url",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,21 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-06 18:42
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
(
"books",
"0007_author_amazon_id_author_bio_author_goodreads_id_and_more",
),
]
operations = [
migrations.AddField(
model_name="book",
name="first_sentence",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:01
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0008_book_first_sentence"),
]
operations = [
migrations.AlterField(
model_name="book",
name="run_time",
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:21
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("books", "0009_alter_book_run_time"),
]
operations = [
migrations.RenameField(
model_name="book",
old_name="run_time",
new_name="run_time_seconds",
),
]

View File

@ -1,25 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-14 22:27
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0033_genre_objectwithgenres"),
("books", "0010_rename_run_time_book_run_time_seconds"),
]
operations = [
migrations.AddField(
model_name="book",
name="genre",
field=taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-03-26 02:34
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0011_book_genre"),
]
operations = [
migrations.AddField(
model_name="page",
name="end_time",
field=models.DateTimeField(blank=True, null=True),
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.1.7 on 2023-03-26 05:31
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("books", "0012_page_end_time"),
]
operations = [
migrations.AddField(
model_name="page",
name="user",
field=models.ForeignKey(
default=1,
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
),
preserve_default=False,
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-18 02:33
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0040_alter_scrobble_media_type"),
("books", "0013_page_user"),
]
operations = [
migrations.AlterField(
model_name="book",
name="genre",
field=taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,30 +0,0 @@
# Generated by Django 4.1.7 on 2023-08-19 02:47
from django.db import migrations, models
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0043_scrobbledpage"),
("books", "0014_alter_book_genre"),
]
operations = [
migrations.AlterField(
model_name="book",
name="first_sentence",
field=models.TextField(blank=True, null=True),
),
migrations.AlterField(
model_name="book",
name="genre",
field=taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,23 +0,0 @@
# Generated by Django 4.1.7 on 2023-08-26 04:48
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0015_alter_book_first_sentence_alter_book_genre"),
]
operations = [
migrations.AddField(
model_name="book",
name="locg_slug",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="book",
name="summary",
field=models.TextField(blank=True, null=True),
),
]

View File

@ -1,20 +0,0 @@
# Generated by Django 4.1.7 on 2023-08-26 16:39
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0016_book_locg_slug_book_summary"),
]
operations = [
migrations.AlterField(
model_name="book",
name="authors",
field=models.ManyToManyField(
blank=True, null=True, to="books.author"
),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-08-31 03:57
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0017_alter_book_authors"),
]
operations = [
migrations.AddField(
model_name="author",
name="locg_slug",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-21 23:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0018_author_locg_slug"),
]
operations = [
migrations.AlterField(
model_name="book",
name="authors",
field=models.ManyToManyField(blank=True, to="books.author"),
),
]

View File

@ -1,33 +1,15 @@
import logging
from datetime import timedelta
from uuid import uuid4
from typing import Dict
import requests
from books.openlibrary import (
lookup_author_from_openlibrary,
lookup_book_from_openlibrary,
)
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.files.base import ContentFile
from django.db import models
from django.urls import reverse
from django_extensions.db.models import TimeStampedModel
from imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFit
from scrobbles.mixins import (
LongPlayScrobblableMixin,
ObjectWithGenres,
ScrobblableMixin,
)
from scrobbles.utils import get_scrobbles_for_media
from taggit.managers import TaggableManager
from scrobbles.mixins import ScrobblableMixin
from vrobbler.apps.books.locg import (
lookup_comic_by_locg_slug,
lookup_comic_from_locg,
lookup_comic_writer_by_locg_slug,
)
from books.utils import lookup_book_from_openlibrary
from scrobbles.utils import get_scrobbles_for_media
logger = logging.getLogger(__name__)
User = get_user_model()
@ -36,60 +18,21 @@ BNULL = {"blank": True, "null": True}
class Author(TimeStampedModel):
name = models.CharField(max_length=255)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
openlibrary_id = models.CharField(max_length=255, **BNULL)
headshot = models.ImageField(upload_to="books/authors/", **BNULL)
headshot_small = ImageSpecField(
source="headshot",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
headshot_medium = ImageSpecField(
source="headshot",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
bio = models.TextField(**BNULL)
wikipedia_url = models.CharField(max_length=255, **BNULL)
isni = models.CharField(max_length=255, **BNULL)
locg_slug = models.CharField(max_length=255, **BNULL)
wikidata_id = models.CharField(max_length=255, **BNULL)
goodreads_id = models.CharField(max_length=255, **BNULL)
librarything_id = models.CharField(max_length=255, **BNULL)
amazon_id = models.CharField(max_length=255, **BNULL)
def __str__(self):
return f"{self.name}"
def fix_metadata(self, data_dict: dict = {}):
if not data_dict and self.openlibrary_id:
data_dict = lookup_author_from_openlibrary(self.openlibrary_id)
if not data_dict or not data_dict.get("name"):
return
headshot_url = data_dict.pop("author_headshot_url", "")
Author.objects.filter(pk=self.id).update(**data_dict)
self.refresh_from_db()
if headshot_url:
r = requests.get(headshot_url)
if r.status_code == 200:
fname = f"{self.name}_{self.uuid}.jpg"
self.headshot.save(fname, ContentFile(r.content), save=True)
def fix_metadata(self):
logger.warn("Not implemented yet")
class Book(LongPlayScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, "BOOK_COMPLETION_PERCENT", 95)
AVG_PAGE_READING_SECONDS = getattr(
settings, "AVERAGE_PAGE_READING_SECONDS", 60
)
class Book(ScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, 'BOOK_COMPLETION_PERCENT', 95)
title = models.CharField(max_length=255)
authors = models.ManyToManyField(Author, blank=True)
authors = models.ManyToManyField(Author)
openlibrary_id = models.CharField(max_length=255, **BNULL)
goodreads_id = models.CharField(max_length=255, **BNULL)
koreader_id = models.IntegerField(**BNULL)
koreader_authors = models.CharField(max_length=255, **BNULL)
@ -98,154 +41,26 @@ class Book(LongPlayScrobblableMixin):
pages = models.IntegerField(**BNULL)
language = models.CharField(max_length=4, **BNULL)
first_publish_year = models.IntegerField(**BNULL)
first_sentence = models.TextField(**BNULL)
openlibrary_id = models.CharField(max_length=255, **BNULL)
locg_slug = models.CharField(max_length=255, **BNULL)
cover = models.ImageField(upload_to="books/covers/", **BNULL)
cover_small = ImageSpecField(
source="cover",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
cover_medium = ImageSpecField(
source="cover",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
summary = models.TextField(**BNULL)
genre = TaggableManager(through=ObjectWithGenres)
def __str__(self):
return f"{self.title}"
@property
def subtitle(self):
return f" by {self.author}"
@property
def primary_image_url(self) -> str:
url = ""
if self.cover:
url = self.cover_medium.url
return url
def get_start_url(self):
return reverse("scrobbles:start", kwargs={"uuid": self.uuid})
def get_absolute_url(self):
return reverse("books:book_detail", kwargs={"slug": self.uuid})
def fix_metadata(self, data: dict = {}, force_update=False):
if (not self.openlibrary_id or not self.locg_slug) or force_update:
author_name = ""
if self.author:
author_name = self.author.name
if not data:
if self.locg_slug:
data = lookup_comic_by_locg_slug(str(self.locg_slug))
else:
data = lookup_comic_from_locg(str(self.title))
if not data:
logger.warn(
f"Book not found on LOCG, checking OL {self.title}"
)
if self.openlibrary_id and force_update:
data = lookup_book_from_openlibrary(
str(self.openlibrary_id)
)
else:
data = lookup_book_from_openlibrary(
str(self.title), author_name
)
if not data:
logger.warn(f"Book not found in OL {self.title}")
return
# We can discard the author name from OL for now, we'll lookup details below
data.pop("ol_author_name", "")
if data.get("ol_author_id"):
self.fix_authors_metadata(data.pop("ol_author_id", ""))
if data.get("locg_writer_slug"):
self.get_author_from_locg(data.pop("locg_writer_slug", ""))
ol_title = data.get("title", "")
# Kick out a little warning if we're about to change KoReader's title
if ol_title.lower() != str(self.title).lower():
logger.warn(
f"OL and KoReader disagree on this book title {self.title} != {ol_title}"
)
# If we don't know pages, don't overwrite existing with None
if data.get("pages") == None:
data.pop("pages")
# Pop this, so we can look it up later
cover_url = data.pop("cover_url", "")
subject_key_list = data.pop("subject_key_list", "")
# Fun trick for updating all fields at once
Book.objects.filter(pk=self.id).update(**data)
self.refresh_from_db()
if subject_key_list:
self.genre.add(*subject_key_list)
if cover_url:
r = requests.get(cover_url)
if r.status_code == 200:
fname = f"{self.title}_{self.uuid}.jpg"
self.cover.save(fname, ContentFile(r.content), save=True)
if self.pages:
self.run_time_seconds = int(self.pages) * int(
self.AVG_PAGE_READING_SECONDS
)
return f"{self.title} by {self.author}"
def fix_metadata(self):
if not self.openlibrary_id:
book_meta = lookup_book_from_openlibrary(self.title, self.author)
self.openlibrary_id = book_meta.get("openlibrary_id")
self.isbn = book_meta.get("isbn")
self.goodreads_id = book_meta.get("goodreads_id")
self.first_pubilsh_year = book_meta.get("first_publish_year")
self.save()
def fix_authors_metadata(self, openlibrary_author_id):
author = Author.objects.filter(
openlibrary_id=openlibrary_author_id
).first()
if not author:
data = lookup_author_from_openlibrary(openlibrary_author_id)
author_image_url = data.pop("author_headshot_url", None)
author = Author.objects.create(**data)
if author_image_url:
r = requests.get(author_image_url)
if r.status_code == 200:
fname = f"{author.name}_{author.uuid}.jpg"
author.headshot.save(
fname, ContentFile(r.content), save=True
)
self.authors.add(author)
def get_author_from_locg(self, locg_slug):
writer = lookup_comic_writer_by_locg_slug(locg_slug)
author, created = Author.objects.get_or_create(
name=writer["name"], locg_slug=writer["locg_slug"]
)
if (created or not author.headshot) and writer["photo_url"]:
r = requests.get(writer["photo_url"])
if r.status_code == 200:
fname = f"{author.name}_{author.uuid}.jpg"
author.headshot.save(fname, ContentFile(r.content), save=True)
self.authors.add(author)
@property
def author(self):
return self.authors.first()
def get_absolute_url(self):
return reverse("books:book_detail", kwargs={'slug': self.uuid})
@property
def pages_for_completion(self) -> int:
if not self.pages:
@ -253,125 +68,6 @@ class Book(LongPlayScrobblableMixin):
return 0
return int(self.pages * (self.COMPLETION_PERCENT / 100))
def update_long_play_seconds(self):
"""Check page timestamps and duration and update"""
if self.page_set.all():
...
def progress_for_user(self, user_id: int) -> int:
"""Used to keep track of whether the book is complete or not"""
user = User.objects.get(id=user_id)
def progress_for_user(self, user: User) -> int:
last_scrobble = get_scrobbles_for_media(self, user).last()
progress = 0
if last_scrobble:
progress = int((last_scrobble.book_pages_read / self.pages) * 100)
return progress
@classmethod
def find_or_create(cls, lookup_id: str, author: str = "") -> "Book":
data = lookup_book_from_openlibrary(lookup_id, author)
book, book_created = cls.objects.get_or_create(isbn=data["isbn"])
if book_created:
book.fix_metadata(data=data)
return book
def save(self, *args, **kwargs):
if (
(not self.isbn and not self.cover)
and (self.locg_slug or self.openlibrary_id)
and self.id
):
self.fix_metadata(force_update=True)
return super(Book, self).save(*args, **kwargs)
class Page(TimeStampedModel):
user = models.ForeignKey(User, on_delete=models.CASCADE)
book = models.ForeignKey(Book, on_delete=models.CASCADE)
number = models.IntegerField()
start_time = models.DateTimeField(**BNULL)
end_time = models.DateTimeField(**BNULL)
duration_seconds = models.IntegerField(**BNULL)
class Meta:
unique_together = (
"book",
"number",
)
def __str__(self):
return f"Page {self.number} of {self.book.pages} in {self.book.title}"
def save(self, *args, **kwargs):
if not self.end_time and self.duration_seconds:
self._set_end_time()
return super(Page, self).save(*args, **kwargs)
@property
def next(self):
page = self.book.page_set.filter(number=self.number + 1).first()
if not page:
page = (
self.book.page_set.filter(created__gt=self.created)
.order_by("created")
.first()
)
return page
@property
def previous(self):
page = self.book.page_set.filter(number=self.number - 1).first()
if not page:
page = (
self.book.page_set.filter(created__lt=self.created)
.order_by("-created")
.first()
)
return page
@property
def seconds_to_next_page(self) -> int:
seconds = 999999 # Effectively infnity time as we have no next
if not self.end_time:
self._set_end_time()
if self.next:
seconds = (self.next.start_time - self.end_time).seconds
return seconds
@property
def is_scrobblable(self) -> bool:
"""A page defines the start of a scrobble if the seconds to next page
are greater than an hour, or 3600 seconds, and it's not a single page,
so the next seconds to next_page is less than an hour as well.
As a special case, the first recorded page is a scrobble, so we establish
when the book was started.
"""
is_scrobblable = False
over_an_hour_since_last_page = False
if not self.previous:
is_scrobblable = True
if self.previous:
over_an_hour_since_last_page = (
self.previous.seconds_to_next_page >= 3600
)
blip = self.seconds_to_next_page >= 3600
if over_an_hour_since_last_page and not blip:
is_scrobblable = True
return is_scrobblable
def _set_end_time(self) -> None:
if self.end_time:
return
self.end_time = self.start_time + timedelta(
seconds=self.duration_seconds
)
self.save(update_fields=["end_time"])
return int((last_scrobble.book_pages_read / self.pages) * 100)

View File

@ -1,133 +0,0 @@
import json
import logging
from typing import Optional
import urllib
import requests
logger = logging.getLogger(__name__)
ISBN_URL = "https://openlibrary.org/isbn/{isbn}.json"
SEARCH_URL = "https://openlibrary.org/search.json?q={query}&sort=editions&mode=everything"
AUTHOR_SEARCH_URL = "https://openlibrary.org/search/authors.json?q={query}"
COVER_URL = "https://covers.openlibrary.org/b/olid/{id}-L.jpg"
AUTHOR_URL = "https://openlibrary.org/authors/{id}.json"
AUTHOR_IMAGE_URL = "https://covers.openlibrary.org/a/olid/{id}-L.jpg"
def get_first(key: str, result: dict) -> str:
obj = ""
if obj_list := result.get(key):
obj = obj_list[0]
return obj
def get_author_openlibrary_id(name: str) -> str:
search_url = AUTHOR_SEARCH_URL.format(query=name)
response = requests.get(search_url)
if response.status_code != 200:
logger.warn(f"Bad response from OL: {response.status_code}")
return ""
results = json.loads(response.content)
if not results:
logger.warn(f"No author results found from search for {name}")
return ""
result = results.get("docs", [])
return result[0].get("key")
def lookup_author_from_openlibrary(olid: str) -> dict:
author_url = AUTHOR_URL.format(id=olid)
response = requests.get(author_url)
if response.status_code != 200:
logger.warn(f"Bad response from OL: {response.status_code}")
return {}
results = json.loads(response.content)
if not results:
logger.warn(f"No author results found from OL for {olid}")
return {}
remote_ids = results.get("remote_ids", {})
bio = ""
if results.get("bio"):
try:
bio = results.get("bio").get("value")
except AttributeError:
bio = results.get("bio")
return {
"name": results.get("name"),
"openlibrary_id": olid,
"wikipedia_url": results.get("wikipedia"),
"wikidata_id": remote_ids.get("wikidata"),
"isni": remote_ids.get("isni"),
"goodreads_id": remote_ids.get("goodreads"),
"librarything_id": remote_ids.get("librarything"),
"amazon_id": remote_ids.get("amazon"),
"bio": bio,
"author_headshot_url": AUTHOR_IMAGE_URL.format(id=olid),
}
def lookup_book_from_openlibrary(
title: str, author: Optional[str] = None
) -> dict:
title_quoted = urllib.parse.quote(title)
author_quoted = ""
if author:
author_quoted = urllib.parse.quote(author)
query = f"{title_quoted} {author_quoted}"
search_url = SEARCH_URL.format(query=query)
response = requests.get(search_url)
if response.status_code != 200:
logger.warn(f"Bad response from OL: {response.status_code}")
return {}
results = json.loads(response.content)
if len(results.get("docs")) == 0:
logger.warn(f"No results found from OL for {title}")
return {}
top = None
for result in results.get("docs"):
# These Summary things suck and ruin our one-shot search
if "Summary of" not in result.get("title"):
top = result
break
if not top:
logger.warn(f"No book found for query {query}")
return {}
ol_id = top.get("cover_edition_key")
ol_author_id = get_first("author_key", top)
first_sentence = ""
if top.get("first_sentence"):
try:
first_sentence = top.get("first_sentence")[0].get("value")
except AttributeError:
first_sentence = top.get("first_sentence")[0]
isbn = None
if top.get("isbn"):
isbn = top.get("isbn")[0]
return {
"title": top.get("title"),
"isbn": isbn,
"openlibrary_id": ol_id,
"goodreads_id": get_first("id_goodreads", top),
"first_publish_year": top.get("first_publish_year"),
"first_sentence": first_sentence,
"pages": top.get("number_of_pages_median", None),
"cover_url": COVER_URL.format(id=ol_id),
"ol_author_id": ol_author_id,
"subject_key_list": top.get("subject_key", []),
}

View File

@ -1,19 +0,0 @@
from django.urls import path
from books import views
app_name = "books"
urlpatterns = [
path("book/", views.BookListView.as_view(), name="book_list"),
path(
"book/<slug:slug>/",
views.BookDetailView.as_view(),
name="book_detail",
),
path(
"author/<slug:slug>/",
views.AuthorDetailView.as_view(),
name="author_detail",
),
]

View File

@ -0,0 +1,47 @@
import json
from typing import Optional
import requests
import logging
logger = logging.getLogger(__name__)
SEARCH_URL = "https://openlibrary.org/search.json?title={title}"
ISBN_URL = "https://openlibrary.org/isbn/{isbn}.json"
def get_first(key: str, result: dict) -> str:
obj = ""
if obj_list := result.get(key):
obj = obj_list[0]
return obj
def lookup_book_from_openlibrary(title: str, author: str = None) -> dict:
search_url = SEARCH_URL.format(title=title)
response = requests.get(search_url)
if response.status_code != 200:
logger.warn(f"Bad response from OL: {response.status_code}")
return {}
results = json.loads(response.content)
if len(results.get('docs')) == 0:
logger.warn(f"No results found from OL for {title}")
return {}
top = results.get('docs')[0]
if author and author not in top['author_name']:
logger.warn(
f"Lookup for {title} found top result with mismatched author"
)
return {
"title": top.get("title"),
"isbn": top.get("isbn")[0],
"openlibrary_id": top.get("cover_edition_key"),
"author_name": get_first("author_name", top),
"author_openlibrary_id": get_first("author_key", top),
"goodreads_id": get_first("id_goodreads", top),
"first_publish_year": top.get("first_publish_year"),
}

View File

@ -1,17 +0,0 @@
from django.views import generic
from books.models import Book, Author
class BookListView(generic.ListView):
model = Book
paginate_by = 20
class BookDetailView(generic.DetailView):
model = Book
slug_field = "uuid"
class AuthorDetailView(generic.DetailView):
model = Author
slug_field = "uuid"

View File

@ -1,36 +0,0 @@
from django.contrib import admin
from locations.models import GeoLocation, RawGeoLocation
from scrobbles.admin import ScrobbleInline
@admin.register(GeoLocation)
class GeoLocationAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"created",
"lat",
"lon",
"title",
"altitude",
)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]
@admin.register(RawGeoLocation)
class RawGeoLocationAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"lat",
"lon",
"altitude",
"speed",
)
ordering = (
"lat",
"lon",
)

View File

@ -1,6 +0,0 @@
from django.apps import AppConfig
class LocationConfig(AppConfig):
name = "locations"

View File

@ -1,7 +0,0 @@
#!/usr/bin/env python3
from collections import defaultdict
LOCATION_PROVIDERS = defaultdict(lambda: "Unknown")
LOCATION_PROVIDERS["gps"] = "GPS"
LOCATION_PROVIDERS["network"] = "Wifi Triangulation"

View File

@ -1,77 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-21 23:29
from django.db import migrations, models
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
initial = True
operations = [
migrations.CreateModel(
name="GeoLocation",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"created",
django_extensions.db.fields.CreationDateTimeField(
auto_now_add=True, verbose_name="created"
),
),
(
"modified",
django_extensions.db.fields.ModificationDateTimeField(
auto_now=True, verbose_name="modified"
),
),
(
"title",
models.CharField(blank=True, max_length=255, null=True),
),
(
"run_time_seconds",
models.IntegerField(blank=True, null=True),
),
(
"run_time_ticks",
models.PositiveBigIntegerField(blank=True, null=True),
),
(
"uuid",
models.UUIDField(
blank=True,
default=uuid.uuid4,
editable=False,
null=True,
),
),
("lat", models.FloatField()),
("lon", models.FloatField()),
("altitude", models.FloatField(blank=True, null=True)),
(
"genre",
taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
],
options={
"unique_together": {("lat", "lon", "altitude")},
},
),
]

View File

@ -1,58 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-22 00:13
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import django_extensions.db.fields
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("locations", "0001_initial"),
]
operations = [
migrations.CreateModel(
name="RawGeoLocation",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
(
"created",
django_extensions.db.fields.CreationDateTimeField(
auto_now_add=True, verbose_name="created"
),
),
(
"modified",
django_extensions.db.fields.ModificationDateTimeField(
auto_now=True, verbose_name="modified"
),
),
("lat", models.FloatField()),
("lon", models.FloatField()),
("altitude", models.FloatField(blank=True, null=True)),
("speed", models.FloatField(blank=True, null=True)),
(
"user",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
to=settings.AUTH_USER_MODEL,
),
),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-22 23:56
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("locations", "0002_rawgeolocation"),
]
operations = [
migrations.AddField(
model_name="rawgeolocation",
name="timestamp",
field=models.DateTimeField(blank=True, null=True),
),
]

View File

@ -1,17 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-24 12:45
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("locations", "0003_rawgeolocation_timestamp"),
]
operations = [
migrations.AlterModelOptions(
name="rawgeolocation",
options={},
),
]

View File

@ -1,27 +0,0 @@
# Generated by Django 4.1.7 on 2023-11-24 18:17
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("locations", "0004_alter_rawgeolocation_options"),
]
operations = [
migrations.AlterModelOptions(
name="rawgeolocation",
options={"get_latest_by": "modified"},
),
migrations.AddField(
model_name="geolocation",
name="truncated_lat",
field=models.FloatField(blank=True, null=True),
),
migrations.AddField(
model_name="geolocation",
name="truncated_lon",
field=models.FloatField(blank=True, null=True),
),
]

View File

@ -1,96 +0,0 @@
import logging
from typing import Dict
from uuid import uuid4
from django.contrib.auth import get_user_model
from django.conf import settings
from django.db import models
from django.urls import reverse
from django_extensions.db.models import TimeStampedModel
from scrobbles.mixins import ScrobblableMixin
logger = logging.getLogger(__name__)
BNULL = {"blank": True, "null": True}
User = get_user_model()
class GeoLocation(ScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, "LOCATION_COMPLETION_PERCENT", 100)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
lat = models.FloatField()
lon = models.FloatField()
truncated_lat = models.FloatField(**BNULL)
truncated_lon = models.FloatField(**BNULL)
altitude = models.FloatField(**BNULL)
class Meta:
unique_together = [["lat", "lon", "altitude"]]
def __str__(self):
if self.title:
return self.title
return f"{self.lat} x {self.lon}"
def get_absolute_url(self):
return reverse(
"locations:geo_location_detail", kwargs={"slug": self.uuid}
)
@classmethod
def find_or_create(cls, data_dict: Dict) -> "GeoLocation":
"""Given a data dict from GPSLogger, does the heavy lifting of looking up
the location, creating if if doesn't exist yet.
"""
# TODO Add constants for all these data keys
if "lat" not in data_dict.keys() or "lon" not in data_dict.keys():
logger.error("No lat or lon keys in data dict")
return
int_lat, r_lat = str(data_dict.get("lat", "")).split(".")
int_lon, r_lon = str(data_dict.get("lon", "")).split(".")
try:
trunc_lat = r_lat[0:4]
except IndexError:
trunc_lat = r_lat
try:
trunc_lon = r_lon[0:4]
except IndexError:
trunc_lon = r_lon
data_dict["lat"] = float(f"{int_lat}.{trunc_lat}")
data_dict["lon"] = float(f"{int_lon}.{trunc_lon}")
int_alt, r_alt = str(data_dict.get("alt", "")).split(".")
try:
trunc_alt = r_lon[0:4]
except IndexError:
trunc_alt = r_alt
data_dict["altitude"] = float(f"{int_alt}.{trunc_alt}")
location = cls.objects.filter(
lat=data_dict.get("lat"),
lon=data_dict.get("lon"),
altitude=data_dict.get("altitude"),
).first()
if not location:
location = cls.objects.create(
lat=data_dict.get("lat"),
lon=data_dict.get("lon"),
altitude=data_dict.get("altitude"),
)
return location
class RawGeoLocation(TimeStampedModel):
user = models.ForeignKey(User, on_delete=models.CASCADE)
lat = models.FloatField()
lon = models.FloatField()
altitude = models.FloatField(**BNULL)
speed = models.FloatField(**BNULL)
timestamp = models.DateTimeField(**BNULL)

View File

@ -1,18 +0,0 @@
from django.urls import path
from locations import views
app_name = "locations"
urlpatterns = [
path(
"locations/",
views.GeoLocationListView.as_view(),
name="geo_locations_list",
),
path(
"locations/<slug:slug>/",
views.GeoLocationDetailView.as_view(),
name="geo_location_detail",
),
]

View File

@ -1,22 +0,0 @@
from django.db.models import Count
from django.views import generic
from locations.models import GeoLocation
from scrobbles.stats import get_scrobble_count_qs
class GeoLocationListView(generic.ListView):
model = GeoLocation
paginate_by = 75
def get_queryset(self):
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)
context_data["latest"] = self.get_queryset().first()
return context_data
class GeoLocationDetailView(generic.DetailView):
model = GeoLocation
slug_field = "uuid"

View File

@ -11,7 +11,7 @@ class AlbumAdmin(admin.ModelAdmin):
list_display = (
"name",
"year",
"album_artist",
"primary_artist",
"theaudiodb_genre",
"theaudiodb_mood",
"musicbrainz_id",
@ -20,28 +20,17 @@ class AlbumAdmin(admin.ModelAdmin):
"theaudiodb_score",
"theaudiodb_genre",
)
ordering = ("-created",)
search_fields = ("name",)
ordering = ("name",)
filter_horizontal = [
"artists",
'artists',
]
@admin.register(Artist)
class ArtistAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"theaudiodb_mood",
"theaudiodb_genre",
"musicbrainz_id",
)
list_filter = (
"theaudiodb_mood",
"theaudiodb_genre",
)
search_fields = ("name",)
ordering = ("-created",)
list_display = ("name", "musicbrainz_id")
ordering = ("name",)
@admin.register(Track)
@ -51,10 +40,10 @@ class TrackAdmin(admin.ModelAdmin):
"title",
"album",
"artist",
"run_time",
"musicbrainz_id",
)
list_filter = ("album", "artist")
search_fields = ("title",)
ordering = ("-created",)
inlines = [
ScrobbleInline,

View File

@ -1,11 +1,13 @@
from datetime import datetime, timedelta
from typing import List
from django.apps import apps
from django.db.models import Count, Q, QuerySet
from django.utils import timezone
from profiles.utils import now_user_timezone
from music.models import Artist, Track
from scrobbles.models import Scrobble
from videos.models import Video
from vrobbler.apps.profiles.utils import now_user_timezone
def scrobble_counts(user=None):
@ -26,29 +28,27 @@ def scrobble_counts(user=None):
starting_day_of_current_year = now.date().replace(month=1, day=1)
finished_scrobbles_qs = Scrobble.objects.filter(
user_filter,
played_to_completion=True,
media_type=Scrobble.MediaType.TRACK,
user_filter, played_to_completion=True
)
data = {}
data["today"] = finished_scrobbles_qs.filter(
data['today'] = finished_scrobbles_qs.filter(
timestamp__gte=start_of_today
).count()
data["week"] = finished_scrobbles_qs.filter(
data['week'] = finished_scrobbles_qs.filter(
timestamp__gte=starting_day_of_current_week
).count()
data["month"] = finished_scrobbles_qs.filter(
data['month'] = finished_scrobbles_qs.filter(
timestamp__gte=starting_day_of_current_month
).count()
data["year"] = finished_scrobbles_qs.filter(
data['year'] = finished_scrobbles_qs.filter(
timestamp__gte=starting_day_of_current_year
).count()
data["alltime"] = finished_scrobbles_qs.count()
data['alltime'] = finished_scrobbles_qs.count()
return data
def week_of_scrobbles(
user=None, start=None, media: str = "tracks"
user=None, start=None, media: str = 'tracks'
) -> dict[str, int]:
now = timezone.now()
@ -64,15 +64,15 @@ def week_of_scrobbles(
base_qs = Scrobble.objects.filter(user_filter, played_to_completion=True)
media_filter = Q(track__isnull=False)
if media == "movies":
if media == 'movies':
media_filter = Q(video__video_type=Video.VideoType.MOVIE)
if media == "series":
if media == 'series':
media_filter = Q(video__video_type=Video.VideoType.TV_EPISODE)
for day in range(6, -1, -1):
start_day = start - timedelta(days=day)
end = datetime.combine(start_day, datetime.max.time(), now.tzinfo)
day_of_week = start_day.strftime("%A")
day_of_week = start_day.strftime('%A')
scrobble_day_dict[day_of_week] = base_qs.filter(
media_filter,
@ -97,25 +97,19 @@ def live_charts(
now = now_user_timezone(user.profile)
tzinfo = now.tzinfo
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 = now - timedelta(days=now.today().isoweekday() % 7)
start_day_of_month = now.replace(day=1)
start_day_of_year = now.replace(month=1, day=1)
media_model = apps.get_model(app_label="music", model_name=media_type)
media_model = apps.get_model(app_label='music', model_name=media_type)
period_queries = {
"today": {"scrobble__timestamp__gte": start_of_today},
"week": {"scrobble__timestamp__gte": start_day_of_week},
"last7": {"scrobble__timestamp__gte": seven_days_ago},
"last30": {"scrobble__timestamp__gte": thirty_days_ago},
"month": {"scrobble__timestamp__gte": start_day_of_month},
"year": {"scrobble__timestamp__gte": start_day_of_year},
"all": {},
'today': {'scrobble__timestamp__gte': start_of_today},
'week': {'scrobble__timestamp__gte': start_day_of_week},
'month': {'scrobble__timestamp__gte': start_day_of_month},
'year': {'scrobble__timestamp__gte': start_day_of_year},
'all': {},
}
time_filter = Q()

View File

@ -1,85 +0,0 @@
import urllib
from typing import Optional
from bs4 import BeautifulSoup
import requests
import logging
logger = logging.getLogger(__name__)
ALLMUSIC_SEARCH_URL = "https://www.allmusic.com/search/{subpath}/{query}"
def strip_and_clean(text):
return text.strip("\n").rstrip().lstrip()
def get_rating_from_soup(soup) -> Optional[int]:
rating = None
try:
potential_rating = soup.find("div", class_="allmusic-rating")
if potential_rating:
rating = int(strip_and_clean(potential_rating.get_text()))
except ValueError:
pass
return rating
def get_review_from_soup(soup) -> str:
review = ""
try:
potential_text = soup.find("div", class_="text")
if potential_text:
review = strip_and_clean(potential_text.get_text())
except ValueError:
pass
return review
def scrape_data_from_allmusic(url) -> dict:
data_dict = {}
headers = {"User-Agent": "Vrobbler 0.11.12"}
r = requests.get(url, headers=headers)
if r.status_code == 200:
soup = BeautifulSoup(r.text, "html.parser")
data_dict["rating"] = get_rating_from_soup(soup)
data_dict["review"] = get_review_from_soup(soup)
return data_dict
def get_allmusic_slug(artist_name=None, album_name=None) -> str:
slug = ""
if not artist_name:
return slug
subpath = "artists"
class_ = "name"
query = urllib.parse.quote(artist_name)
if album_name:
subpath = "albums"
class_ = "title"
query = "+".join([query, urllib.parse.quote(album_name)])
url = ALLMUSIC_SEARCH_URL.format(subpath=subpath, query=query)
headers = {"User-Agent": "Vrobbler 0.11.12"}
r = requests.get(url, headers=headers)
if r.status_code != 200:
logger.info(f"Bad http response from Allmusic {r}")
return slug
soup = BeautifulSoup(r.text, "html.parser")
results = soup.find("ul", class_="search-results")
if not results:
logger.info(f"No search results for {query}")
return slug
prime_result = results.findAll("div", class_=class_)
if not prime_result:
logger.info(f"Could not find specific result for search {query}")
result_url = prime_result[0].find_all("a")[0]["href"]
slug = result_url.split("/")[-1:][0]
return slug

View File

@ -9,18 +9,18 @@ from music.models import Artist, Album, Track
class ArtistViewSet(viewsets.ModelViewSet):
queryset = Artist.objects.all().order_by("-created")
queryset = Artist.objects.all().order_by('-created')
serializer_class = ArtistSerializer
permission_classes = [permissions.IsAuthenticated]
class AlbumViewSet(viewsets.ModelViewSet):
queryset = Album.objects.all().order_by("-created")
queryset = Album.objects.all().order_by('-created')
serializer_class = AlbumSerializer
permission_classes = [permissions.IsAuthenticated]
class TrackViewSet(viewsets.ModelViewSet):
queryset = Track.objects.all().order_by("-created")
queryset = Track.objects.all().order_by('-created')
serializer_class = TrackSerializer
permission_classes = [permissions.IsAuthenticated]

View File

@ -2,4 +2,4 @@ from django.apps import AppConfig
class MusicConfig(AppConfig):
name = "music"
name = 'music'

View File

@ -1,49 +0,0 @@
import logging
import urllib
import requests
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
BANDCAMP_SEARCH_URL = "https://bandcamp.com/search?q={query}&item_type={itype}"
def get_bandcamp_slug(artist_name=None, album_name=None) -> str:
slug = ""
if not artist_name:
return slug
query = urllib.parse.quote(artist_name)
item_type = "b"
class_ = "heading"
if album_name:
item_type = "a"
query = "+".join([query, urllib.parse.quote(album_name)])
url = BANDCAMP_SEARCH_URL.format(query=query, itype=item_type)
headers = {"User-Agent": "Vrobbler 0.11.12"}
r = requests.get(url, headers=headers)
if r.status_code != 200:
logger.info(f"Bad http response from Bandcamp {r}")
return slug
soup = BeautifulSoup(r.text, "html.parser")
results = soup.find("ul", class_="result-items")
if not results:
logger.info(f"No search results for {query}")
return slug
prime_result = results.findAll("div", class_=class_)
if not prime_result:
logger.info(f"Could not find specific result for search {query}")
result_url = prime_result[0].find_all("a")[0]["href"]
if item_type == "b":
slug = result_url.split("/")[2].split(".")[0]
else:
slug = result_url.split("?")[0]
return slug

View File

@ -1,21 +1,16 @@
VARIOUS_ARTIST_DICT = {
"name": "Various Artists",
"theaudiodb_id": "113641",
"musicbrainz_id": "89ad4ac3-39f7-470e-963a-56509c546377",
}
JELLYFIN_POST_KEYS = {
"ITEM_TYPE": "ItemType",
"RUN_TIME": "RunTime",
"TITLE": "Name",
"TIMESTAMP": "UtcTimestamp",
"YEAR": "Year",
"PLAYBACK_POSITION_TICKS": "PlaybackPositionTicks",
"PLAYBACK_POSITION": "PlaybackPosition",
"ARTIST_MB_ID": "Provider_musicbrainzartist",
"ALBUM_MB_ID": "Provider_musicbrainzalbum",
"RELEASEGROUP_MB_ID": "Provider_musicbrainzreleasegroup",
"TRACK_MB_ID": "Provider_musicbrainztrack",
"ALBUM_NAME": "Album",
"ARTIST_NAME": "Artist",
'ITEM_TYPE': 'ItemType',
'RUN_TIME_TICKS': 'RunTimeTicks',
'RUN_TIME': 'RunTime',
'TITLE': 'Name',
'TIMESTAMP': 'UtcTimestamp',
'YEAR': 'Year',
'PLAYBACK_POSITION_TICKS': 'PlaybackPositionTicks',
'PLAYBACK_POSITION': 'PlaybackPosition',
'ARTIST_MB_ID': 'Provider_musicbrainzartist',
'ALBUM_MB_ID': 'Provider_musicbrainzalbum',
'RELEASEGROUP_MB_ID': 'Provider_musicbrainzreleasegroup',
'TRACK_MB_ID': 'Provider_musicbrainztrack',
'ALBUM_NAME': 'Album',
'ARTIST_NAME': 'Artist',
}

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:01
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("music", "0014_album_theaudiodb_year_released"),
]
operations = [
migrations.AlterField(
model_name="track",
name="run_time",
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:21
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("music", "0015_alter_track_run_time"),
]
operations = [
migrations.RenameField(
model_name="track",
old_name="run_time",
new_name="run_time_seconds",
),
]

View File

@ -1,25 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-14 22:27
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0033_genre_objectwithgenres"),
("music", "0016_rename_run_time_track_run_time_seconds"),
]
operations = [
migrations.AddField(
model_name="track",
name="genre",
field=taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,23 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-15 01:17
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("music", "0017_track_genre"),
]
operations = [
migrations.AddField(
model_name="album",
name="allmusic_rating",
field=models.IntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="album",
name="allmusic_review",
field=models.TextField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-15 03:05
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("music", "0018_album_allmusic_rating_album_allmusic_review"),
]
operations = [
migrations.AddField(
model_name="artist",
name="allmusic_id",
field=models.CharField(blank=True, max_length=100, null=True),
),
]

View File

@ -1,23 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-15 03:32
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("music", "0019_artist_allmusic_id"),
]
operations = [
migrations.AddField(
model_name="album",
name="bandcamp_id",
field=models.CharField(blank=True, max_length=100, null=True),
),
migrations.AddField(
model_name="artist",
name="bandcamp_id",
field=models.CharField(blank=True, max_length=100, null=True),
),
]

View File

@ -1,25 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-15 16:22
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
("music", "0020_album_bandcamp_id_artist_bandcamp_id"),
]
operations = [
migrations.AddField(
model_name="album",
name="album_artist",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
related_name="albums",
to="music.artist",
),
),
]

View File

@ -1,20 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-15 16:29
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("music", "0021_album_album_artist"),
]
operations = [
migrations.AddField(
model_name="artist",
name="theaudiodb_id",
field=models.CharField(
blank=True, max_length=255, null=True, unique=True
),
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.1.7 on 2023-04-18 02:33
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0040_alter_scrobble_media_type"),
("music", "0022_artist_theaudiodb_id"),
]
operations = [
migrations.AlterField(
model_name="track",
name="genre",
field=taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -5,19 +5,15 @@ from urllib.request import urlopen
from uuid import uuid4
import musicbrainzngs
import requests
from django.conf import settings
from django.core.files.base import ContentFile, File
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 imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFit
from music.allmusic import get_allmusic_slug, scrape_data_from_allmusic
from music.bandcamp import get_bandcamp_slug
from music.theaudiodb import lookup_album_from_tadb, lookup_artist_from_tadb
from scrobbles.mixins import ScrobblableMixin
from scrobbles.theaudiodb import lookup_artist_from_tadb
from vrobbler.apps.scrobbles.theaudiodb import lookup_album_from_tadb
logger = logging.getLogger(__name__)
BNULL = {"blank": True, "null": True}
@ -27,157 +23,70 @@ class Artist(TimeStampedModel):
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
name = models.CharField(max_length=255)
biography = models.TextField(**BNULL)
theaudiodb_id = models.CharField(max_length=255, unique=True, **BNULL)
theaudiodb_genre = models.CharField(max_length=255, **BNULL)
theaudiodb_mood = models.CharField(max_length=255, **BNULL)
musicbrainz_id = models.CharField(max_length=255, **BNULL)
allmusic_id = models.CharField(max_length=100, **BNULL)
bandcamp_id = models.CharField(max_length=100, **BNULL)
thumbnail = models.ImageField(upload_to="artist/", **BNULL)
thumbnail_small = ImageSpecField(
source="thumbnail",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
thumbnail_medium = ImageSpecField(
source="thumbnail",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
class Meta:
unique_together = [["name", "musicbrainz_id"]]
unique_together = [['name', 'musicbrainz_id']]
def __str__(self):
return self.name
@property
def primary_image_url(self) -> str:
if self.thumbnail:
return self.thumbnail.url
if self.album_set.first().cover_image:
return self.album_set.first().cover_image.url
return ""
@property
def mb_link(self):
return f"https://musicbrainz.org/artist/{self.musicbrainz_id}"
@property
def allmusic_link(self):
if self.allmusic_id:
return f"https://www.allmusic.com/artist/{self.allmusic_id}"
return ""
@property
def bandcamp_link(self):
if self.bandcamp_id:
return f"https://{self.bandcamp_id}.bandcamp.com/"
return ""
def get_absolute_url(self):
return reverse("music:artist_detail", kwargs={"slug": self.uuid})
return reverse('music:artist_detail', kwargs={'slug': self.uuid})
def scrobbles(self):
from scrobbles.models import Scrobble
return Scrobble.objects.filter(
track__in=self.track_set.all()
).order_by("-timestamp")
).order_by('-timestamp')
@property
def tracks(self):
return (
self.track_set.all()
.annotate(scrobble_count=models.Count("scrobble"))
.order_by("-scrobble_count")
.annotate(scrobble_count=models.Count('scrobble'))
.order_by('-scrobble_count')
)
def charts(self):
from scrobbles.models import ChartRecord
return ChartRecord.objects.filter(track__artist=self).order_by("-year")
def scrape_allmusic(self, force=False) -> None:
if not self.allmusic_id or force:
slug = get_allmusic_slug(self.name)
if not slug:
logger.info(f"No allmsuic link for {self}")
return
self.allmusic_id = slug
self.save(update_fields=["allmusic_id"])
def scrape_bandcamp(self, force=False) -> None:
if not self.bandcamp_id or force:
slug = get_bandcamp_slug(self.name)
if not slug:
logger.info(f"No bandcamp link for {self}")
return
self.bandcamp_id = slug
self.save(update_fields=["bandcamp_id"])
def fix_metadata(self, force_update=False):
tadb_info = {}
if self.theaudiodb_id and force_update:
tadb_info = lookup_artist_from_tadb(self.theaudiodb_id)
if not self.theaudiodb_id or (force_update and not tadb_info):
tadb_info = lookup_artist_from_tadb(self.name)
return ChartRecord.objects.filter(track__artist=self).order_by('-year')
def fix_metadata(self):
tadb_info = lookup_artist_from_tadb(self.name)
if not tadb_info:
logger.warn(
f"No response from TADB for artist {self.name}, try force_update=True"
)
logger.warn(f"No response from TADB for artist {self.name}")
return
self.biography = tadb_info["biography"]
self.theaudiodb_genre = tadb_info["genre"]
self.theaudiodb_mood = tadb_info["mood"]
self.biography = tadb_info['biography']
self.theaudiodb_genre = tadb_info['genre']
self.theaudiodb_mood = tadb_info['mood']
thumb_url = tadb_info.get("thumb_url", "")
if thumb_url:
r = requests.get(thumb_url)
if r.status_code == 200:
fname = f"{self.name}_{self.uuid}.jpg"
self.thumbnail.save(fname, ContentFile(r.content), save=True)
@property
def rym_link(self):
artist_slug = self.name.lower().replace(" ", "-")
return f"https://rateyourmusic.com/artist/{artist_slug}/"
@property
def bandcamp_search_link(self):
artist = self.name.lower()
return f"https://bandcamp.com/search?q={artist}&item_type=b"
img_temp = NamedTemporaryFile(delete=True)
img_temp.write(urlopen(tadb_info['thumb_url']).read())
img_temp.flush()
img_filename = f"{self.name}_{self.uuid}.jpg"
self.thumbnail.save(img_filename, File(img_temp))
class Album(TimeStampedModel):
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
name = models.CharField(max_length=255)
album_artist = models.ForeignKey(
Artist, related_name="albums", on_delete=models.DO_NOTHING, **BNULL
)
artists = models.ManyToManyField(Artist)
year = models.IntegerField(**BNULL)
musicbrainz_id = models.CharField(max_length=255, unique=True, **BNULL)
musicbrainz_releasegroup_id = models.CharField(max_length=255, **BNULL)
musicbrainz_albumartist_id = models.CharField(max_length=255, **BNULL)
cover_image = models.ImageField(upload_to="albums/", **BNULL)
cover_image_small = ImageSpecField(
source="cover_image",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
cover_image_medium = ImageSpecField(
source="cover_image",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
theaudiodb_id = models.CharField(max_length=255, unique=True, **BNULL)
theaudiodb_description = models.TextField(**BNULL)
theaudiodb_year_released = models.IntegerField(**BNULL)
@ -189,9 +98,6 @@ class Album(TimeStampedModel):
theaudiodb_speed = models.CharField(max_length=255, **BNULL)
theaudiodb_theme = models.CharField(max_length=255, **BNULL)
allmusic_id = models.CharField(max_length=255, **BNULL)
allmusic_rating = models.IntegerField(**BNULL)
allmusic_review = models.TextField(**BNULL)
bandcamp_id = models.CharField(max_length=100, **BNULL)
rateyourmusic_id = models.CharField(max_length=255, **BNULL)
wikipedia_slug = models.CharField(max_length=255, **BNULL)
discogs_id = models.CharField(max_length=255, **BNULL)
@ -201,101 +107,57 @@ class Album(TimeStampedModel):
return self.name
def get_absolute_url(self):
return reverse("music:album_detail", kwargs={"slug": self.uuid})
return reverse("music:album_detail", kwargs={'slug': self.uuid})
def scrobbles(self):
from scrobbles.models import Scrobble
return Scrobble.objects.filter(
track__in=self.track_set.all()
).order_by("-timestamp")
@property
def primary_image_url(self) -> str:
if self.cover_image:
return self.cover_image_medium.url
return ""
).order_by('-timestamp')
@property
def tracks(self):
return (
self.track_set.all()
.annotate(scrobble_count=models.Count("scrobble"))
.order_by("-scrobble_count")
.annotate(scrobble_count=models.Count('scrobble'))
.order_by('-scrobble_count')
)
def fix_album_artist(self):
from music.utils import get_or_create_various_artists
multiple_artists = self.artists.count() > 1
if multiple_artists:
self.album_artist = get_or_create_various_artists()
else:
self.album_artist = self.artists.first()
self.save(update_fields=["album_artist"])
def scrape_allmusic(self, force=False) -> None:
if not self.allmusic_id or force:
slug = get_allmusic_slug(self.name, self.album_artist.name)
if not slug:
logger.info(
f"No allmsuic link for {self} by {self.album_artist}"
)
return
self.allmusic_id = slug
self.save(update_fields=["allmusic_id"])
allmusic_data = scrape_data_from_allmusic(self.allmusic_link)
if not allmusic_data:
logger.info(f"No allmsuic data for {self} by {self.album_artist}")
return
self.allmusic_review = allmusic_data["review"]
self.allmusic_rating = allmusic_data["rating"]
self.save(update_fields=["allmusic_review", "allmusic_rating"])
@property
def primary_artist(self):
return self.artists.first()
def scrape_theaudiodb(self) -> None:
artist = "Various Artists"
if self.album_artist:
artist = self.album_artist.name
artist = self.primary_artist.name
album_data = lookup_album_from_tadb(self.name, artist)
if not album_data.get("theaudiodb_id"):
if not album_data.get('theaudiodb_id'):
logger.info(f"No data for {self} found in TheAudioDB")
return
Album.objects.filter(pk=self.pk).update(**album_data)
def scrape_bandcamp(self, force=False) -> None:
if not self.bandcamp_id or force:
slug = get_bandcamp_slug(self.album_artist.name, self.name)
if not slug:
logger.info(f"No bandcamp link for {self}")
return
self.bandcamp_id = slug
self.save(update_fields=["bandcamp_id"])
def fix_metadata(self):
if (
not self.musicbrainz_albumartist_id
or not self.year
or not self.musicbrainz_releasegroup_id
):
musicbrainzngs.set_useragent("vrobbler", "0.3.0")
musicbrainzngs.set_useragent('vrobbler', '0.3.0')
mb_data = musicbrainzngs.get_release_by_id(
self.musicbrainz_id, includes=["artists", "release-groups"]
self.musicbrainz_id, includes=['artists', 'release-groups']
)
if not self.musicbrainz_releasegroup_id:
self.musicbrainz_releasegroup_id = mb_data["release"][
"release-group"
]["id"]
self.musicbrainz_releasegroup_id = mb_data['release'][
'release-group'
]['id']
if not self.musicbrainz_albumartist_id:
self.musicbrainz_albumartist_id = mb_data["release"][
"artist-credit"
][0]["artist"]["id"]
self.musicbrainz_albumartist_id = mb_data['release'][
'artist-credit'
][0]['artist']['id']
if not self.year:
try:
self.year = mb_data["release"]["date"][0:4]
self.year = mb_data['release']['date'][0:4]
except KeyError:
pass
except IndexError:
@ -303,9 +165,9 @@ class Album(TimeStampedModel):
self.save(
update_fields=[
"musicbrainz_albumartist_id",
"musicbrainz_releasegroup_id",
"year",
'musicbrainz_albumartist_id',
'musicbrainz_releasegroup_id',
'year',
]
)
@ -319,12 +181,10 @@ class Album(TimeStampedModel):
self.artists.add(t.artist)
if (
not self.cover_image
or self.cover_image == "default-image-replace-me"
or self.cover_image == 'default-image-replace-me'
):
self.fetch_artwork()
self.fix_album_artist()
self.scrape_theaudiodb()
self.scrape_allmusic()
def fetch_artwork(self, force=False):
if not self.cover_image and not force:
@ -335,10 +195,10 @@ class Album(TimeStampedModel):
)
name = f"{self.name}_{self.uuid}.jpg"
self.cover_image = ContentFile(img_data, name=name)
logger.info(f"Setting image to {name}")
logger.info(f'Setting image to {name}')
except musicbrainzngs.ResponseError:
logger.warning(
f"No cover art found for {self.name} by release"
f'No cover art found for {self.name} by release'
)
if (
@ -351,10 +211,10 @@ class Album(TimeStampedModel):
)
name = f"{self.name}_{self.uuid}.jpg"
self.cover_image = ContentFile(img_data, name=name)
logger.info(f"Setting image to {name}")
logger.info(f'Setting image to {name}')
except musicbrainzngs.ResponseError:
logger.warning(
f"No cover art found for {self.name} by release group"
f'No cover art found for {self.name} by release group'
)
if not self.cover_image:
logger.debug(
@ -369,7 +229,7 @@ class Album(TimeStampedModel):
@property
def allmusic_link(self) -> str:
if self.allmusic_id:
return f"https://www.allmusic.com/album/{self.allmusic_id}"
return f"https://www.allmusic.com/artist/{self.allmusic_id}"
return ""
@property
@ -384,45 +244,27 @@ class Album(TimeStampedModel):
return f"https://www.theaudiodb.com/album/{self.theaudiodb_id}"
return ""
@property
def rym_link(self):
artist_slug = self.album_artist.name.lower().replace(" ", "-")
album_slug = self.name.lower().replace(" ", "-")
return f"https://rateyourmusic.com/release/album/{artist_slug}/{album_slug}/"
@property
def bandcamp_link(self):
if self.bandcamp_id and self.album_artist.bandcamp_id:
return f"https://{self.album_artist.bandcamp_id}.bandcamp.com/album/{self.bandcamp_id}"
return ""
@property
def bandcamp_search_link(self):
artist = self.album_artist.name.lower()
album = self.name.lower()
return f"https://bandcamp.com/search?q={album} {artist}&item_type=a"
class Track(ScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, "MUSIC_COMPLETION_PERCENT", 90)
COMPLETION_PERCENT = getattr(settings, 'MUSIC_COMPLETION_PERCENT', 90)
class Opinion(models.IntegerChoices):
DOWN = -1, "Thumbs down"
NEUTRAL = 0, "No opinion"
UP = 1, "Thumbs up"
DOWN = -1, 'Thumbs down'
NEUTRAL = 0, 'No opinion'
UP = 1, 'Thumbs up'
artist = models.ForeignKey(Artist, on_delete=models.DO_NOTHING)
album = models.ForeignKey(Album, on_delete=models.DO_NOTHING, **BNULL)
musicbrainz_id = models.CharField(max_length=255, **BNULL)
class Meta:
unique_together = [["album", "musicbrainz_id"]]
unique_together = [['album', 'musicbrainz_id']]
def __str__(self):
return f"{self.title} by {self.artist}"
def get_absolute_url(self):
return reverse("music:track_detail", kwargs={"slug": self.uuid})
return reverse('music:track_detail', kwargs={'slug': self.uuid})
@property
def subtitle(self):
@ -436,15 +278,6 @@ class Track(ScrobblableMixin):
def info_link(self):
return self.mb_link
@property
def primary_image_url(self) -> str:
url = ""
if self.artist.thumbnail:
url = self.artist.thumbnail_medium.url
if self.album and self.album.cover_image:
url = self.album.cover_image_medium.url
return url
@classmethod
def find_or_create(
cls, artist_dict: Dict, album_dict: Dict, track_dict: Dict
@ -454,8 +287,8 @@ class Track(ScrobblableMixin):
exist.
"""
if not artist_dict.get("name") or not artist_dict.get(
"musicbrainz_id"
if not artist_dict.get('name') or not artist_dict.get(
'musicbrainz_id'
):
logger.warning(
f"No artist or artist musicbrainz ID found in message from source, not scrobbling"
@ -469,8 +302,8 @@ class Track(ScrobblableMixin):
if not album.cover_image:
album.fetch_artwork()
track_dict["album_id"] = getattr(album, "id", None)
track_dict["artist_id"] = artist.id
track_dict['album_id'] = getattr(album, "id", None)
track_dict['artist_id'] = artist.id
track, created = cls.objects.get_or_create(**track_dict)

View File

@ -1,122 +0,0 @@
import logging
from typing import Iterable
import musicbrainzngs
from dateutil.parser import parse
logger = logging.getLogger(__name__)
def lookup_album_from_mb(musicbrainz_id: str) -> dict:
release_dict = {}
musicbrainzngs.set_useragent("vrobbler", "0.3.0")
release_data = musicbrainzngs.get_release_by_id(
musicbrainz_id,
includes=["artists", "release-groups", "recordings"],
).get("release")
if not release_data:
return release_dict
primary_artist = release_data.get("artist-credit")[0]
release_dict = {
"artist": {
"name": primary_artist.get("name"),
"musicbrainz_id": primary_artist.get("id"),
},
"album": {
"name": release_data.get("title"),
"musicbrainz_id": musicbrainz_id,
"musicbrainz_releasegroup_id": release_data.get(
"release-group"
).get("id"),
"musicbrainz_albumaritist_id": primary_artist.get("id"),
"year": release_data.get("year")[0:4],
},
}
release_dict["tracks"] = []
for track in release_data.get("medium-list")[0]["track-list"]:
recording = track["recording"]
release_dict["tracks"].append(
{
"title": recording["title"],
"musicbrainz_id": recording["id"],
"run_time": track["length"] / 1000,
}
)
return release_dict
def lookup_album_dict_from_mb(release_name: str, artist_name: str) -> dict:
musicbrainzngs.set_useragent("vrobbler", "0.3.0")
top_result = {}
try:
top_result = musicbrainzngs.search_releases(
release_name, artist=artist_name
)["release-list"][0]
except IndexError:
logger.info(
f"No release found on MB for {artist_name} and {release_name}"
)
score = int(top_result.get("ext:score", 0))
if score < 85:
logger.debug(
"Album lookup score below 85 threshold",
extra={"result": top_result},
)
return {}
year = None
if top_result.get("date"):
year = parse(top_result["date"]).year
return {
"year": year,
"title": top_result["title"],
"mb_id": top_result["id"],
"mb_group_id": top_result["release-group"]["id"],
}
def lookup_artist_from_mb(artist_name: str) -> dict:
musicbrainzngs.set_useragent("vrobbler", "0.3.0")
try:
top_result = musicbrainzngs.search_artists(artist=artist_name)[
"artist-list"
][0]
except IndexError:
return {}
score = int(top_result.get("ext:score"))
if score < 85:
logger.debug(
"Artist lookup score below 85 threshold",
extra={"result": top_result},
)
return {}
return top_result
def lookup_track_from_mb(
track_name: str, artist_mbid: str, album_mbid: str
) -> str:
musicbrainzngs.set_useragent("vrobbler", "0.3.0")
top_result = musicbrainzngs.search_recordings(
query=track_name, artist=artist_mbid, release=album_mbid
)["recording-list"][0]
score = int(top_result.get("ext:score"))
if score < 85:
logger.debug(
"Track lookup score below 85 threshold",
extra={"result": top_result},
)
return ""
return top_result

View File

@ -1,107 +0,0 @@
import urllib
import json
import logging
import requests
from django.conf import settings
THEAUDIODB_API_KEY = getattr(settings, "THEAUDIODB_API_KEY")
ARTIST_SEARCH_URL = f"https://www.theaudiodb.com/api/v1/json/{THEAUDIODB_API_KEY}/search.php?s="
ARTIST_FETCH_URL = f"https://www.theaudiodb.com/api/v1/json/{THEAUDIODB_API_KEY}/artist.php?i="
ALBUM_SEARCH_URL = f"https://www.theaudiodb.com/api/v1/json/{THEAUDIODB_API_KEY}/searchalbum.php?s="
logger = logging.getLogger(__name__)
def lookup_artist_from_tadb(name_or_id: str) -> dict:
artist_info = {}
response = None
# Look for an ID as an integer
try:
tadb_id = int(name_or_id)
except ValueError:
tadb_id = None
if tadb_id:
response = requests.get(ARTIST_FETCH_URL + str(tadb_id))
if response.status_code != 200:
logger.warn(f"Bad response from TADB: {response.status_code}")
return artist_info
if not response.content:
logger.warn(f"Bad content from TADB: {response.content}")
return artist_info
if not response:
name = urllib.parse.quote(name_or_id)
response = requests.get(ARTIST_SEARCH_URL + name)
if response.status_code != 200:
logger.warn(f"Bad response from TADB: {response.status_code}")
return artist_info
if not response.content:
logger.warn(f"Bad content from TADB: {response.content}")
return artist_info
if '{"artists": null}' in str(response.content):
logger.warn(f"Bad content from TADB: {response.content}")
return artist_info
results = json.loads(response.content)
if results["artists"]:
artist = results["artists"][0]
artist_info["biography"] = artist.get("strBiographyEN")
artist_info["genre"] = artist.get("strGenre")
artist_info["mood"] = artist.get("strMood")
artist_info["thumb_url"] = artist.get("strArtistThumb")
return artist_info
def lookup_album_from_tadb(name: str, artist: str) -> dict:
album_info = {}
artist = urllib.parse.quote(artist)
name = urllib.parse.quote(name)
response = requests.get("".join([ALBUM_SEARCH_URL, artist, "&a=", name]))
if response.status_code != 200:
logger.warn(f"Bad response from TADB: {response.status_code}")
return {}
if not response.content:
logger.warn(f"Bad content from TADB: {response.content}")
return {}
results = json.loads(response.content)
if results["album"]:
album = results["album"][0]
album_info["theaudiodb_id"] = album.get("idAlbum")
album_info["theaudiodb_description"] = album.get("strDescriptionEN")
album_info["theaudiodb_genre"] = album.get("strGenre")
album_info["theaudiodb_style"] = album.get("strStyle")
album_info["theaudiodb_mood"] = album.get("strMood")
album_info["theaudiodb_speed"] = album.get("strSpeed")
album_info["theaudiodb_theme"] = album.get("strTheme")
album_info["allmusic_id"] = album.get("strAllMusicID")
album_info["wikipedia_slug"] = album.get("strWikipediaID")
album_info["discogs_id"] = album.get("strDiscogsID")
album_info["wikidata_id"] = album.get("strWikidataID")
album_info["rateyourmusic_id"] = album.get("strRateYourMusicID")
if album.get("intYearReleased"):
album_info["theaudiodb_year_released"] = float(
album.get("intYearReleased")
)
if album.get("intScore"):
album_info["theaudiodb_score"] = float(album.get("intScore"))
if album.get("intScoreVotes"):
album_info["theaudiodb_score_votes"] = int(
album.get("intScoreVotes")
)
return album_info

View File

@ -1,26 +1,26 @@
from django.urls import path
from music import views
app_name = "music"
app_name = 'music'
urlpatterns = [
path("albums/", views.AlbumListView.as_view(), name="albums_list"),
path('albums/', views.AlbumListView.as_view(), name='albums_list'),
path(
"album/<slug:slug>/",
'album/<slug:slug>/',
views.AlbumDetailView.as_view(),
name="album_detail",
name='album_detail',
),
path("tracks/", views.TrackListView.as_view(), name="tracks_list"),
path("tracks/", views.TrackListView.as_view(), name='tracks_list'),
path(
"tracks/<slug:slug>/",
'tracks/<slug:slug>/',
views.TrackDetailView.as_view(),
name="track_detail",
name='track_detail',
),
path("artists/", views.ArtistListView.as_view(), name="artist_list"),
path('artists/', views.ArtistListView.as_view(), name='artist_list'),
path(
"artists/<slug:slug>/",
'artists/<slug:slug>/',
views.ArtistDetailView.as_view(),
name="artist_detail",
name='artist_detail',
),
]

View File

@ -1,13 +1,13 @@
import logging
import re
from typing import Optional
from music.musicbrainz import (
from musicbrainzngs.caa import musicbrainz
from scrobbles.musicbrainz import (
lookup_album_dict_from_mb,
lookup_artist_from_mb,
lookup_track_from_mb,
)
from music.constants import VARIOUS_ARTIST_DICT
logger = logging.getLogger(__name__)
@ -17,92 +17,82 @@ from music.models import Album, Artist, Track
def get_or_create_artist(name: str, mbid: str = None) -> Artist:
artist = None
logger.debug(f'Got artist {name} and mbid: {mbid}')
if "feat." in name.lower():
if 'feat.' in name.lower():
name = re.split("feat.", name, flags=re.IGNORECASE)[0].strip()
if "featuring" in name.lower():
if 'featuring' in name.lower():
name = re.split("featuring", name, flags=re.IGNORECASE)[0].strip()
if "&" in name.lower():
if '&' in name.lower():
name = re.split("&", name, flags=re.IGNORECASE)[0].strip()
artist_dict = lookup_artist_from_mb(name)
mbid = mbid or artist_dict.get("id", None)
mbid = mbid or artist_dict['id']
if mbid:
artist = Artist.objects.filter(musicbrainz_id=mbid).first()
logger.debug(f'Looking up artist {name} and mbid: {mbid}')
artist = Artist.objects.filter(musicbrainz_id=mbid).first()
if not artist:
artist = Artist.objects.create(name=name, musicbrainz_id=mbid)
logger.debug(
f"Created artist {artist.name} ({artist.musicbrainz_id}) "
)
artist.fix_metadata()
return artist
def get_or_create_album(
name: str, artist: Artist, mbid: str = None
) -> Optional[Album]:
def get_or_create_album(name: str, artist: Artist, mbid: str = None) -> Album:
album = None
album_dict = lookup_album_dict_from_mb(name, artist_name=artist.name)
album_created = False
albums = Album.objects.filter(name__iexact=name)
if albums.count() == 1:
album = albums.first()
else:
for potential_album in albums:
if artist in album.artist_set.all():
album = potential_album
if not album:
album_created = True
album = Album.objects.create(name=name, musicbrainz_id=mbid)
album.save()
album.artists.add(artist)
name = name or album_dict.get("title", None)
if not name:
logger.debug(
f"Cannot get or create album by {artist} with no name ({name})"
if album_created or not mbid:
album_dict = lookup_album_dict_from_mb(
album.name, artist_name=artist.name
)
return
album = Album.objects.filter(
musicbrainz_id=mbid, name=name, artists__in=[artist]
).first()
if not album:
mbid_group = album_dict.get("mb_group_id")
album = Album.objects.filter(
musicbrainz_releasegroup_id=mbid_group
).first()
if not album and name:
mbid = mbid or album_dict["mb_id"]
album, album_created = Album.objects.get_or_create(musicbrainz_id=mbid)
if album_created:
album.name = name
album.year = album_dict["year"]
album.musicbrainz_releasegroup_id = album_dict["mb_group_id"]
album.musicbrainz_albumartist_id = artist.musicbrainz_id
album.save(
update_fields=[
"name",
"musicbrainz_id",
"year",
"musicbrainz_releasegroup_id",
"musicbrainz_albumartist_id",
]
)
album.artists.add(artist)
album.fix_album_artist()
album.fetch_artwork()
album.scrape_allmusic()
if not album:
logger.warn(f"No album found for {name} and {mbid}")
album.fix_album_artist()
album.year = album_dict["year"]
album.musicbrainz_id = album_dict["mb_id"]
album.musicbrainz_releasegroup_id = album_dict["mb_group_id"]
album.musicbrainz_albumartist_id = artist.musicbrainz_id
album.save(
update_fields=[
"year",
"musicbrainz_id",
"musicbrainz_releasegroup_id",
"musicbrainz_albumartist_id",
]
)
album.artists.add(artist)
album.fetch_artwork()
return album
def get_or_create_track(
title: str,
artist: Artist,
album: Album = None,
album: Album,
mbid: str = None,
run_time_seconds=None,
run_time=None,
run_time_ticks=None,
) -> Track:
track = None
if not mbid and album:
if not mbid:
mbid = lookup_track_from_mb(
title,
artist.musicbrainz_id,
album.musicbrainz_id,
)["id"]
)['id']
track = Track.objects.filter(musicbrainz_id=mbid).first()
@ -112,15 +102,8 @@ def get_or_create_track(
artist=artist,
album=album,
musicbrainz_id=mbid,
run_time_seconds=run_time_seconds,
run_time=run_time,
run_time_ticks=run_time_ticks,
)
return track
def get_or_create_various_artists():
artist = Artist.objects.filter(name="Various Artists").first()
if not artist:
artist = Artist.objects.create(**VARIOUS_ARTIST_DICT)
logger.info("Created Various Artists placeholder")
return artist

View File

@ -1,4 +1,5 @@
from django.db.models import Count
from datetime import timedelta
from django.utils import timezone
from django.views import generic
from music.models import Album, Artist, Track
from scrobbles.models import ChartRecord
@ -17,11 +18,12 @@ class TrackListView(generic.ListView):
class TrackDetailView(generic.DetailView):
model = Track
slug_field = "uuid"
slug_field = 'uuid'
def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)
context_data["charts"] = ChartRecord.objects.filter(
context_data['charts'] = ChartRecord.objects.filter(
track=self.object, rank__in=[1, 2, 3]
)
return context_data
@ -32,39 +34,23 @@ class ArtistListView(generic.ListView):
paginate_by = 100
def get_queryset(self):
return (
super()
.get_queryset()
.annotate(scrobble_count=Count("track__scrobble"))
.order_by("-scrobble_count")
)
return super().get_queryset().order_by("name")
def get_context_data(self, *, object_list=None, **kwargs):
context_data = super().get_context_data(
object_list=object_list, **kwargs
)
context_data["view"] = self.request.GET.get("view")
context_data['view'] = self.request.GET.get('view')
return context_data
class ArtistDetailView(generic.DetailView):
model = Artist
slug_field = "uuid"
slug_field = 'uuid'
def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)
artist = context_data["object"]
rank = 1
tracks_ranked = []
scrobbles = artist.tracks.first().scrobble_count
for track in artist.tracks:
if scrobbles > track.scrobble_count:
rank += 1
tracks_ranked.append((rank, track))
scrobbles = track.scrobble_count
context_data["tracks_ranked"] = tracks_ranked
context_data["charts"] = ChartRecord.objects.filter(
context_data['charts'] = ChartRecord.objects.filter(
artist=self.object, rank__in=[1, 2, 3]
)
return context_data
@ -72,19 +58,22 @@ class ArtistDetailView(generic.DetailView):
class AlbumListView(generic.ListView):
model = Album
paginate_by = 50
def get_queryset(self):
return (
super()
.get_queryset()
.annotate(scrobble_count=Count("track__scrobble"))
.order_by("-scrobble_count")
return super().get_queryset().order_by("name")
def get_context_data(self, *, object_list=None, **kwargs):
context_data = super().get_context_data(
object_list=object_list, **kwargs
)
context_data['view'] = self.request.GET.get('view')
return context_data
class AlbumDetailView(generic.DetailView):
model = Album
slug_field = "uuid"
slug_field = 'uuid'
def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)

View File

@ -27,6 +27,7 @@ class EpisodeAdmin(admin.ModelAdmin):
list_display = (
"title",
"podcast",
"run_time",
)
list_filter = ("podcast",)
ordering = ("-created",)

View File

@ -2,4 +2,4 @@ from django.apps import AppConfig
class PodcastsConfig(AppConfig):
name = "podcasts"
name = 'podcasts'

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:01
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("podcasts", "0005_alter_episode_options_alter_episode_title"),
]
operations = [
migrations.AlterField(
model_name="episode",
name="run_time",
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-12 01:21
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("podcasts", "0006_alter_episode_run_time"),
]
operations = [
migrations.RenameField(
model_name="episode",
old_name="run_time",
new_name="run_time_seconds",
),
]

View File

@ -1,25 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-14 22:27
from django.db import migrations
import taggit.managers
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0033_genre_objectwithgenres"),
("podcasts", "0007_rename_run_time_episode_run_time_seconds"),
]
operations = [
migrations.AddField(
model_name="episode",
name="genre",
field=taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
]

View File

@ -1,20 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-22 22:26
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("podcasts", "0008_episode_genre"),
]
operations = [
migrations.AddField(
model_name="podcast",
name="cover",
field=models.ImageField(
blank=True, null=True, upload_to="pdocasts/covers/"
),
),
]

View File

@ -1,29 +0,0 @@
# Generated by Django 4.1.5 on 2023-03-23 02:57
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("podcasts", "0009_podcast_cover"),
]
operations = [
migrations.RemoveField(
model_name="podcast",
name="cover",
),
migrations.AddField(
model_name="podcast",
name="cover_image",
field=models.ImageField(
blank=True, null=True, upload_to="podcasts/covers/"
),
),
migrations.AddField(
model_name="podcast",
name="description",
field=models.TextField(blank=True, null=True),
),
]

Some files were not shown because too many files have changed in this diff Show More