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
369 changed files with 4256 additions and 23917 deletions

View File

@ -4,17 +4,17 @@
################
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
- pip install poetry
- poetry install --with dev
- poetry install
# Start with a fresh database (which is already running as a service from Drone)
- poetry run pytest --cov-report term:skip-covered --cov=vrobbler tests
environment:
@ -23,46 +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: jail_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:0.3-linux-amd64
when:
status: [success]
settings:
url: https://ntfy.unbl.ink
topic: drone
priority: low
tags:
- failure
- vrobbler
- name: build failure notification
image: parrazam/drone-ntfy:0.3-linux-amd64
when:
status: [failure]
settings:
url: https://ntfy.unbl.ink
topic: drone
priority: high
tags:
- success
- vrobbler
volumes:
- name: docker
host:

2
.gitignore vendored
View File

@ -1,4 +1,4 @@
db.sqlite3*
db.sqlite3
vrobbler.conf
media/
dist/

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.

File diff suppressed because one or more lines are too long

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()

5423
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.15.4"
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,23 +36,8 @@ 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"
stream-sqlite = "^0.0.41"
ipython = "^8.14.0"
pendulum = "^2.1.2"
trafilatura = "^1.6.3"
django-imagekit = "^5.0.0"
thefuzz = "^0.22.1"
dataclass-wizard = "0.22.0"
webdavclient3 = "^3.14.6"
boto3 = "^1.35.14"
[tool.poetry.group.dev]
optional = true
[tool.poetry.group.dev.dependencies]
[tool.poetry.dev-dependencies]
Werkzeug = "2.0.3"
black = "^22.3"
coverage = "^7.0.5"
@ -63,6 +49,7 @@ pytest-django = "^4.5.2"
pytest-flake8 = "^1.1"
pytest-isort = "^3.0"
pytest-runner = "^6.0"
pytest-selenium = "^2.0.1"
time-machine = "^2.9.0"
types-pytz = "^2022.1"
types-requests = "^2.27"
@ -70,12 +57,13 @@ bandit = "^1.7.4"
[tool.pytest.ini_options]
minversion = "6.0"
addopts = "-ra -q --reuse-db"
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,21 +0,0 @@
import pytest
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"
@pytest.mark.skip("Google Podcasts is gone")
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,49 +1,30 @@
import json
import pytest
from django.contrib.auth import get_user_model
from rest_framework.authtoken.models import Token
from scrobbles.models import Scrobble
from boardgames.models import BoardGame
from rest_framework.authtoken.models import Token
from django.contrib.auth import get_user_model
User = get_user_model()
@pytest.fixture
def boardgame_scrobble():
user = User.objects.create(
email="test@exmaple.com", first_name="Test", last_name="User"
)
return Scrobble.objects.create(
board_game=BoardGame.objects.create(title="Test Board Game"),
media_type="BoardGame",
played_to_completion=True,
log={
"players": [
{"user_id": user.id, "win": True, "score": 30, "color": "Blue"}
]
},
)
class MopidyRequest:
name = "Same in the End"
artist = "Sublime"
album = "Sublime"
track_number = 4
run_time_ticks = 156604
run_time = 60
run_time = "156"
playback_time_ticks = 15045
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)),
@ -80,13 +61,13 @@ 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
@pytest.fixture
def mopidy_track():
return MopidyRequest()
def mopidy_track_request_data():
return MopidyRequest().request_json
@pytest.fixture
@ -98,61 +79,6 @@ def mopidy_track_diff_album_request_data(**kwargs):
@pytest.fixture
def mopidy_podcast():
def mopidy_podcast_request_data():
mopidy_uri = "local:podcast:Up%20First/2022-01-01%20Up%20First.mp3"
return MopidyRequest(mopidy_uri=mopidy_uri)
class JellyfinTrackRequest:
name = "Emotion"
artist = "Carly Rae Jepsen"
album = "Emotion"
track_number = 1
item_type = "Audio"
timestamp = "2024-01-14 12:00:19"
run_time_ticks = 156604
run_time = "00:00:60"
playback_time_ticks = 15045
musicbrainz_track_id = "54214d63-5adf-4909-87cd-c65c37a6d558"
musicbrainz_album_id = "03b864cd-7761-314c-a892-05a89ddff00d"
musicbrainz_artist_id = "95f5b748-d370-47fe-85bd-0af2dc450bc0"
status = "resumed"
def __init__(self, **kwargs):
self.request_data = {
"Name": kwargs.get("name", self.name),
"Artist": kwargs.get("artist", self.artist),
"Album": kwargs.get("album", self.album),
"TrackNumber": int(kwargs.get("track_number", self.track_number)),
"RunTime": kwargs.get("run_time", self.run_time),
"ItemType": kwargs.get("item_type", self.item_type),
"UtcTimestamp": kwargs.get("timestamp", self.timestamp),
"PlaybackPositionTicks": int(
kwargs.get("playback_time_ticks", self.playback_time_ticks)
),
"Provider_musicbrainztrack": kwargs.get(
"musicbrainz_track_id", self.musicbrainz_track_id
),
"Provider_musicbrainzalbum": kwargs.get(
"musicbrainz_album_id", self.musicbrainz_album_id
),
"Provider_musicbrainzartist": kwargs.get(
"musicbrainz_artist_id", self.musicbrainz_artist_id
),
"Status": kwargs.get("status", self.status),
}
def __eq__(self, other):
for key in self.request_data.keys():
if self.request_data[key] != getattr(self, key):
return False
return True
@property
def request_json(self):
return json.dumps(self.request_data)
@pytest.fixture
def jellyfin_track():
return JellyfinTrackRequest()
return MopidyRequest(mopidy_uri=mopidy_uri).request_json

View File

@ -6,18 +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_json, num=7, spacing=2):
url = reverse("scrobbles:mopidy-webhook")
user = get_user_model().objects.create(username="Test User")
user.profile.timezone = "US/Eastern"
user.profile.save()
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')
for i in range(num):
client.post(url, request_json, 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)
@ -27,45 +25,78 @@ def build_scrobbles(client, request_json, num=7, spacing=2):
@pytest.mark.django_db
@time_machine.travel(datetime(2022, 3, 4, 1, 24))
def test_scrobble_counts_data(client, mopidy_track):
build_scrobbles(client, mopidy_track.request_json)
def test_scrobble_counts_data(client, mopidy_track_request_data):
build_scrobbles(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):
build_scrobbles(client, mopidy_track.request_json, 7, 1)
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,31 +0,0 @@
import pytest
from scrobbles.dataclasses import BoardGameLogData, BoardGameScoreLogData
@pytest.mark.django_db
def test_boardgame_log_data(boardgame_scrobble):
assert not boardgame_scrobble.geo_location
assert boardgame_scrobble.logdata == BoardGameLogData(
players=[
BoardGameScoreLogData(
user_id=1,
name_str="",
bgg_username="",
color="Blue",
character=None,
team=None,
score=30,
win=True,
new=None,
)
],
location=None,
geo_location_id=None,
difficulty=None,
solo=None,
two_handed=None,
)
assert len(boardgame_scrobble.logdata.players) == 1
assert boardgame_scrobble.logdata.players[0].user.id == 1
assert boardgame_scrobble.logdata.players[0].name == "Test"

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,111 +1,80 @@
from datetime import datetime, timedelta
from unittest.mock import patch
from django.utils import timezone
import json
import pytest
import time_machine
from django.urls import reverse
from music.models import Track
from podcasts.models import PodcastEpisode
from scrobbles.models import Scrobble
from music.models import Track
from podcasts.models import Episode
@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)'
)
@pytest.mark.parametrize(
"seconds, expected_percent_played, expected_scrobble_id",
[
(1, 1, 1),
(58, 96, 1),
(59, 98, 1),
(60, 100, 1),
(1, 1, 2),
(1, 1, 3),
],
)
@pytest.mark.django_db
def test_scrobble_mopidy_track(
client,
mopidy_track,
valid_auth_token,
seconds,
expected_percent_played,
expected_scrobble_id,
client, mopidy_track_request_data, valid_auth_token
):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
# Start new scrobble
minutes = 0
calc_seconds = seconds
if seconds >= 60:
minutes = 1
calc_seconds = calc_seconds % 10
with time_machine.travel(datetime(2024, 1, 14, 12, minutes, calc_seconds)):
mopidy_track.request_data["playback_time_ticks"] = seconds * 1000
response = client.post(
url,
mopidy_track.request_json,
content_type="application/json",
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": expected_scrobble_id}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.percent_played == expected_percent_played
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Same in the End"
@pytest.mark.skip(reason="Allmusic API is unstable")
@pytest.mark.django_db
def test_scrobble_mopidy_same_track_different_album(
client,
mopidy_track,
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",
mopidy_track_request_data,
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.__class__ == Track
assert scrobble.media_obj.title == "Same in the End"
@pytest.mark.django_db
def test_scrobble_mopidy_same_track_different_album(
client,
mopidy_track_request_data,
mopidy_track_diff_album_request_data,
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',
headers=headers,
)
assert response.status_code == 200
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"
@ -115,97 +84,17 @@ 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__ == PodcastEpisode
assert scrobble.media_obj.__class__ == Episode
assert scrobble.media_obj.title == "Up First"
@pytest.mark.django_db
@patch("music.utils.lookup_artist_from_mb", return_value={})
@patch(
"music.utils.lookup_album_dict_from_mb",
return_value={"year": "1999", "mb_group_id": 1},
)
@patch("music.utils.lookup_track_from_mb", return_value={})
@patch("music.models.lookup_artist_from_tadb", return_value={})
@patch("music.models.lookup_album_from_tadb", return_value={"year": "1999"})
@patch("music.models.Album.fetch_artwork", return_value=None)
@patch("music.models.Album.scrape_allmusic", return_value=None)
def test_scrobble_jellyfin_track(
mock_lookup_artist,
mock_lookup_album,
mock_lookup_track,
mock_lookup_artist_tadb,
mock_lookup_album_tadb,
mock_fetch_artwork,
mock_scrape_allmusic,
client,
jellyfin_track,
valid_auth_token,
):
url = reverse("scrobbles:jellyfin-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
with time_machine.travel(datetime(2024, 1, 14, 12, 00, 1)):
jellyfin_track.request_data["UtcTimestamp"] = timezone.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
response = client.post(
url,
jellyfin_track.request_json,
content_type="application/json",
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 1}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Emotion"
with time_machine.travel(datetime(2024, 1, 14, 12, 0, 58)):
jellyfin_track.request_data["UtcTimestamp"] = timezone.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
response = client.post(
url,
jellyfin_track.request_json,
content_type="application/json",
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 1}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Emotion"
with time_machine.travel(datetime(2024, 1, 14, 12, 1, 1)):
jellyfin_track.request_data["UtcTimestamp"] = timezone.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
response = client.post(
url,
jellyfin_track.request_json,
content_type="application/json",
headers=headers,
)
assert response.status_code == 200
assert response.data == {"scrobble_id": 2}
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Emotion"

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,27 +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>"
VROBBLER_COMICVINE_API_KEY="<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,152 +0,0 @@
import csv
import json
import logging
from typing import TYPE_CHECKING, Optional
import requests
from bs4 import BeautifulSoup
from django.contrib.auth import get_user_model
User = get_user_model()
if TYPE_CHECKING:
from scrobbles.models import Scrobble
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
def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]:
bgg_username = user.profile.bgg_username
bgg_password = user.profile.bgg_password
if not bgg_username or bgg_password:
return
login_payload = {
"credentials": {"username": bgg_username, "password": bgg_password}
}
headers = {"content-type": "application/json"}
# TODO Look up past plays for scrobble.media_obj.bggeek_id, and make sure we haven't scrobbled this before
with requests.Session() as s:
p = s.post(
"https://boardgamegeek.com/login/api/v1",
data=json.dumps(login_payload),
headers=headers,
)
players = []
if scrobble.metadata:
for player in scrobble.metadata.players:
if player["user_id"]:
player_user = User.objects.filter(
id=player["user_id"]
).first()
if player_user:
if player_user.bgg_username:
player["username"] = player_user.bgg_username
else:
player["name"] = player_user.username
player["win"] = player.get("win")
player["color"] = player.get("color")
player["new"] = player.get("new")
player["score"] = player.get("score")
players.append(player)
play_payload = {
"playdate": scrobble.timestamp.date.strftime("%Y-%m-%d"),
"length": scrobble.playback_position_seconds / 60,
"comments": "Uploaded from Vrobbler",
"location": scrobble.log.location or None,
"objectid": scrobble.media_obj.bggeek_id,
"quantity": "1",
"action": "save",
"players": players,
"objecttype": "thing",
"ajax": 1,
}

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,170 +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.dataclasses import BoardGameLogData
from scrobbles.mixins import ScrobblableMixin
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}
)
@property
def logdata_cls(self):
return BoardGameLogData
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 = cls.objects.filter(bggeek_id=lookup_id).first()
if not data or not boardgame:
data = lookup_boardgame_from_bgg(lookup_id)
if data and not boardgame:
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,23 +0,0 @@
from django.urls import path
from boardgames import views
app_name = "boardgames"
urlpatterns = [
path(
"board-games/",
views.BoardGameListView.as_view(),
name="boardgame_list",
),
path(
"board-games/<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,142 +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")
# TODO Fix this scraper
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,233 +0,0 @@
"""
ComicVine API Information & Documentation:
https://comicvine.gamespot.com/api/
https://comicvine.gamespot.com/api/documentation
"""
import json
import logging
from django.conf import settings
import requests
logger = logging.getLogger(__name__)
class ComicVineClient(object):
"""
Interacts with the ``search`` resource of the ComicVine API. Requires an
account on https://comicvine.gamespot.com/ in order to obtain an API key.
"""
# All API requests made by this client will be made to this URL.
API_URL = "https://www.comicvine.com/api/search/"
# A valid User-Agent header must be set in order for our API requests to
# be accepted, otherwise our request will be rejected with a
# **403 - Forbidden** error.
HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:7.0) "
"Gecko/20130825 Firefox/36.0"
}
# A set of valid resource types to return in results.
RESOURCE_TYPES = {
"character",
"issue",
"location",
"object",
"person",
"publisher",
"story_arc",
"team",
"volume",
}
def __init__(self, api_key, expire_after=300):
"""
Store the API key in a class variable, and install the requests cache,
configuring it using the ``expire_after`` parameter.
:param api_key: Your personal ComicVine API key.
:type api_key: str
:param expire_after: The number of seconds to retain an entry in cache.
:type expire_after: int or None
"""
self.api_key = api_key
def search(self, query, offset=0, limit=10, resources=None):
"""
Perform a search against the API, using the provided query term. If
required, a list of resource types to filter search results to can
be included.
Take the JSON contained in the response and provide it to the custom
``Response`` object's constructor. Return the ``Response`` object.
:param query: The search query with which to make the request.
:type query: str
:param offset: The index of the first record returned.
:type offset: int or None
:param limit: How many records to return **(max 10)**
:type limit: int or None
:param resources: A list of resources to include in the search results.
:type resources: list or None
:type use_cache: bool
:return: The response object containing the results of the search
query.
:rtype: comicvine_search.response.Response
"""
params = self._request_params(query, offset, limit, resources)
json_data = self._query_api(params)
return json_data
def _request_params(self, query, offset, limit, resources):
"""
Construct a dict containing the required key-value pairs of parameters
required in order to make the API request.
The documentation for the ``search`` resource can be found at
https://comicvine.gamespot.com/api/documentation#toc-0-30.
Regarding 'limit', as per the documentation:
The number of results to display per page. This value defaults to
10 and can not exceed this number.
:param query: The search query with which to make the request.
:type query: str
:param offset: The index of the first record returned.
:type offset: int
:param limit: How many records to return **(max 10)**
:type limit: int
:param resources: A list of resources to include in the search results.
:type resources: list or None
:return: A dictionary of request parameters.
:rtype: dict
"""
return {
"api_key": self.api_key,
"format": "json",
"limit": min(10, limit), # hard limit of 10
"offset": max(0, offset), # cannot provide negative offset
"query": query,
"resources": self._validate_resources(resources),
}
def _validate_resources(self, resources):
"""
Provided a list of resources, first convert it to a set and perform an
intersection with the set of valid resource types, ``RESOURCE_TYPES``.
Return a comma-separted string of the remaining valid resources, or
None if the set is empty.
:param resources: A list of resources to include in the search results.
:type resources: list or None
:return: A comma-separated string of valid resources.
:rtype: str or None
"""
if not resources:
return None
valid_resources = self.RESOURCE_TYPES & set(resources)
return ",".join(valid_resources) if valid_resources else None
def _query_api(self, params):
"""
Query the ComicVine API's ``search`` resource, providing the required
headers and parameters with the request. Optionally allow the caller
of the function to disable the request cache.
If an error occurs during the request, handle it accordingly. Upon
success, return the JSON from the response.
:param params: Parameters to include with the request.
:type params: dict
:param use_cache: Toggle the use of requests_cache.
:type use_cache: bool
:return: The JSON contained in the response.
:rtype: dict
"""
# Since we're performing the identical action regardless of whether
# or not the request cache is to be used, store the procedure in a
# local function to avoid repetition.
def __httpget():
response = requests.get(
self.API_URL, headers=self.HEADERS, params=params
)
if not response.ok:
self._handle_http_error(response)
return response.json()
return __httpget()
def _handle_http_error(self, response):
"""
Provided a ``requests.Response`` object, if the status code is
anything other than **200**, we will treat it as an error.
Using the response's status code, determine which type of exception to
raise. Construct an exception message from the response's status code
and reason properties before raising the exception.
:param response: The requests.Response object returned by the HTTP
request.
:type response: requests.Response
:raises ComicVineUnauthorizedException: if no API key provided.
:raises ComicVineForbiddenException: if no User-Agent header provided.
:raises ComicVineApiException: if an unidentified error occurs.
"""
exception = {
401: Exception,
403: Exception,
}.get(response.status_code, Exception)
message = f"{response.status_code} {response.reason}"
raise exception(message)
def lookup_comic_from_comicvine(title: str) -> dict:
api_key = getattr(settings, "COMICVINE_API_KEY", "")
if not api_key:
logger.warn("No ComicVine API key configured, not looking anything up")
return {}
client = ComicVineClient(
api_key=getattr(settings, "COMICVINE_API_KEY", None)
)
result = [
r
for r in client.search(title).get("results")
if r.get("resource_type") == "volume"
][0]
if "volume" not in result.keys():
logger.warn("No result found on ComicVine", extra={"title": title})
return {}
title = " ".join([result.get("volume").get("name"), result.get("name)")])
data_dict = {
"title": title,
"cover_url": result.get("image").get("original_url"),
"comicvine_data": {
"id": result.get("id"),
"site_detail_url": result.get("site_detail_url"),
"description": result.get("description"),
"image": result.get("image").get("original_url"),
},
}
return data_dict

View File

@ -1,405 +0,0 @@
from collections import OrderedDict
import logging
import re
import sqlite3
from datetime import datetime, timedelta
from enum import Enum
import pytz
import requests
from books.models import Author, Book
from books.openlibrary import get_author_openlibrary_id
from django.apps import apps
from django.contrib.auth import get_user_model
from stream_sqlite import stream_sqlite
logger = logging.getLogger(__name__)
User = get_user_model()
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 requests.get(sqlite_url, stream=True) as r:
yield from r.iter_content(chunk_size=65_536)
# Grace period between page reads for it to be a new scrobble
SESSION_GAP_SECONDS = 1800 # a half hour
def get_author_str_from_row(row):
"""Given a the raw author string from KoReader, convert it to a single line and
strip the middle initials, as OpenLibrary lookup usually fails with those.
"""
ko_authors = row[KoReaderBookColumn.AUTHORS.value].replace("\n", ", ")
# Strip middle initials, OpenLibrary often fails with these
return re.sub(" [A-Z]. ", " ", ko_authors)
def lookup_or_create_authors_from_author_str(ko_author_str: str) -> list:
"""Takes a string of authors from KoReader and returns a list
of Authors from our database
"""
author_str_list = ko_author_str.split(", ")
author_list = []
for author_str in author_str_list:
logger.debug(f"Looking up author {author_str}")
# KoReader gave us nothing, bail
if author_str == "N/A":
logger.warn(f"KoReader author string is N/A, no authors to find")
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}")
author_list.append(author)
return author_list
def create_book_from_row(row: list):
# No KoReader book yet, create it
author_str = get_author_str_from_row(row)
total_pages = row[KoReaderBookColumn.PAGES.value]
run_time = total_pages * Book.AVG_PAGE_READING_SECONDS
book = Book.objects.create(
title=row[KoReaderBookColumn.TITLE.value],
pages=total_pages,
koreader_data_by_hash={
str(row[KoReaderBookColumn.MD5.value]): {
"title": row[KoReaderBookColumn.TITLE.value],
"author_str": author_str,
"book_id": row[KoReaderBookColumn.ID.value],
"pages": total_pages,
}
},
run_time_seconds=run_time,
)
book.fix_metadata()
# Add authors
author_list = lookup_or_create_authors_from_author_str(author_str)
if author_list:
book.authors.add(*author_list)
# self._lookup_authors
return book
def build_book_map(rows) -> 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:
if (
book_row[KoReaderBookColumn.TITLE.value]
== "KOReader Quickstart Guide"
):
logger.info(
"Ignoring the KOReader quickstart guide. No on wants that."
)
continue
book = Book.objects.filter(
koreader_data_by_hash__icontains=book_row[
KoReaderBookColumn.MD5.value
]
).first()
if not book:
book = create_book_from_row(book_row)
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": book.id,
"hash": book_row[KoReaderBookColumn.MD5.value],
"total_seconds": total_seconds,
}
return book_id_map
def build_page_data(page_rows: list, book_map: dict, user_tz=None) -> dict:
"""Given rows of page data from KoReader, parse each row and build
scrobbles for our user, loading the page data into the page_data
field on the scrobble instance.
"""
book_ids_not_found = []
for page_row in page_rows:
koreader_book_id = page_row[KoReaderPageStatColumn.ID_BOOK.value]
if koreader_book_id not in book_map.keys():
book_ids_not_found.append(koreader_book_id)
continue
if "pages" not in book_map[koreader_book_id].keys():
book_map[koreader_book_id]["pages"] = {}
page_number = page_row[KoReaderPageStatColumn.PAGE.value]
duration = page_row[KoReaderPageStatColumn.DURATION.value]
start_ts = page_row[KoReaderPageStatColumn.START_TIME.value]
book_map[koreader_book_id]["pages"][page_number] = {
"duration": duration,
"start_ts": start_ts,
"end_ts": start_ts + duration,
}
if book_ids_not_found:
logger.info(
f"Found pages for books not in file: {set(book_ids_not_found)}"
)
return book_map
def build_scrobbles_from_book_map(
book_map: dict, user: "User"
) -> list["Scrobble"]:
Scrobble = apps.get_model("scrobbles", "Scrobble")
scrobbles_to_create = []
pages_not_found = []
for koreader_book_id, book_dict in book_map.items():
book_id = book_dict["book_id"]
if "pages" not in book_dict.keys():
pages_not_found.append(book_id)
continue
should_create_scrobble = False
scrobble_page_data = {}
playback_position_seconds = 0
prev_page_stats = {}
last_page_number = 0
pages_processed = 0
total_pages_read = len(book_map[koreader_book_id]["pages"])
ordered_pages = sorted(
book_map[koreader_book_id]["pages"].items(),
key=lambda x: x[1]["start_ts"],
)
for cur_page_number, stats in ordered_pages:
pages_processed += 1
seconds_from_last_page = 0
if prev_page_stats:
seconds_from_last_page = stats.get(
"end_ts"
) - prev_page_stats.get("start_ts")
playback_position_seconds = playback_position_seconds + stats.get(
"duration"
)
end_of_reading = pages_processed == total_pages_read
big_jump_to_this_page = (cur_page_number - last_page_number) > 10
is_session_gap = seconds_from_last_page > SESSION_GAP_SECONDS
if (
is_session_gap and not big_jump_to_this_page
) or end_of_reading:
should_create_scrobble = True
if should_create_scrobble:
scrobble_page_data = dict(
sorted(
scrobble_page_data.items(),
key=lambda x: x[1]["start_ts"],
)
)
try:
first_page = scrobble_page_data.get(
list(scrobble_page_data.keys())[0]
)
last_page = scrobble_page_data.get(
list(scrobble_page_data.keys())[-1]
)
except IndexError:
logger.error(
"Could not process book, no page data found",
extra={"scrobble_page_data": scrobble_page_data},
)
continue
timezone = user.profile.timezone
timestamp = datetime.fromtimestamp(
int(first_page.get("start_ts"))
).replace(tzinfo=pytz.timezone(timezone))
# Add a shim here temporarily to fix imports while we were in France
# if date is between 10/15 and 12/15, cast it to Europe/Central
if (
datetime(2023, 10, 15).replace(
tzinfo=pytz.timezone("Europe/Paris")
)
<= timestamp
<= datetime(2023, 12, 15).replace(
tzinfo=pytz.timezone("Europe/Paris")
)
):
timezone = "Europe/Paris"
stop_timestamp = datetime.fromtimestamp(
int(last_page.get("end_ts"))
).replace(tzinfo=pytz.timezone(timezone))
if (
timestamp.tzinfo._dst.seconds == 0
or stop_timestamp.tzinfo._dst.seconds == 0
):
timestamp = timestamp - timedelta(hours=1)
stop_timestamp = stop_timestamp - timedelta(hours=1)
scrobble = Scrobble.objects.filter(
timestamp=timestamp,
book_id=book_id,
user_id=user.id,
).first()
if not scrobble:
logger.info(
f"Queueing scrobble for {book_id}, page {cur_page_number}"
)
log_data = {
"koreader_hash": book_dict.get("hash"),
"page_data": scrobble_page_data,
"pages_read": len(scrobble_page_data.keys()),
}
scrobbles_to_create.append(
Scrobble(
book_id=book_id,
user_id=user.id,
source="KOReader",
media_type=Scrobble.MediaType.BOOK,
timestamp=timestamp,
log=log_data,
stop_timestamp=stop_timestamp,
playback_position_seconds=playback_position_seconds,
in_progress=False,
played_to_completion=True,
long_play_complete=False,
timezone=timezone,
)
)
# Then start over
should_create_scrobble = False
playback_position_seconds = 0
scrobble_page_data = {}
# We accumulate pages for the scrobble until we should create a new one
scrobble_page_data[cur_page_number] = stats
last_page_number = cur_page_number
prev_page_stats = stats
if pages_not_found:
logger.info(f"Pages not found for books: {set(pages_not_found)}")
return scrobbles_to_create
def fix_long_play_stats_for_scrobbles(scrobbles: list) -> None:
"""Given a list of scrobbles, update pages read, long play seconds and check
for media completion"""
for scrobble in scrobbles:
# But if there's a next scrobble, set pages read to their starting page
if scrobble.previous and not scrobble.previous.long_play_complete:
scrobble.long_play_seconds = scrobble.playback_position_seconds + (
scrobble.previous.long_play_seconds or 0
)
else:
scrobble.long_play_seconds = scrobble.playback_position_seconds
scrobble.log["book_pages_read"] = scrobble.calc_pages_read()
scrobble.save(update_fields=["log", "long_play_seconds"])
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"""
Scrobble = apps.get_model("scrobbles", "Scrobble")
new_scrobbles = []
user = User.objects.filter(id=user_id).first()
tz = pytz.utc
if user:
tz = user.profile.timezone
is_os_file = "https://" not in file_path
if is_os_file:
# Loading sqlite file from local filesystem
con = sqlite3.connect(file_path)
cur = con.cursor()
try:
book_map = build_book_map(cur.execute("SELECT * FROM book"))
except sqlite3.OperationalError:
logger.warning("KOReader sqlite file had not table: book")
return new_scrobbles
book_map = build_page_data(
cur.execute(
"SELECT * from page_stat_data ORDER BY id_book, start_time"
),
book_map,
tz,
)
new_scrobbles = build_scrobbles_from_book_map(book_map, user)
else:
# Streaming the sqlite file off S3
book_map = {}
for table_name, pragma_table_info, rows in stream_sqlite(
_sqlite_bytes(file_path), max_buffer_size=1_048_576
):
logger.debug(f"Found table {table_name} - processing")
if table_name == "book":
book_map = build_book_map(rows)
for table_name, pragma_table_info, rows in stream_sqlite(
_sqlite_bytes(file_path), max_buffer_size=1_048_576
):
if table_name == "page_stat_data":
book_map = build_page_data(rows, book_map, tz)
new_scrobbles = build_scrobbles_from_book_map(book_map, user)
logger.info(f"Creating {len(new_scrobbles)} new scrobbles")
created = []
if new_scrobbles:
created = Scrobble.objects.bulk_create(new_scrobbles)
fix_long_play_stats_for_scrobbles(created)
logger.info(
f"Created {len(created)} scrobbles",
extra={"created_scrobbles": created},
)
return created

View File

@ -1,123 +0,0 @@
#!/usr/bin/env python3
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,42 +0,0 @@
#!/usr/bin/env python3
import logging
import pytz
from datetime import datetime, timedelta
from books.models import Book
from django.core.management.base import BaseCommand
from scrobbles.models import Scrobble
from vrobbler.apps.books.koreader import fix_long_play_stats_for_scrobbles
from vrobbler.apps.scrobbles.utils import timestamp_user_tz_to_utc
logger = logging.getLogger(__name__)
# Grace period between page reads for it to be a new scrobble
SESSION_GAP_SECONDS = 1800 # a half hour
def update_scrobble_from_page_data(scrobble, commit=True):
page_list = list(scrobble.book_page_data.items())
first_page_start_ts = datetime.fromtimestamp(page_list[0][1]["start_ts"])
last_page_end_ts = datetime.fromtimestamp(page_list[-1][1]["end_ts"])
if (
datetime(2023, 10, 15) <= first_page_start_ts <= datetime(2023, 12, 15)
):
first_page_start_ts.replace(tzinfo=pytz.timezone("Europe/Paris"))
last_page_end_ts.replace(tzinfo=pytz.timezone("Europe/Paris"))
else:
first_page_start_ts.replace(tzinfo=pytz.timezone("US/Eastern"))
last_page_end_ts.replace(tzinfo=pytz.timezone("US/Eastern"))
scrobble.timestamp = first_page_start_ts
scrobble.stop_timestamp = last_page_end_ts
if commit:
scrobble.save(update_fields=["timestamp", "stop_timestamp"])
class Command(BaseCommand):
def handle(self, *args, **options):
scrobbles_to_create = []
for scrobble in Scrobble.objects.filter(media_type="Book", source="KOReader"):
update_scrobble_from_page_data(scrobble)

View File

@ -1,160 +0,0 @@
import logging
import pytz
from datetime import datetime, timedelta
from books.models import Book
from django.core.management.base import BaseCommand
from scrobbles.models import Scrobble
from vrobbler.apps.books.koreader import fix_long_play_stats_for_scrobbles
from vrobbler.apps.scrobbles.utils import timestamp_user_tz_to_utc
logger = logging.getLogger(__name__)
# Grace period between page reads for it to be a new scrobble
SESSION_GAP_SECONDS = 1800 # a half hour
class Command(BaseCommand):
def handle(self, *args, **options):
scrobbles_to_create = []
for book in Book.objects.filter(koreader_id__isnull=False):
book.scrobble_set.all().delete()
koreader_data = book.koreader_data_by_hash or {}
if book.koreader_md5:
koreader_data[book.koreader_md5] = {
"title": book.title,
"book_id": book.koreader_id,
"author_str": book.koreader_authors,
"pages": book.pages,
}
book.koreader_data_by_hash = koreader_data
book.save(update_fields=["koreader_data_by_hash"])
# Next parse all this book's pages into new scrobbles
should_create_scrobble = False
scrobble_page_data = {}
playback_position_seconds = 0
prev_page = None
pages_processed = 0
total_pages = book.pages
for page in book.page_set.order_by("number"):
user = page.user
book_id = page.book.id
pages_processed += 1
scrobble_page_data[page.number] = {
"duration": page.duration_seconds,
"start_ts": page.start_time.timestamp(),
"end_ts": page.end_time.timestamp(),
}
seconds_from_last_page = 0
if prev_page:
seconds_from_last_page = (
page.end_time.timestamp()
- prev_page.start_time.timestamp()
)
playback_position_seconds = (
playback_position_seconds + page.duration_seconds
)
end_of_reading = pages_processed == total_pages
big_jump_to_this_page = False
if prev_page:
big_jump_to_this_page = (
page.number - prev_page.number
) > 10
if (
seconds_from_last_page > SESSION_GAP_SECONDS
and not big_jump_to_this_page
):
should_create_scrobble = True
if should_create_scrobble:
first_page = scrobble_page_data.get(
list(scrobble_page_data.keys())[0]
)
last_page = scrobble_page_data.get(
list(scrobble_page_data.keys())[-1]
)
start_ts = int(first_page.get("start_ts"))
end_ts = start_ts + playback_position_seconds
timestamp = datetime.fromtimestamp(start_ts).replace(
tzinfo=user.profile.tzinfo
)
stop_timestamp = datetime.fromtimestamp(end_ts).replace(
tzinfo=user.profile.tzinfo
)
# Add a shim here temporarily to fix imports while we were in France
# if date is between 10/15 and 12/15, cast it to Europe/Central
if (
datetime(2023, 10, 15).replace(
tzinfo=pytz.timezone("Europe/Paris")
)
<= timestamp
<= datetime(2023, 12, 15).replace(
tzinfo=pytz.timezone("Europe/Paris")
)
):
timestamp.replace(tzinfo=pytz.timezone("Europe/Paris"))
elif (
timestamp.tzinfo._dst.seconds == 0
or stop_timestamp.tzinfo._dst.seconds == 0
):
timestamp = timestamp - timedelta(hours=1)
stop_timestamp = stop_timestamp - timedelta(hours=1)
scrobble = Scrobble.objects.filter(
timestamp=timestamp,
book_id=book_id,
user_id=user.id,
).first()
if scrobble:
logger.info(
f"Found existing scrobble {scrobble}, updating"
)
scrobble.book_page_data = scrobble_page_data
scrobble.playback_position_seconds = (
scrobble.calc_reading_duration()
)
scrobble.save(
update_fields=[
"book_page_data",
"playback_position_seconds",
]
)
if not scrobble:
logger.info(
f"Queueing scrobble for {book_id}, page {page.number}"
)
scrobbles_to_create.append(
Scrobble(
book_id=book_id,
user_id=user.id,
source="KOReader",
media_type=Scrobble.MediaType.BOOK,
timestamp=timestamp,
stop_timestamp=stop_timestamp,
playback_position_seconds=playback_position_seconds,
book_koreader_hash=list(
book.koreader_data_by_hash.keys()
)[0],
book_page_data=scrobble_page_data,
book_pages_read=page.number,
in_progress=False,
played_to_completion=True,
long_play_complete=False,
)
)
# Then start over
should_create_scrobble = False
playback_position_seconds = 0
scrobble_page_data = {}
prev_page = page
created = Scrobble.objects.bulk_create(scrobbles_to_create)
fix_long_play_stats_for_scrobbles(created)

View File

@ -1,29 +0,0 @@
import logging
import pytz
from datetime import datetime, timedelta
from books.models import Book
from django.core.management.base import BaseCommand
from scrobbles.models import Scrobble
from vrobbler.apps.books.koreader import fix_long_play_stats_for_scrobbles
from vrobbler.apps.scrobbles.utils import timestamp_user_tz_to_utc
logger = logging.getLogger(__name__)
class Command(BaseCommand):
def handle(self, *args, **options):
total_books = Book.objects.all().count()
processed_books = 0
for book in Book.objects.all():
for scrobble in book.scrobble_set.all():
log_data = {
"koreader_hash": scrobble.book_koreader_hash,
"page_data": scrobble.book_page_data,
"pages_read": scrobble.book_pages_read,
}
scrobble.log = log_data
scrobble.save(update_fields=["log"])
processed_books += 1
logger.info(f"Processed book {processed_books} of {total_books}")

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,28 +0,0 @@
# Generated by Django 4.2.9 on 2024-01-29 05:55
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0019_alter_book_authors"),
]
operations = [
migrations.AddField(
model_name="author",
name="comicvine_data",
field=models.JSONField(blank=True, null=True),
),
migrations.AddField(
model_name="book",
name="comicvine_data",
field=models.JSONField(blank=True, null=True),
),
migrations.AddField(
model_name="book",
name="koreader_data_by_hash",
field=models.JSONField(blank=True, null=True),
),
]

View File

@ -1,42 +1,15 @@
from collections import OrderedDict
import logging
from datetime import timedelta, datetime
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.mixins import ScrobblableMixin
from books.utils import lookup_book_from_openlibrary
from scrobbles.utils import get_scrobbles_for_media
from taggit.managers import TaggableManager
from thefuzz import fuzz
from vrobbler.apps.books.comicvine import (
ComicVineClient,
lookup_comic_from_comicvine,
)
from vrobbler.apps.books.locg import (
lookup_comic_by_locg_slug,
lookup_comic_from_locg,
lookup_comic_writer_by_locg_slug,
)
from vrobbler.apps.scrobbles.dataclasses import BookLogData
COMICVINE_API_KEY = getattr(settings, "COMICVINE_API_KEY", "")
logger = logging.getLogger(__name__)
User = get_user_model()
@ -45,269 +18,49 @@ 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)
comicvine_data = models.JSONField(**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)
# All individual koreader fields are deprecated
koreader_id = models.IntegerField(**BNULL)
koreader_authors = models.CharField(max_length=255, **BNULL)
koreader_md5 = models.CharField(max_length=255, **BNULL)
koreader_data_by_hash = models.JSONField(**BNULL)
isbn = models.CharField(max_length=255, **BNULL)
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)
comicvine_data = models.JSONField(**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 logdata_cls(self):
return BookLogData
@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:
logger.warn(f"Checking openlibrary for {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:
if self.locg_slug:
logger.warn(
f"Checking LOCG for {self.title} with slug {self.locg_slug}"
)
data = lookup_comic_by_locg_slug(str(self.locg_slug))
else:
logger.warn(f"Checking LOCG for {self.title}")
data = lookup_comic_from_locg(str(self.title))
if not data and COMICVINE_API_KEY:
logger.warn(f"Checking ComicVine for {self.title}")
cv_client = ComicVineClient(api_key=COMICVINE_API_KEY)
data = lookup_comic_from_comicvine(str(self.title))
if not data:
logger.warn(f"Book not found in any sources: {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", "")
data.pop("ol_author_id", "")
# Kick out a little warning if we're about to change KoReader's title
if (
fuzz.ratio(ol_title.lower(), str(self.title).lower()) < 80
and not force_update
):
logger.warn(
f"OL and KoReader disagree on this book title {self.title} != {ol_title}, check manually"
)
self.openlibrary_id = data.get("openlibrary_id")
self.save(update_fields=["openlibrary_id"])
return
# If we don't know pages, don't overwrite existing with None
if "pages" in data.keys() and data.get("pages") == None:
data.pop("pages")
if (
not isinstance(data.get("pages"), int)
and "pages" in data.keys()
):
logger.info(
f"Pages for {self} from OL expected to be int, but got {data.get('pages')}"
)
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)
def page_data_for_user(
self, user_id: int, convert_timestamps: bool = True
) -> dict:
scrobbles = self.scrobble_set.filter(user=user_id)
pages = {}
for scrobble in scrobbles:
if scrobble.logdata.page_data:
for page, data in scrobble.logdata.page_data.items():
if convert_timestamps:
data["start_ts"] = datetime.fromtimestamp(
data["start_ts"]
)
data["end_ts"] = datetime.fromtimestamp(data["end_ts"])
pages[page] = data
sorted_pages = OrderedDict(
sorted(pages.items(), key=lambda x: x[1]["start_ts"])
)
return sorted_pages
@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:
@ -315,126 +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.last_page_read / self.pages) * 100)
return progress
@classmethod
def find_or_create(cls, lookup_id: str, author: str = "") -> "Book":
book = cls.objects.filter(openlibrary_id=lookup_id).first()
if not book:
data = lookup_book_from_openlibrary(lookup_id, author)
if not data:
logger.error(
f"No book found on openlibrary, or in our database for {lookup_id}"
)
return book
book, book_created = cls.objects.get_or_create(isbn=data["isbn"])
if book_created:
book.fix_metadata(data=data)
return book
class Page(TimeStampedModel):
"""DEPRECATED, we need to migrate pages into page_data on scrobbles and move on"""
book = models.ForeignKey(Book, on_delete=models.CASCADE)
number = models.IntegerField()
user = models.ForeignKey(User, on_delete=models.CASCADE)
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,149 +0,0 @@
import json
import logging
from typing import Optional
import urllib
import requests
from thefuzz import fuzz
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 ""
try:
result = results.get("docs", [])[0]
except IndexError:
result = {"key": ""}
return result.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"):
if fuzz.ratio(title.lower(), result.get("title", "").lower()) > 90:
top = result
break
if not top:
for result in results.get("docs"):
# These Summary things suck and ruin our one-shot search
if "Summary of" in result.get("title"):
continue
if title.lower() in result.get("title", "").lower():
top = result
if not top and len(results.get("docs")) > 0:
top = results.get("docs")[0]
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,123 +0,0 @@
import hashlib
import random
import pytest
from django.contrib.auth import get_user_model
from vrobbler.apps.books.koreader import KoReaderBookColumn
User = get_user_model()
ordinal = lambda n: "%d%s" % (
n,
"tsnrhtdd"[(n // 10 % 10 != 1) * (n % 10 < 4) * n % 10 :: 4],
)
AVERAGE_PAGE_READING_SECONDS = 60
class DummyResponse:
status_code = 200
def status_code(self):
return self.status_code
@pytest.fixture
def valid_response():
return DummyResponse()
class KoReaderBookRows:
id = 1
DEFAULT_STR = "N/A"
DEFAULT_INT = 0
DEFAULT_TIME = 1703800469
BOOK_ROWS = []
PAGE_STATS_ROWS = []
def _gen_random_row(self, i):
wiggle = random.randrange(15)
title = f"Memoirs, Volume {i}"
return [
i,
title,
f"Lord Beaverbrook the {ordinal(i)}",
self.DEFAULT_INT + wiggle / 10,
self.DEFAULT_TIME + i * wiggle,
0,
300 + wiggle,
self.DEFAULT_STR,
self.DEFAULT_STR,
hashlib.md5(title.encode()).hexdigest(),
i * wiggle * 20,
120,
]
def _generate_random_book_rows(self, book_count):
if book_count > 0:
for i in range(1, book_count + 1):
self.BOOK_ROWS.append(self._gen_random_row(i))
def _generate_custom_book_row(self, **kwargs):
title = kwargs.get("title", self.DEFAULT_STR)
if title and title != "N/A":
self.BOOK_ROWS.append(
[
kwargs.get("id", self.id),
kwargs.get("title", self.DEFAULT_STR),
kwargs.get("authors", self.DEFAULT_STR),
kwargs.get("notes", self.DEFAULT_INT),
kwargs.get("last_open", self.DEFAULT_TIME),
kwargs.get("highlights", self.DEFAULT_INT),
kwargs.get("pages", self.DEFAULT_INT),
kwargs.get("series", self.DEFAULT_STR),
kwargs.get("language", self.DEFAULT_STR),
hashlib.md5(title.encode()),
kwargs.get("total_read_time", self.DEFAULT_INT),
kwargs.get("total_read_pages", self.DEFAULT_INT),
]
)
def _generate_random_page_stats_rows(self):
for book in self.BOOK_ROWS:
pages = book[KoReaderBookColumn.PAGES.value]
pages_per_session = 20
start_time = book[KoReaderBookColumn.LAST_OPEN.value]
end_session = False
for page_num in range(
1, book[KoReaderBookColumn.TOTAL_READ_PAGES.value] + 1
):
wiggle = random.randrange(5)
self.PAGE_STATS_ROWS.append(
[
book[KoReaderBookColumn.ID.value],
page_num,
start_time,
AVERAGE_PAGE_READING_SECONDS + wiggle,
pages,
]
)
if end_session:
start_time += 3600 # one second over an hour, marking a new reading session
end_session = False
else:
start_time += AVERAGE_PAGE_READING_SECONDS
if page_num % pages_per_session == 0:
end_session = True
def __init__(self, book_count=0, **kwargs):
self._generate_random_book_rows(book_count)
self._generate_custom_book_row(**kwargs)
self._generate_random_page_stats_rows()
@pytest.fixture
def koreader_rows():
return KoReaderBookRows(book_count=1)
@pytest.fixture
def demo_user():
return User.objects.create(email="demo@example.com")

View File

@ -1,54 +0,0 @@
import pytest
from unittest import mock
from books.koreader import (
KoReaderBookColumn,
build_book_map,
build_page_data,
build_scrobbles_from_book_map,
)
@pytest.mark.django_db
@mock.patch("requests.get")
def test_build_book_map(get_mock, koreader_rows, valid_response):
get_mock.return_value = valid_response
book_map = build_book_map(koreader_rows.BOOK_ROWS)
assert len(book_map) == 1
@pytest.mark.django_db
@mock.patch("requests.get")
def test_load_page_data_to_map(get_mock, koreader_rows, valid_response):
get_mock.return_value = valid_response
book_map = build_page_data(
koreader_rows.PAGE_STATS_ROWS,
build_book_map(koreader_rows.BOOK_ROWS),
)
assert (
len(book_map[1]["pages"])
== koreader_rows.BOOK_ROWS[0][
KoReaderBookColumn.TOTAL_READ_PAGES.value
]
)
@pytest.mark.django_db
@mock.patch("requests.get")
def test_build_scrobbles_from_pages(
get_mock, koreader_rows, demo_user, valid_response
):
get_mock.return_value = valid_response
book_map = build_book_map(koreader_rows.BOOK_ROWS)
book_map = build_page_data(koreader_rows.PAGE_STATS_ROWS, book_map)
scrobbles = build_scrobbles_from_book_map(book_map, demo_user)
# Corresponds to number of sessions per book ( 20 pages per session, 120 +/- 15 pages read )
expected_scrobbles = 6 * len(book_map.keys())
assert len(scrobbles) == expected_scrobbles
assert len(scrobbles[0].logdata.page_data.keys()) == 21
assert len(scrobbles[1].logdata.page_data.keys()) == 20
assert len(scrobbles[2].logdata.page_data.keys()) == 20
assert len(scrobbles[3].logdata.page_data.keys()) == 20
assert len(scrobbles[4].logdata.page_data.keys()) == 20
assert len(scrobbles[5].logdata.page_data.keys()) == 18

View File

@ -1,36 +0,0 @@
from unittest import skip
import pytest
from books.openlibrary import lookup_book_from_openlibrary
def test_lookup_modern_book():
book = lookup_book_from_openlibrary("Matrix", "Lauren Groff")
assert book.get("title") == "Matrix"
assert book.get("openlibrary_id") == "OL32170218M"
assert book.get("ol_author_id") == "OL3675729A"
def test_lookup_classic_book():
book = lookup_book_from_openlibrary(
"The Life of Castruccio Castracani", "Machiavelli"
)
assert book.get("title") == "The Life of Castruccio Castracani of Lucca"
assert book.get("openlibrary_id") == "OL8950869M"
assert book.get("ol_author_id") == "OL23135A"
def test_lookup_foreign_book():
book = lookup_book_from_openlibrary("Ravagé", "René Barjavel")
assert book.get("title") == "Ravage"
assert book.get("openlibrary_id") == "OL8837839M"
assert book.get("ol_author_id") == "OL152472A"
@skip("This is rotten in OL, updated but waiting for it to update")
def test_lookup_book():
book = lookup_book_from_openlibrary("Hark! A Vagrant")
assert book.get("title") == "Hark! A Vagrant"
assert book.get("openlibrary_id") == "OL8837839M"
assert book.get("ol_author_id") == "OL152472A"

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
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class BookListView(ScrobbleableListView):
model = Book
class BookDetailView(ScrobbleableDetailView):
model = Book
class AuthorDetailView(generic.DetailView):
model = Author
slug_field = "uuid"

View File

@ -1,23 +0,0 @@
from django.contrib import admin
from bricksets.models import BrickSet
from scrobbles.admin import ScrobbleInline
class BrickSetInline(admin.TabularInline):
model = BrickSet
extra = 0
@admin.register(BrickSet)
class BrickSetAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"uuid",
"title",
)
ordering = ("-created",)
search_fields = ("title",)
inlines = [
ScrobbleInline,
]

View File

@ -1,5 +0,0 @@
from django.apps import AppConfig
class BrickSetsConfig(AppConfig):
name = "bricksets"

View File

@ -1,106 +0,0 @@
# Generated by Django 4.2.15 on 2024-09-07 05:38
from django.db import migrations, models
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
("scrobbles", "0059_remove_scrobble_book_koreader_hash_and_more"),
]
operations = [
migrations.CreateModel(
name="BrickSet",
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"
),
),
(
"uuid",
models.UUIDField(
blank=True,
default=uuid.uuid4,
editable=False,
null=True,
),
),
(
"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),
),
(
"number",
models.CharField(blank=True, max_length=10, null=True),
),
("release_year", models.IntegerField(blank=True, null=True)),
("piece_count", models.IntegerField(blank=True, null=True)),
(
"brickset_rating",
models.DecimalField(
blank=True, decimal_places=1, max_digits=3, null=True
),
),
(
"lego_item_number",
models.CharField(blank=True, max_length=10, null=True),
),
(
"box_image",
models.ImageField(
blank=True, null=True, upload_to="brickset/boxes/"
),
),
(
"set_image",
models.ImageField(
blank=True, null=True, upload_to="brickset/sets/"
),
),
(
"genre",
taggit.managers.TaggableManager(
blank=True,
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
],
options={
"abstract": False,
},
),
]

View File

@ -1,63 +0,0 @@
from django.apps import apps
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.dataclasses import BrickSetLogData
from scrobbles.mixins import LongPlayScrobblableMixin
BNULL = {"blank": True, "null": True}
class BrickSet(LongPlayScrobblableMixin):
""""""
number = models.CharField(max_length=10, **BNULL)
release_year = models.IntegerField(**BNULL)
piece_count = models.IntegerField(**BNULL)
brickset_rating = models.DecimalField(max_digits=3, decimal_places=1, **BNULL)
lego_item_number = models.CharField(max_length=10, **BNULL)
box_image = models.ImageField(upload_to="brickset/boxes/", **BNULL)
box_image_small = ImageSpecField(
source="box_image",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
box_image_medium = ImageSpecField(
source="box_image",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
set_image = models.ImageField(upload_to="brickset/sets/", **BNULL)
set_image_small = ImageSpecField(
source="set_image",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
set_image_medium = ImageSpecField(
source="set_image",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
def get_absolute_url(self):
return reverse("bricksets:brickset_detail", kwargs={"slug": self.uuid})
@property
def logdata_cls(self):
return BrickSetLogData
@classmethod
def find_or_create(cls, title: str) -> "BrickSet":
return cls.objects.filter(title=title).first()
@property
def primary_image_url(self) -> str:
if self.box_image:
return self.box_image.url
return ""

View File

@ -1,14 +0,0 @@
from django.urls import path
from bricksets import views
app_name = "bricksets"
urlpatterns = [
path("bricksets/", views.BrickSetListView.as_view(), name="brickset_list"),
path(
"bricksets/<slug:slug>/",
views.BrickSetDetailView.as_view(),
name="brickset_detail",
),
]

View File

@ -1,10 +0,0 @@
from bricksets.models import BrickSet
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class BrickSetListView(ScrobbleableListView):
model = BrickSet
class BrickSetDetailView(ScrobbleableDetailView):
model = BrickSet

View File

@ -1,16 +0,0 @@
from django.contrib import admin
from lifeevents.models import LifeEvent
from scrobbles.admin import ScrobbleInline
@admin.register(LifeEvent)
class EventAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = ("title",)
search_fields = ("title",)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]

View File

@ -1,79 +0,0 @@
# Generated by Django 4.2.11 on 2024-05-07 13:37
from django.db import migrations, models
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
("scrobbles", "0055_rename_scrobble_log_scrobble_log"),
]
operations = [
migrations.CreateModel(
name="LifeEvent",
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"
),
),
(
"uuid",
models.UUIDField(
blank=True,
default=uuid.uuid4,
editable=False,
null=True,
),
),
(
"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),
),
("description", models.TextField(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={
"abstract": False,
},
),
]

View File

@ -1,33 +0,0 @@
from django.apps import apps
from django.db import models
from django.urls import reverse
from scrobbles.dataclasses import LifeEventLogData
from scrobbles.mixins import ScrobblableMixin
BNULL = {"blank": True, "null": True}
class LifeEvent(ScrobblableMixin):
description = models.TextField(**BNULL)
def __str__(self):
return self.title
def get_absolute_url(self):
return reverse(
"life-events:life-event_detail", kwargs={"slug": self.uuid}
)
@property
def logdata_cls(self):
return LifeEventLogData
@classmethod
def find_or_create(cls, title: str) -> "LifeEvent":
return cls.objects.filter(title=title).first()
def scrobbles(self, user_id):
Scrobble = apps.get_model("scrobbles", "Scrobble")
return Scrobble.objects.filter(
user_id=user_id, life_event=self
).order_by("-timestamp")

View File

@ -1,16 +0,0 @@
from django.urls import path
from lifeevents import views
app_name = "lifeevents"
urlpatterns = [
path(
"lifeevents/", views.LifeEventListView.as_view(), name="lifeevent_list"
),
path(
"lifeevent/<slug:slug>/",
views.LifeEventDetailView.as_view(),
name="life-event_detail",
),
]

View File

@ -1,12 +0,0 @@
from django.views import generic
from lifeevents.models import LifeEvent
class LifeEventListView(generic.ListView):
model = LifeEvent
paginate_by = 20
class LifeEventDetailView(generic.DetailView):
model = LifeEvent
slug_field = "uuid"

View File

@ -1,21 +0,0 @@
from django.contrib import admin
from locations.models import GeoLocation
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,
]

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,16 +0,0 @@
# Generated by Django 4.2.9 on 2024-02-20 00:20
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("locations", "0005_alter_rawgeolocation_options_and_more"),
]
operations = [
migrations.DeleteModel(
name="RawGeoLocation",
),
]

View File

@ -1,132 +0,0 @@
from decimal import Decimal, getcontext
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()
GEOLOC_ACCURACY = int(getattr(settings, "GEOLOC_ACCURACY", 4))
GEOLOC_PROXIMITY = Decimal(getattr(settings, "GEOLOC_PROXIMITY", "0.0001"))
class GeoLocation(ScrobblableMixin):
COMPLETION_PERCENT = getattr(settings, "LOCATION_COMPLETION_PERCENT", 100)
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:GEOLOC_ACCURACY]
except IndexError:
trunc_lat = r_lat
try:
trunc_lon = r_lon[0:GEOLOC_ACCURACY]
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(".")
data_dict["altitude"] = float(int_alt)
location = cls.objects.filter(
lat=data_dict.get("lat"),
lon=data_dict.get("lon"),
).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
@property
def subtitle(self) -> str:
if self.title:
return f"{self.lat} x {self.lon}"
return ""
def loc_diff(self, old_lat_lon: tuple) -> tuple:
return (
abs(Decimal(old_lat_lon[0]) - Decimal(self.lat)),
abs(Decimal(old_lat_lon[1]) - Decimal(self.lon)),
)
def has_moved(self, previous_location: "GeoLocation") -> bool:
has_moved = False
loc_diff = self.loc_diff(
(previous_location.lat, previous_location.lon)
)
if loc_diff[0] > GEOLOC_PROXIMITY or loc_diff[1] > GEOLOC_PROXIMITY:
has_moved = True
logger.debug(
f"[locations] checked whether location has moved against proximity setting",
extra={
"location_id": self.id,
"loc_diff": loc_diff,
"has_moved": has_moved,
"previous_location_id": previous_location.id,
"geoloc_proximity": GEOLOC_PROXIMITY,
},
)
return has_moved
def in_proximity(self, named=False) -> models.QuerySet:
lat_min = Decimal(self.lat) - GEOLOC_PROXIMITY
lat_max = Decimal(self.lat) + GEOLOC_PROXIMITY
lon_min = Decimal(self.lon) - GEOLOC_PROXIMITY
lon_max = Decimal(self.lon) + GEOLOC_PROXIMITY
is_title_null = not named
close_locations = GeoLocation.objects.filter(
title__isnull=is_title_null,
lat__lte=lat_max,
lat__gte=lat_min,
lon__lte=lon_max,
lon__gte=lon_min,
).exclude(id=self.id)
return close_locations

View File

@ -1,90 +0,0 @@
import pytest
import logging
from locations.models import GeoLocation
logger = logging.getLogger(__name__)
def test_find_or_create(caplog):
assert not GeoLocation.find_or_create({})
assert "No lat or lon keys in data dict" in caplog.text
@pytest.mark.django_db
def test_find_or_create_truncation():
loc = GeoLocation.find_or_create(
{"lat": 44.2345, "lon": -68.2345, "alt": 60.356}
)
assert loc.lat == 44.234
assert loc.lon == -68.234
assert loc.altitude == 60
@pytest.mark.django_db
def test_find_or_create_finds_existing():
extant = GeoLocation.objects.create(lat=44.234, lon=-68.234, altitude=50)
loc = GeoLocation.find_or_create(
{"lat": 44.2345, "lon": -68.2345, "alt": 60.356}
)
assert loc.id == extant.id
@pytest.mark.django_db
def test_find_or_create_creates_new():
extant = GeoLocation.objects.create(lat=44.234, lon=-69.234, altitude=60)
loc = GeoLocation.find_or_create(
{"lat": 44.2345, "lon": -68.2345, "alt": 60.356}
)
assert not loc.id == extant.id
@pytest.mark.django_db
def test_found_in_proximity_location():
lat = 44.234
lon = -69.234
loc = GeoLocation.objects.create(lat=lat, lon=lon, altitude=60)
close = GeoLocation.objects.create(
lat=lat + 0.0001, lon=lon - 0.0001, altitude=60
)
assert close not in loc.in_proximity(named=True)
assert close in loc.in_proximity()
@pytest.mark.django_db
def test_not_found_in_proximity_location():
lat = 44.234
lon = -69.234
loc = GeoLocation.objects.create(lat=lat, lon=lon, altitude=60)
far = GeoLocation.objects.create(
lat=lat + 0.0002, lon=lon - 0.0001, altitude=60
)
assert far not in loc.in_proximity()
@pytest.mark.django_db
def test_has_moved(caplog):
lat = 44.234
lon = -69.234
loc = GeoLocation.objects.create(lat=lat, lon=lon, altitude=60)
past = GeoLocation.objects.get_or_create(
lat=lat + 0.0009, lon=lon - 0.002, altitude=60
)[0]
assert loc.has_moved(past)
@pytest.mark.django_db
def test_has_not_moved():
lat = 44.234
lon = -69.234
loc = GeoLocation.objects.create(lat=lat, lon=lon, altitude=60)
past = GeoLocation.objects.get_or_create(
lat=lat + 0.00009, lon=lon - 0.00009, altitude=60
)[0]
assert not loc.has_moved(past)

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

@ -1,23 +0,0 @@
from django.contrib import admin
from moods.models import Mood
from scrobbles.admin import ScrobbleInline
class MoodInline(admin.TabularInline):
model = Mood
extra = 0
@admin.register(Mood)
class MoodAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"uuid",
"title",
)
ordering = ("-created",)
search_fields = ("title",)
inlines = [
ScrobbleInline,
]

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