Compare commits

..

1 Commits
35 ... 0.12.0

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

View File

@ -4,65 +4,25 @@
################
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 test
- poetry install
# Start with a fresh database (which is already running as a service from Drone)
- poetry run pytest -n 5 --cov-report term:skip-covered --cov=vrobbler tests
- poetry run pytest --cov-report term:skip-covered --cov=vrobbler tests
environment:
VROBBLER_DATABASE_URL: sqlite:///test.db
volumes:
# 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:
ref:
- refs/tags/*
- 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:

View File

@ -1,31 +0,0 @@
name: Django CI
on:
push:
branches: [ "develop" ]
pull_request:
branches: [ "main" ]
jobs:
build:
runs-on: ubuntu-latest
strategy:
max-parallel: 4
matrix:
python-version: [3.9, 3.11, 3.12]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v3
with:
python-version: ${{ matrix.python-version }}
- name: Install Dependencies
run: |
python -m pip install --upgrade pip
python -m pip install poetry
poetry install
- name: Run Tests
run: |
pytest

2
.gitignore vendored
View File

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

View File

@ -1,6 +0,0 @@
deploy:
ssh vrobbler.service "rm -rf /usr/local/lib/python3.11/site-packages/vrobbler-0.15.4.dist-info/ && pip install git+https://code.unbl.ink/secstate/vrobbler.git@develop && immortalctl restart vrobbler && immortalctl restart vrobbler-celery && vrobbler migrate"
logs:
ssh life.unbl.ink tail -n 100 -f /var/log/vrobbler.json
test:
pytest vrobbler

File diff suppressed because it is too large Load Diff

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

@ -1,4 +1,4 @@
export ENV_PATH=$(poetry env info --path)
source "${ENV_PATH}/bin/activate"
#export PYPI_PASSWORD="$(pass personal/apikey/pypi)"
export PYPI_PASSWORD="$(pass personal/apikey/pypi)"

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

8016
poetry.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -1,23 +1,22 @@
[tool.poetry]
name = "vrobbler"
version = "0.16.1"
version = "0.11.5"
description = ""
authors = ["Colin Powell <colin@unbl.ink>"]
[tool.poetry.dependencies]
python = ">=3.11,<3.14"
python = "^3.8"
Django = "^4.0.3"
django-extensions = "^3.1.5"
python-dateutil = "^2.8.2"
python-dotenv = "^0.20.0"
python-json-logger = "^2.0.2"
colorlog = "^6.6.0"
httpx = "<=0.27.2"
djangorestframework = "^3.13.1"
Markdown = "^3.3.6"
django-filter = "^21.1"
Pillow = "^10.0.0"
psycopg2 = "2.9.10"
Pillow = "^9.0.1"
psycopg2 = "^2.9.3"
dj-database-url = "^0.5.0"
django-mathfilters = "^1.0.0"
django-allauth = "^0.50.0"
@ -27,8 +26,9 @@ 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"
cinemagoerng = {git = "https://github.com/cinemagoer/cinemagoerng"}
cinemagoer = "^2022.12.27"
pysportsdb = "^0.1.0"
pytz = "^2022.7.1"
django-redis = "^5.2.0"
@ -36,34 +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 = "^3"
trafilatura = "^1.6.3"
django-imagekit = "^5.0.0"
thefuzz = "^0.22.1"
dataclass-wizard = "^0.35.0"
webdavclient3 = "^3.14.6"
boto3 = "^1.35.37"
urllib3 = "<2"
django-oauth-toolkit = "^3.0.1"
meta-yt = "^0.1.9"
berserk = "^0.13.2"
poetry-bumpversion = "^0.3.3"
orgparse = "^0.4.20250520"
tmdbv3api = "^1.9.0"
themoviedb = "^1.0.2"
feedparser = "^6.0.12"
titlecase = "^2.4.1"
bgg-api = "^1.1.13"
[tool.poetry.group.test]
optional = true
[tool.poetry.group.test.dependencies]
[tool.poetry.dev-dependencies]
Werkzeug = "2.0.3"
black = "^22.3"
coverage = "^7.0.5"
@ -72,10 +46,10 @@ pytest = "^7.1"
pytest-black = "^0.3.12"
pytest-cov = "^3.0"
pytest-django = "^4.5.2"
pytest-xdist= "^1.0.0"
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"
@ -83,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,31 +0,0 @@
import pytest
from boardgames.bgg import (
take_first,
lookup_boardgame_id_from_bgg,
lookup_boardgame_from_bgg,
)
@pytest.mark.skip(reason="Deprecated library")
def test_take_first():
assert take_first([]) == ""
assert take_first(["a", "b"]) == "a"
@pytest.mark.skip(reason="Deprecated library")
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
@pytest.mark.skip(reason="Deprecated library")
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,60 +1,30 @@
import json
import pytest
from django.contrib.auth import get_user_model
from rest_framework.authtoken.models import Token
from boardgames.models import BoardGame
from music.models import Track, Artist
from scrobbles.models import Scrobble
from people.models import Person
from rest_framework.authtoken.models import Token
from django.contrib.auth import get_user_model
User = get_user_model()
@pytest.fixture
def boardgame_scrobble():
first = Person.objects.create(name="First Player")
second = Person.objects.create(name="Second Player")
return Scrobble.objects.create(
board_game=BoardGame.objects.create(title="Test Board Game"),
media_type="BoardGame",
played_to_completion=True,
log={
"players": [
{"person_id": first.id, "win": True, "score": 30, "color": "Blue"},
{"person_id": second.id, "win": False, "score": 28, "color": "Red"}
],
},
)
@pytest.fixture
def test_track():
Track.objects.create(
title="Emotion",
artist=Artist.objects.create(name="Carly Rae Jepsen"),
base_run_time_seconds=60,
)
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)),
@ -91,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
@ -111,67 +81,4 @@ def mopidy_track_diff_album_request_data(**kwargs):
@pytest.fixture
def mopidy_podcast_request_data():
mopidy_uri = "local:podcast:Up%20First/2022-01-01%20Up%20First.mp3"
return MopidyRequest(
mopidy_uri=mopidy_uri, artist="NPR", album="Up First"
).request_json
@pytest.fixture
def mopidy_podcast_https_request_data():
mopidy_uri = "podcast+https://feeds.npr.org/510318/podcast.xml#85b9c4c4-ae09-43d9-8853-31ccf43f68e6"
return MopidyRequest(
mopidy_uri=mopidy_uri, artist="NPR", album="Up First"
).request_json
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

@ -0,0 +1,11 @@
import pytest
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')
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,44 +0,0 @@
import pytest
#from scrobbles.dataclasses import BoardGameLogData, BoardGameScoreLogData
@pytest.mark.skip("Need to get local tests running working again")
@pytest.mark.django_db
def test_boardgame_log_data(boardgame_scrobble):
assert boardgame_scrobble.logdata == BoardGameLogData(
players=[
BoardGameScoreLogData(
person_id=1,
bgg_username="",
color="Blue",
character=None,
team=None,
score=30,
win=True,
new=None,
rank=None,
seat_order=None,
role=None
),
BoardGameScoreLogData(
person_id=2,
bgg_username="",
color="Red",
character=None,
team=None,
score=28,
win=False,
new=None,
rank=None,
seat_order=None,
role=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,252 +1,100 @@
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.skip("Need to refactor")
@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_mopidy_same_track_different_album(
mock_lookup_artist,
mock_lookup_album,
mock_lookup_track,
mock_lookup_artist_tadb,
mock_lookup_album_tadb,
mock_fetch_artwork,
mock_scrape_allmusic,
client,
mopidy_track,
mopidy_track_diff_album_request_data,
valid_auth_token,
def test_scrobble_mopidy_track(
client, mopidy_track_request_data, valid_auth_token
):
url = reverse("scrobbles:mopidy-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
url = reverse('scrobbles:mopidy-webhook')
headers = {'Authorization': f'Token {valid_auth_token}'}
response = client.post(
url,
mopidy_track.request_data,
content_type="application/json",
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 == "Sublime"
assert scrobble.media_obj.album.name == "Gold"
assert scrobble.media_obj.title == "Same in the End"
@pytest.mark.django_db
@patch(
"podcasts.sources.podcastindex.lookup_podcast_from_podcastindex",
return_value={},
)
def test_scrobble_mopidy_podcast(
mock_lookup_podcast, client, mopidy_podcast_request_data, valid_auth_token
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.skip("Need to refactor")
@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"
@pytest.mark.skip("Need to refactor")
@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_update(
mock_lookup_artist,
mock_lookup_album,
mock_lookup_track,
mock_lookup_artist_tadb,
mock_lookup_album_tadb,
mock_fetch_artwork,
mock_scrape_allmusic,
test_track,
client,
jellyfin_track,
valid_auth_token,
):
Scrobble.objects.create(
timestamp=timezone.now() - timedelta(minutes=0.5),
track=Track.objects.first(),
user_id=1,
)
url = reverse("scrobbles:jellyfin-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
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"
@pytest.mark.skip("Need to refactor")
@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_create_new(
mock_lookup_artist,
mock_lookup_album,
mock_lookup_track,
mock_lookup_artist_tadb,
mock_lookup_album_tadb,
mock_fetch_artwork,
mock_scrape_allmusic,
test_track,
client,
jellyfin_track,
valid_auth_token,
):
url = reverse("scrobbles:jellyfin-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
Scrobble.objects.create(
timestamp=timezone.now() - timedelta(minutes=1),
track=Track.objects.first(),
user_id=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"

View File

@ -1,6 +0,0 @@
from videos.sources.imdb import lookup_video_from_imdb
def test_lookup_imdb():
metadata = lookup_video_from_imdb("8946378")
assert metadata.title == "Knives Out"

View File

@ -1,9 +0,0 @@
import pytest
from videos.sources.youtube import lookup_video_from_youtube
@pytest.mark.skip(reason="Need to configure Youtube API stuffs in CI")
@pytest.mark.django_db
def test_lookup_youtube_id():
metadata = lookup_video_from_youtube("RZxs9pAv99Y")
assert metadata.title == "No Pun Included's Board Game of the Year 2024"

382
todos.org Normal file
View File

@ -0,0 +1,382 @@
#+title: TODOs
A fun way to keep track of things in the project to fix or improve.
* 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:
CLOSED: [2023-01-30 Mon 18:31]
:LOGBOOK:
CLOCK: [2023-01-30 Mon 18:00]--[2023-01-30 Mon 18:31] => 0:31
:END:
If we play music from Jellyfin and the track reaches 90% completion, the
scrobbling goes crazy and starts creating new scrobbles with every update.
The cause is pretty simple, but the solution is hard. We want to mark a scrobble
as complete for the following conditions:
- Play stopped and percent played beyond 90%
- Play completely finished
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:
CLOSED: [2023-02-03 Fri 16:52]
An example of the format:
#+begin_src csv
,
#AUDIOSCROBBLER/1.1
#TZ/UNKNOWN
#CLIENT/Rockbox sansaclipplus $Revision$
75 Dollar Bill I Was Real I Was Real 4 1015 S 1740494944 64ff5f53-d187-4512-827e-7606c69e66ff
75 Dollar Bill I Was Real I Was Real 4 1015 S 1740494990 64ff5f53-d187-4512-827e-7606c69e66ff
311 311 Down 1 173 S 1740495003 00476c23-fd9e-464b-9b27-a62d69f3d4f4
311 311 Down 1 173 L 1740495049 00476c23-fd9e-464b-9b27-a62d69f3d4f4
311 311 Down 1 173 L 1740495113 00476c23-fd9e-464b-9b27-a62d69f3d4f4
311 311 Random 2 187 S 1740495190 530c09f3-46fe-4d90-b11f-7b63bcb4b373
311 311 Random 2 187 L 1740495194 530c09f3-46fe-4d90-b11f-7b63bcb4b373
311 311 Jackolanterns Weather 3 204 L 1740495382 cc3b2dec-5d99-47ea-8930-20bf258be4ea
311 311 All Mixed Up 4 182 L 1740495586 980a78b5-5bdd-4f50-9e3a-e13261e2817b
311 311 Hive 5 179 L 1740495768 18f6dc98-d3a2-4f81-b967-97359d14c68c
311 311 Guns (Are for Pussies) 6 137 L 1740495948 5e97ed9f-c8cc-4282-9cbe-f8e17aee5128
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:
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:
CLOSED: [2023-02-17 Fri 00:11]
* 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.
* 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:
Maloja does this cool thing where artists and tracks get recorded as the top
track of a given week, month or year. They get gold, silver or bronze stars for
their place in the time period.
I could see this being implemented as a separate Chart table which gets
populated at the end of a time period and has a start and end date that defines
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.
* 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",
"event": "track_playback_ended",
"data": {
"tl_track": {
"__model__": "TlTrack",
"tlid": 13,
"track": {
"__model__": "Track",
"uri": "file:///var/lib/mopidy/media/podcasts/The%20Prince/2022-09-28-Wolf-warriors.mp3",
"name": "Wolf warriors",
"artists": [
{
"__model__": "Artist",
"name": "The Economist"
}
],
"album": {
"__model__": "Album",
"name": "The Prince",
"date": "2022"
},
"genre": "Blues",
"date": "2022",
"length": 2437778,
"bitrate": 127988
}
},
"time_position": 3290
}
}
#+end_src
*** Podcast playback state changes
#+begin_src json
{
"type": "event",
"event": "playback_state_changed",
"data": {
"old_state": "paused",
"new_state": "playing"
}
}
#+end_src
#+begin_src json
{
"type": "event",
"event": "playback_state_changed",
"data": {
"old_state": "stopped",
"new_state": "playing"
}
}
#+end_src
*** Podcast playback started
#+begin_src json
{
"type": "event",
"event": "track_playback_started",
"data": {
"tl_track": {
"__model__": "TlTrack",
"tlid": 13,
"track": {
"__model__": "Track",
"uri": "file:///var/lib/mopidy/media/podcasts/The%20Prince/2022-09-28-Wolf-warriors.mp3",
"name": "Wolf warriors",
"artists": [
{
"__model__": "Artist",
"name": "The Economist"
}
],
"album": {
"__model__": "Album",
"name": "The Prince",
"date": "2022"
},
"genre": "Blues",
"date": "2022",
"length": 2437778,
"bitrate": 127988
}
}
}
}
#+end_src
*** Podcast playback paused
#+begin_src json
{
"type": "status",
"data": {
"state": "paused",
"current_track": {
"__model__": "Track",
"uri": "file:///var/lib/mopidy/media/podcasts/The%20Prince/2022-09-28-Wolf-warriors.mp3",
"name": "Wolf warriors",
"artists": [
{
"__model__": "Artist",
"name": "The Economist"
}
],
"album": {
"__model__": "Album",
"name": "The Prince",
"date": "2022"
},
"genre": "Blues",
"date": "2022",
"length": 2437778,
"bitrate": 127988
},
"time_position": 2350
}
}
#+end_src
*** Track playback started
#+begin_src json
{
"type": "event",
"event": "track_playback_started",
"data": {
"tl_track": {
"__model__": "TlTrack",
"tlid": 14,
"track": {
"__model__": "Track",
"uri": "local:track:Various%20Artists%20-%202008%20-%20Twilight%20OST/01-muse-supermassive_black_hole.mp3",
"name": "Supermassive Black Hole",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:250dd6551b66a58a6b4897aa697f200c",
"name": "Muse",
"musicbrainz_id": "9c9f1380-2516-4fc9-a3e6-f9f61941d090"
}
],
"album": {
"__model__": "Album",
"uri": "local:album:md5:455343d54cdd89cb5a3b5ad537ea99d0",
"name": "Twilight: Original Motion Picture Soundtrack",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:54e4db2d5624f80b0cc290346e696756",
"name": "Various Artists",
"musicbrainz_id": "89ad4ac3-39f7-470e-963a-56509c546377"
}
],
"num_tracks": 12,
"num_discs": 1,
"date": "2008-11-04",
"musicbrainz_id": "b4889eaf-d9f4-434c-a68d-69227b12b6a4"
},
"composers": [
{
"__model__": "Artist",
"uri": "local:artist:md5:4d49cbca0b347e0a89047bb019d2779d",
"name": "Matt Bellamy"
}
],
"genre": "Rock",
"track_no": 1,
"disc_no": 1,
"date": "2008-11-04",
"length": 211121,
"musicbrainz_id": "ff1e3e1a-f6e8-4692-b426-355880383bb6",
"last_modified": 1672712949510
}
}
}
}
#+end_src
*** Track playback in progress
#+begin_src json
{
"type": "status",
"data": {
"state": "playing",
"current_track": {
"__model__": "Track",
"uri": "local:track:Various%20Artists%20-%202008%20-%20Twilight%20OST/01-muse-supermassive_black_hole.mp3",
"name": "Supermassive Black Hole",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:250dd6551b66a58a6b4897aa697f200c",
"name": "Muse",
"musicbrainz_id": "9c9f1380-2516-4fc9-a3e6-f9f61941d090"
}
],
"album": {
"__model__": "Album",
"uri": "local:album:md5:455343d54cdd89cb5a3b5ad537ea99d0",
"name": "Twilight: Original Motion Picture Soundtrack",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:54e4db2d5624f80b0cc290346e696756",
"name": "Various Artists",
"musicbrainz_id": "89ad4ac3-39f7-470e-963a-56509c546377"
}
],
"num_tracks": 12,
"num_discs": 1,
"date": "2008-11-04",
"musicbrainz_id": "b4889eaf-d9f4-434c-a68d-69227b12b6a4"
},
"composers": [
{
"__model__": "Artist",
"uri": "local:artist:md5:4d49cbca0b347e0a89047bb019d2779d",
"name": "Matt Bellamy"
}
],
"genre": "Rock",
"track_no": 1,
"disc_no": 1,
"date": "2008-11-04",
"length": 211121,
"musicbrainz_id": "ff1e3e1a-f6e8-4692-b426-355880383bb6",
"last_modified": 1672712949510
},
"time_position": 17031
}
}
#+end_src
*** Track event playback paused
#+begin_src json
{
"type": "event",
"event": "track_playback_paused",
"data": {
"tl_track": {
"__model__": "TlTrack",
"tlid": 14,
"track": {
"__model__": "Track",
"uri": "local:track:Various%20Artists%20-%202008%20-%20Twilight%20OST/01-muse-supermassive_black_hole.mp3",
"name": "Supermassive Black Hole",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:250dd6551b66a58a6b4897aa697f200c",
"name": "Muse",
"musicbrainz_id": "9c9f1380-2516-4fc9-a3e6-f9f61941d090"
}
],
"album": {
"__model__": "Album",
"uri": "local:album:md5:455343d54cdd89cb5a3b5ad537ea99d0",
"name": "Twilight: Original Motion Picture Soundtrack",
"artists": [
{
"__model__": "Artist",
"uri": "local:artist:md5:54e4db2d5624f80b0cc290346e696756",
"name": "Various Artists",
"musicbrainz_id": "89ad4ac3-39f7-470e-963a-56509c546377"
}
],
"num_tracks": 12,
"num_discs": 1,
"date": "2008-11-04",
"musicbrainz_id": "b4889eaf-d9f4-434c-a68d-69227b12b6a4"
},
"composers": [
{
"__model__": "Artist",
"uri": "local:artist:md5:4d49cbca0b347e0a89047bb019d2779d",
"name": "Matt Bellamy"
}
],
"genre": "Rock",
"track_no": 1,
"disc_no": 1,
"date": "2008-11-04",
"length": 211121,
"musicbrainz_id": "ff1e3e1a-f6e8-4692-b426-355880383bb6",
"last_modified": 1672712949510
}
},
"time_position": 67578
}
}
#+end_src
* 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,31 +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>"
VROBBLER_TODOIST_CLIENT_ID="<id>"
VROBBLER_TODOIST_CLIENT_SECRET="<key>"
VROBBLER_GOOGLE_API_KEY="<key>"
VROBBLER_LICHESS_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,34 +0,0 @@
from beers.models import Beer, BeerProducer, BeerStyle
from django.contrib import admin
from scrobbles.admin import ScrobbleInline
class BeerInline(admin.TabularInline):
model = Beer
extra = 0
@admin.register(BeerStyle)
class BeerStyle(admin.ModelAdmin):
date_hierarchy = "created"
search_fields = ("name",)
@admin.register(BeerProducer)
class BeerProducer(admin.ModelAdmin):
date_hierarchy = "created"
search_fields = ("name",)
@admin.register(Beer)
class BeerAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"uuid",
"title",
)
ordering = ("-created",)
search_fields = ("title",)
inlines = [
ScrobbleInline,
]

View File

@ -1,18 +0,0 @@
from rest_framework import serializers
from beers.models import Beer, BeerProducer, BeerStyle
class BeerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = Beer
fields = "__all__"
class BeerProducerSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = BeerProducer
fields = "__all__"
class BeerStyleSerializer(serializers.HyperlinkedModelSerializer):
class Meta:
model = BeerStyle
fields = "__all__"

View File

@ -1,19 +0,0 @@
from rest_framework import permissions, viewsets
from beers.api import serializers
from beers import models
class BeerViewSet(viewsets.ModelViewSet):
queryset = models.Beer.objects.all().order_by("-created")
serializer_class = serializers.BeerSerializer
permission_classes = [permissions.IsAuthenticated]
class BeerProducerViewSet(viewsets.ModelViewSet):
queryset = models.BeerProducer.objects.all().order_by("-created")
serializer_class = serializers.BeerProducerSerializer
permission_classes = [permissions.IsAuthenticated]
class BeerStyleViewSet(viewsets.ModelViewSet):
queryset = models.BeerStyle.objects.all().order_by("-created")
serializer_class = serializers.BeerStyleSerializer
permission_classes = [permissions.IsAuthenticated]

View File

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

View File

@ -1,133 +0,0 @@
# Generated by Django 4.2.16 on 2024-10-22 21:26
from django.db import migrations, models
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
initial = True
dependencies = [
("scrobbles", "0065_alter_scrobble_log"),
]
operations = [
migrations.CreateModel(
name="BeerProducer",
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"
),
),
("description", models.TextField(blank=True, null=True)),
(
"location",
models.CharField(blank=True, max_length=255, null=True),
),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
migrations.CreateModel(
name="Beer",
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)),
("ibu", models.SmallIntegerField(blank=True, null=True)),
("abv", models.FloatField(blank=True, null=True)),
(
"style",
models.CharField(blank=True, max_length=100, null=True),
),
("non_alcoholic", models.BooleanField(default=False)),
(
"beeradvocate_id",
models.CharField(blank=True, max_length=255, null=True),
),
(
"beeradvocate_score",
models.SmallIntegerField(blank=True, null=True),
),
(
"untappd_id",
models.CharField(blank=True, max_length=255, 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,36 +0,0 @@
# Generated by Django 4.2.16 on 2024-10-22 21:34
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
("beers", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="beer",
name="beeradvocate_image",
field=models.ImageField(
blank=True, null=True, upload_to="beers/beeradvcoate/"
),
),
migrations.AddField(
model_name="beer",
name="producer",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
to="beers.beerproducer",
),
),
migrations.AddField(
model_name="beerproducer",
name="beeradvocate_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,75 +0,0 @@
# Generated by Django 4.2.16 on 2024-10-22 21:47
from django.db import migrations, models
import django_extensions.db.fields
class Migration(migrations.Migration):
dependencies = [
("beers", "0002_beer_beeradvocate_image_beer_producer_and_more"),
]
operations = [
migrations.CreateModel(
name="BeerStyle",
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"
),
),
("description", models.TextField(blank=True, null=True)),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
migrations.RemoveField(
model_name="beer",
name="beeradvocate_image",
),
migrations.RemoveField(
model_name="beer",
name="style",
),
migrations.AddField(
model_name="beer",
name="untappd_image",
field=models.ImageField(
blank=True, null=True, upload_to="beers/untappd/"
),
),
migrations.AddField(
model_name="beer",
name="untappd_rating",
field=models.FloatField(blank=True, null=True),
),
migrations.AddField(
model_name="beerproducer",
name="untappd_id",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="beer",
name="styles",
field=models.ManyToManyField(to="beers.beerstyle"),
),
]

View File

@ -1,47 +0,0 @@
# Generated by Django 4.2.16 on 2024-10-22 21:52
from django.db import migrations, models
import uuid
class Migration(migrations.Migration):
dependencies = [
("beers", "0003_beerstyle_remove_beer_beeradvocate_image_and_more"),
]
operations = [
migrations.AddField(
model_name="beerproducer",
name="name",
field=models.CharField(default="Untitled", max_length=255),
preserve_default=False,
),
migrations.AddField(
model_name="beerproducer",
name="uuid",
field=models.UUIDField(
blank=True, default=uuid.uuid4, editable=False, null=True
),
),
migrations.AddField(
model_name="beerstyle",
name="name",
field=models.CharField(default="Untitled", max_length=255),
preserve_default=False,
),
migrations.AddField(
model_name="beerstyle",
name="uuid",
field=models.UUIDField(
blank=True, default=uuid.uuid4, editable=False, null=True
),
),
migrations.AlterField(
model_name="beer",
name="styles",
field=models.ManyToManyField(
related_name="styles", to="beers.beerstyle"
),
),
]

View File

@ -1,21 +0,0 @@
# Generated by Django 4.2.16 on 2025-01-22 03:55
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
(
"beers",
"0004_beerproducer_name_beerproducer_uuid_beerstyle_name_and_more",
),
]
operations = [
migrations.AlterField(
model_name="beer",
name="run_time_seconds",
field=models.IntegerField(default=900),
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.2.19 on 2025-10-30 01:52
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('beers', '0005_alter_beer_run_time_seconds'),
]
operations = [
migrations.RemoveField(
model_name='beer',
name='run_time_seconds',
),
migrations.RemoveField(
model_name='beer',
name='run_time_ticks',
),
migrations.AddField(
model_name='beer',
name='base_run_time_seconds',
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,147 +0,0 @@
from dataclasses import dataclass
from typing import Optional
from uuid import uuid4
from beers.untappd import get_beer_from_untappd_id
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 BaseLogData
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
BNULL = {"blank": True, "null": True}
@dataclass
class BeerLogData(BaseLogData):
rating: Optional[str] = None
class BeerStyle(TimeStampedModel):
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
name = models.CharField(max_length=255)
description = models.TextField(**BNULL)
def __str__(self):
return self.name
class BeerProducer(TimeStampedModel):
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
name = models.CharField(max_length=255)
description = models.TextField(**BNULL)
location = models.CharField(max_length=255, **BNULL)
beeradvocate_id = models.CharField(max_length=255, **BNULL)
untappd_id = models.CharField(max_length=255, **BNULL)
def find_or_create(cls, title: str) -> "BeerProducer":
return cls.objects.filter(title=title).first()
def __str__(self):
return self.name
class Beer(ScrobblableMixin):
description = models.TextField(**BNULL)
ibu = models.SmallIntegerField(**BNULL)
abv = models.FloatField(**BNULL)
styles = models.ManyToManyField(BeerStyle, related_name="styles")
non_alcoholic = models.BooleanField(default=False)
beeradvocate_id = models.CharField(max_length=255, **BNULL)
beeradvocate_score = models.SmallIntegerField(**BNULL)
untappd_image = models.ImageField(upload_to="beers/untappd/", **BNULL)
untappd_image_small = ImageSpecField(
source="untappd_image",
processors=[ResizeToFit(100, 100)],
format="JPEG",
options={"quality": 60},
)
untappd_image_medium = ImageSpecField(
source="untappd_image",
processors=[ResizeToFit(300, 300)],
format="JPEG",
options={"quality": 75},
)
untappd_id = models.CharField(max_length=255, **BNULL)
untappd_rating = models.FloatField(**BNULL)
producer = models.ForeignKey(
BeerProducer, on_delete=models.DO_NOTHING, **BNULL
)
def get_absolute_url(self) -> str:
return reverse("beers:beer_detail", kwargs={"slug": self.uuid})
def __str__(self):
return f"{self.title} by {self.producer}"
@property
def subtitle(self):
return self.producer.name
@property
def strings(self) -> ScrobblableConstants:
return ScrobblableConstants(verb="Drinking", tags="beer")
@property
def beeradvocate_link(self) -> str:
link = ""
if self.producer and self.beeradvocate_id:
if self.beeradvocate_id:
link = f"https://www.beeradvocate.com/beer/profile/{self.producer.beeradvocate_id}/{self.beeradvocate_id}/"
return link
@property
def untappd_link(self) -> str:
link = ""
if self.untappd_id:
link = f"https://www.untappd.com/beer/{self.untappd_id}/"
return link
@property
def primary_image_url(self) -> str:
url = ""
if self.untappd_image:
url = self.untappd_image.url
return url
@property
def logdata_cls(self):
return BeerLogData
@classmethod
def find_or_create(cls, untappd_id: str) -> "Beer":
beer = cls.objects.filter(untappd_id=untappd_id).first()
if not beer:
beer_dict = get_beer_from_untappd_id(untappd_id)
producer_dict = {}
style_ids = []
for key in list(beer_dict.keys()):
if "producer__" in key:
pkey = key.replace("producer__", "")
producer_dict[pkey] = beer_dict.pop(key)
if "styles" in key:
for style in beer_dict.pop("styles"):
style_inst, created = BeerStyle.objects.get_or_create(
name=style
)
style_ids.append(style_inst.id)
producer, _created = BeerProducer.objects.get_or_create(
**producer_dict
)
beer_dict["producer_id"] = producer.id
beer = Beer.objects.create(**beer_dict)
for style_id in style_ids:
beer.styles.add(style_id)
return beer
def scrobbles(self, user_id):
Scrobble = apps.get_model("scrobbles", "Scrobble")
return Scrobble.objects.filter(user_id=user_id, beer=self).order_by(
"-timestamp"
)

View File

@ -1,142 +0,0 @@
import logging
from typing import Optional
import requests
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
UNTAPPD_URL = "https://untappd.com/beer/{id}"
def get_first(key: str, result: dict) -> str:
obj = ""
if obj_list := result.get(key):
obj = obj_list[0]
return obj
def get_title_from_soup(soup) -> str:
title = ""
try:
title = soup.find("h1").get_text()
except AttributeError:
pass
except ValueError:
pass
return title
def get_description_from_soup(soup) -> str:
desc = ""
try:
desc = (
soup.find(class_="beer-descrption-read-less")
.get_text()
.replace("Show Less", "")
.strip()
)
except AttributeError:
pass
except ValueError:
pass
return desc
def get_styles_from_soup(soup) -> list[str]:
styles = []
try:
styles = soup.find("p", class_="style").get_text().split(" - ")
except AttributeError:
pass
except ValueError:
pass
return styles
def get_abv_from_soup(soup) -> Optional[float]:
abv = None
try:
abv = soup.find(class_="abv").get_text()
if abv:
abv = float(abv.strip("\n").strip("% ABV").strip())
except AttributeError:
pass
except ValueError:
pass
except TypeError:
pass
return abv
def get_ibu_from_soup(soup) -> Optional[int]:
ibu = None
try:
ibu = soup.find(class_="ibu").get_text()
if ibu:
ibu = int(ibu.strip("\n").strip(" IBU").strip())
except AttributeError:
pass
except ValueError:
ibu = None
return ibu
def get_rating_from_soup(soup) -> str:
rating = ""
try:
rating = float(
soup.find(class_="num").get_text().strip("(").strip(")")
)
except AttributeError:
rating = None
except ValueError:
rating = None
return rating
def get_producer_id_from_soup(soup) -> str:
id = ""
try:
id = soup.find(class_="brewery").find("a")["href"].strip("/")
except ValueError:
pass
except IndexError:
pass
return id
def get_producer_name_from_soup(soup) -> str:
name = ""
try:
name = soup.find(class_="brewery").find("a").get_text()
except AttributeError:
pass
except ValueError:
pass
return name
def get_beer_from_untappd_id(untappd_id: str) -> dict:
beer_url = UNTAPPD_URL.format(id=untappd_id)
headers = {"User-Agent": "Vrobbler 0.11.12"}
response = requests.get(beer_url, headers=headers)
beer_dict = {"untappd_id": untappd_id}
if response.status_code != 200:
logger.warn(
"Bad response from untappd.com", extra={"response": response}
)
return beer_dict
soup = BeautifulSoup(response.text, "html.parser")
beer_dict["title"] = get_title_from_soup(soup)
beer_dict["description"] = get_description_from_soup(soup)
beer_dict["styles"] = get_styles_from_soup(soup)
beer_dict["abv"] = get_abv_from_soup(soup)
beer_dict["ibu"] = get_ibu_from_soup(soup)
beer_dict["untappd_rating"] = get_rating_from_soup(soup)
beer_dict["producer__untappd_id"] = get_producer_id_from_soup(soup)
beer_dict["producer__name"] = get_producer_name_from_soup(soup)
return beer_dict

View File

@ -1,14 +0,0 @@
from django.urls import path
from beers import views
app_name = "beers"
urlpatterns = [
path("beers/", views.BeerListView.as_view(), name="beer_list"),
path(
"beers/<slug:slug>/",
views.BeerDetailView.as_view(),
name="beer_detail",
),
]

View File

@ -1,11 +0,0 @@
from beers.models import Beer
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class BeerListView(ScrobbleableListView):
model = Beer
class BeerDetailView(ScrobbleableDetailView):
model = Beer

View File

@ -1,56 +0,0 @@
from django.contrib import admin
from boardgames.models import (
BoardGame,
BoardGameLocation,
BoardGamePublisher,
BoardGameDesigner,
)
from scrobbles.admin import ScrobbleInline
@admin.register(BoardGamePublisher)
class BoardGamePublisherAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"uuid",
)
ordering = ("-created",)
@admin.register(BoardGameDesigner)
class BoardGameDesignerAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"uuid",
)
ordering = ("-created",)
@admin.register(BoardGameLocation)
class BoardGameLocationAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"name",
"uuid",
"geo_location",
)
ordering = ("-created",)
@admin.register(BoardGame)
class BoardGameAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"bggeek_id",
"title",
"published_year",
)
search_fields = ("title",)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]

View File

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

View File

@ -1,28 +0,0 @@
from rest_framework import permissions, viewsets
from boardgames.api import serializers
from boardgames import models
class BoardGameDesignerViewSet(viewsets.ModelViewSet):
queryset = models.BoardGameDesigner.objects.all().order_by("-created")
serializer_class = serializers.BoardGameDesignerSerializer
permission_classes = [permissions.IsAuthenticated]
class BoardGamePublisherViewSet(viewsets.ModelViewSet):
queryset = models.BoardGamePublisher.objects.all().order_by("-created")
serializer_class = serializers.BoardGamePublisherSerializer
permission_classes = [permissions.IsAuthenticated]
class BoardGameLocationViewSet(viewsets.ModelViewSet):
queryset = models.BoardGameLocation.objects.all().order_by("-created")
serializer_class = serializers.BoardGameLocationSerializer
permission_classes = [permissions.IsAuthenticated]
class BoardGameViewSet(viewsets.ModelViewSet):
queryset = models.BoardGame.objects.all().order_by("-created")
serializer_class = serializers.BoardGameSerializer
permission_classes = [permissions.IsAuthenticated]

View File

@ -1,155 +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
from django.conf import settings
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}"
BGG_ACCESS_TOKEN = getattr(settings, "BGG_ACCESS_TOKEN", "")
BASE_HEADERS = {"User-Agent": "Vrobbler 31.0", "Authorization": f"Bearer {BGG_ACCESS_TOKEN}"}
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
game_id = None
url = SEARCH_ID_URL.format(query=title)
r = requests.get(url, headers=BASE_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 = {}
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=BASE_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 = "secstate" # user.profile.bgg_username
bgg_password = "yYFCKnfo8AK89lc68q0S"
if not bgg_username or bgg_password:
return
login_payload = {
"credentials": {"username": bgg_username, "password": bgg_password}
}
headers = BASE_HEADERS
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.log:
for player in scrobble.log.get("players"):
player_person = Person.objects.filter(
id=player.get("person_id")
).first()
if player_person.get("bgg_username"):
player["username"] = player_person.get("bgg_username")
player["name"] = player_person.get("name")
player["win"] = player.get("win")
# player["role"] = player.get("role")
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,
}
r = s.post(
"https://boardgamegeek.com/geekplay.php",
data=json.dumps(play_payload),
headers=headers,
)

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,18 +0,0 @@
# Generated by Django 4.2.16 on 2025-01-22 03:55
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0006_alter_boardgame_genre"),
]
operations = [
migrations.AlterField(
model_name="boardgame",
name="run_time_seconds",
field=models.IntegerField(default=900),
),
]

View File

@ -1,167 +0,0 @@
# Generated by Django 4.2.19 on 2025-07-03 01:57
from django.db import migrations, models
import django.db.models.deletion
import django_extensions.db.fields
import uuid
class Migration(migrations.Migration):
dependencies = [
("locations", "0007_alter_geolocation_run_time_seconds"),
("boardgames", "0007_alter_boardgame_run_time_seconds"),
]
operations = [
migrations.CreateModel(
name="BoardGameDesigner",
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,
),
),
("bgg_id", models.IntegerField(blank=True, null=True)),
("bio", models.TextField(blank=True, null=True)),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
migrations.RenameField(
model_name="boardgamepublisher",
old_name="igdb_id",
new_name="bgg_id",
),
migrations.AddField(
model_name="boardgame",
name="bgstats_id",
field=models.UUIDField(blank=True, null=True),
),
migrations.AddField(
model_name="boardgame",
name="cooperative",
field=models.BooleanField(default=False),
),
migrations.AddField(
model_name="boardgame",
name="expansion_for_boardgame",
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
to="boardgames.boardgame",
),
),
migrations.AddField(
model_name="boardgame",
name="highest_wins",
field=models.BooleanField(default=True),
),
migrations.AddField(
model_name="boardgame",
name="max_play_time",
field=models.IntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="boardgame",
name="min_play_time",
field=models.IntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="boardgame",
name="no_points",
field=models.BooleanField(default=False),
),
migrations.AddField(
model_name="boardgame",
name="uses_teams",
field=models.BooleanField(default=False),
),
migrations.CreateModel(
name="BoardGameLocation",
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,
),
),
("bgstats_id", models.UUIDField(blank=True, null=True)),
("description", models.TextField(blank=True, null=True)),
(
"geo_location",
models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.DO_NOTHING,
to="locations.geolocation",
),
),
],
options={
"get_latest_by": "modified",
"abstract": False,
},
),
migrations.AddField(
model_name="boardgame",
name="designers",
field=models.ManyToManyField(
related_name="board_games", to="boardgames.boardgamedesigner"
),
),
]

View File

@ -1,33 +0,0 @@
# Generated by Django 4.2.19 on 2025-07-03 02:21
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0008_boardgamedesigner_and_more"),
]
operations = [
migrations.AlterField(
model_name="boardgame",
name="cooperative",
field=models.BooleanField(blank=True, default=False, null=True),
),
migrations.AlterField(
model_name="boardgame",
name="highest_wins",
field=models.BooleanField(blank=True, default=True, null=True),
),
migrations.AlterField(
model_name="boardgame",
name="no_points",
field=models.BooleanField(blank=True, default=False, null=True),
),
migrations.AlterField(
model_name="boardgame",
name="uses_teams",
field=models.BooleanField(blank=True, default=False, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.2.19 on 2025-07-03 04:05
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0009_alter_boardgame_cooperative_and_more"),
]
operations = [
migrations.AddField(
model_name="boardgame",
name="published_year",
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,26 +0,0 @@
# Generated by Django 4.2.19 on 2025-10-30 01:52
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('boardgames', '0010_boardgame_published_year'),
]
operations = [
migrations.RemoveField(
model_name='boardgame',
name='run_time_seconds',
),
migrations.RemoveField(
model_name='boardgame',
name='run_time_ticks',
),
migrations.AddField(
model_name='boardgame',
name='base_run_time_seconds',
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.2.19 on 2025-11-03 04:34
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('boardgames', '0011_remove_boardgame_run_time_seconds_and_more'),
]
operations = [
migrations.AddField(
model_name='boardgame',
name='bgg_rank',
field=models.IntegerField(blank=True, null=True),
),
]

View File

@ -1,18 +0,0 @@
# Generated by Django 4.2.19 on 2025-11-03 04:41
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('boardgames', '0012_boardgame_bgg_rank'),
]
operations = [
migrations.AddField(
model_name='boardgame',
name='publishers',
field=models.ManyToManyField(related_name='board_games', to='boardgames.boardgamepublisher'),
),
]

View File

@ -1,363 +0,0 @@
from functools import cached_property
import logging
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Any
from uuid import uuid4
from django import forms
import requests
from boardgames.sources.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 locations.models import GeoLocation
from people.models import Person
from scrobbles.dataclasses import BaseLogData, LongPlayLogData
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
logger = logging.getLogger(__name__)
BNULL = {"blank": True, "null": True}
@dataclass
class BoardGameScoreLogData(BaseLogData):
person_id: Optional[int] = None
bgg_username: Optional[str] = None
color: Optional[str] = None
character: Optional[str] = None
team: Optional[str] = None
score: Optional[int] = None
win: Optional[bool] = None
new: Optional[bool] = None
rank: Optional[int] = None
seat_order: Optional[int] = None
role: Optional[str] = None
rank: Optional[int] = None
seat_order: Optional[int] = None
role: Optional[str] = None
lichess_username: Optional[str] = None
@property
def person(self) -> Optional[Person]:
return Person.objects.filter(id=self.person_id).first()
@property
def name(self) -> str:
name = ""
if self.person:
name = self.person.name
return name
def __str__(self) -> str:
out = self.name
if self.score:
out += f" {self.score}"
if self.color:
out += f" ({self.color})"
if self.win:
out += f" [W]"
return out
@dataclass
class BoardGameLogData(BaseLogData, LongPlayLogData):
players: Optional[list[BoardGameScoreLogData]] = None
location_id: Optional[int] = None
difficulty: Optional[int] = None
solo: Optional[bool] = None
two_handed: Optional[bool] = None
expansion_ids: Optional[int] = None
moves: Optional[list] = None
rated: Optional[str] = None
speed: Optional[str] = None
variant: Optional[str] = None
lichess_id: Optional[int] = None
board: Optional[str] = None
rounds: Optional[int] = None
details: Optional[str] = None
_excluded_fields = {
"lichess_id",
"speed",
"rated",
"moves",
"variant",
}
@cached_property
def location(self):
if not self.location_id:
return
return BoardGameLocation.objects.filter(id=self.location_id).first()
@cached_property
def player_log(self) -> str:
if self.players:
return ", ".join(
[
BoardGameScoreLogData(**player).__str__()
for player in self.players
]
)
return ""
@classmethod
def override_fields(cls) -> dict:
fields = {}
for base in cls.mro()[1:]:
if hasattr(base, "override_fields"):
base_fields = base.override_fields()
fields.update(base_fields)
custom_fields = {
"location_id": forms.ModelChoiceField(
queryset=BoardGameLocation.objects.all(),
required=False,
widget=forms.Select(),
)
}
fields.update(custom_fields)
return fields
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)
bgg_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 BoardGameDesigner(TimeStampedModel):
name = models.CharField(max_length=255)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
bgg_id = models.IntegerField(**BNULL)
bio = models.TextField(**BNULL)
def __str__(self) -> str:
return str(self.name)
def get_absolute_url(self):
return reverse(
"boardgames:designer_detail", kwargs={"slug": self.uuid}
)
class BoardGameLocation(TimeStampedModel):
name = models.CharField(max_length=255)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
bgstats_id = models.UUIDField(**BNULL)
description = models.TextField(**BNULL)
geo_location = models.ForeignKey(
GeoLocation, **BNULL, on_delete=models.DO_NOTHING
)
def __str__(self) -> str:
return str(self.name)
def get_absolute_url(self):
return reverse(
"boardgames:location_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
)
publishers = models.ManyToManyField(
BoardGamePublisher,
related_name="board_games",
)
designers = models.ManyToManyField(
BoardGameDesigner,
related_name="board_games",
)
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)
bgg_rank = models.IntegerField(**BNULL)
max_players = models.PositiveSmallIntegerField(**BNULL)
min_players = models.PositiveSmallIntegerField(**BNULL)
published_date = models.DateField(**BNULL)
published_year = models.IntegerField(**BNULL)
recommended_age = models.PositiveSmallIntegerField(**BNULL)
bggeek_id = models.CharField(max_length=255, **BNULL)
bgstats_id = models.UUIDField(**BNULL)
uses_teams = models.BooleanField(default=False, **BNULL)
cooperative = models.BooleanField(default=False, **BNULL)
highest_wins = models.BooleanField(default=True, **BNULL)
no_points = models.BooleanField(default=False, **BNULL)
min_play_time = models.IntegerField(**BNULL)
max_play_time = models.IntegerField(**BNULL)
expansion_for_boardgame = models.ForeignKey(
"self", **BNULL, on_delete=models.DO_NOTHING
)
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
@property
def strings(self) -> ScrobblableConstants:
return ScrobblableConstants(verb="Playing", tags="game_die")
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 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_year"] = int(year)
if not data["min_players"]:
data.pop("min_players")
if not data["min_players"]:
data.pop("max_players")
# 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:
self.save_image_from_url(cover_url)
def save_image_from_url(self, url):
headers = {"User-Agent": "Vrobbler 0.11.12"}
r = requests.get(url, headers=headers)
if r.status_code == 200:
fname = f"{self.title}_cover_{self.uuid}.jpg"
self.cover.save(fname, ContentFile(r.content), save=True)
@classmethod
def find_or_create(
cls, lookup_id: str, data: dict[str, Any] = {}
) -> "BoardGame":
"""Given a Lookup ID (either BGG or BGA ID), return a board game object"""
game = cls.objects.filter(bggeek_id=lookup_id).first()
if game:
logger.info("Board game exists in database.", extra={"lookup_id": lookup_id, "data": data})
return game
bgg_data = lookup_boardgame_from_bgg(data.get("name"))
mechanics = bgg_data.pop("mechanics", [])
designers = bgg_data.pop("designers", [])
categories = bgg_data.pop("categories", [])
publishers = bgg_data.pop("publishers", [])
cover_url = bgg_data.pop("cover_url")
game = cls.objects.create(
**bgg_data
)
game.save_image_from_url(cover_url)
game.cooperative = data.get("cooperative", False)
game.highest_wins = data.get("highestWins", True)
game.no_points = data.get("noPoints", False)
game.uses_teams = data.get("useTeams", False)
game.bgstats_id = data.get("uuid", None)
game.save()
if designers:
for designer_name in designers:
designer, created = BoardGameDesigner.objects.get_or_create(
name=designer_name
)
game.designers.add(designer.id)
if publishers:
for name in publishers:
publisher, _ = BoardGamePublisher.objects.get_or_create(
name=name
)
game.publishers.add(publisher)
return game

View File

@ -1,29 +0,0 @@
from typing import Any
from boardgamegeek import BGGClient
from django.conf import settings
def lookup_boardgame_from_bgg(title: str) -> dict[str, Any]:
game_dict = {"title": title}
bgg = BGGClient(access_token=settings.BGG_ACCESS_TOKEN)
game = bgg.game(title)
if game:
game_dict["description"] = game.description
game_dict["published_year"] = game.yearpublished
game_dict["cover_url"] = game.image
game_dict["min_players"] = game.minplayers
game_dict["max_players"] = game.maxplayers
game_dict["recommended_age"] = game.minage
game_dict["rating"] = game.rating_average
game_dict["bgg_rank"] = game.bgg_rank
game_dict["base_run_time_seconds"] = int(game.playingtime) * 60 if game.playingtime else None
game_dict["mechanics"] = game.mechanics
game_dict["categories"] = game.categories
game_dict["designers"] = game.designers
game_dict["publishers"] = game.publishers
return game_dict

View File

@ -1,128 +0,0 @@
import berserk
from boardgames.models import BoardGame
from django.conf import settings
from django.contrib.auth import get_user_model
from scrobbles.models import Scrobble
from scrobbles.notifications import ScrobbleNtfyNotification
User = get_user_model()
def import_chess_games_for_user_id(user_id: int, commit: bool = False) -> dict:
user = User.objects.get(id=user_id)
client = berserk.Client(
session=berserk.TokenSession(settings.LICHESS_API_KEY)
)
games = client.games.export_by_player(user.profile.lichess_username)
for game_dict in games:
chess, created = BoardGame.objects.get_or_create(title="Chess")
if created:
chess.base_run_time_seconds = 1800
chess.bggeek_id = 171
chess.save(update_fields=["base_run_time_seconds", "bggeek_id"])
scrobble = Scrobble.objects.filter(
user_id=user.id,
timestamp=game_dict.get("createdAt"),
board_game_id=chess.id,
).first()
if scrobble:
continue
log_data = {
"variant": game_dict.get("variant"),
"lichess_id": game_dict.get("id"),
"rated": game_dict.get("rated"),
"speed": game_dict.get("speed"),
"moves": game_dict.get("moves"),
"players": [],
}
winner = game_dict.get("winner")
black_player = game_dict.get("players", {}).get("black", {})
white_player = game_dict.get("players", {}).get("white", {})
user_player = {
"user_id": user.id,
"lichess_username": user.profile.lichess_username,
"bgg_username": user.profile.bgg_username,
"color": "",
"win": False,
}
other_player = {"name_str": "", "color": "", "win": False}
if (
black_player.get("user", {}).get("name", "")
== user.profile.lichess_username
):
user_player["color"] = "black"
if "aiLevel" in white_player.keys():
other_player["name_str"] = "aiLevel_" + str(
white_player.get("aiLevel", "")
)
else:
other_player["name_str"] = white_player.get("user", {}).get(
"name", ""
)
other_player["lichess_username"] = other_player["name_str"]
other_player["color"] = "white"
if winner == "black":
user_player["win"] = True
else:
other_player["win"] = True
if (
white_player.get("user", {}).get("name", "")
== user.profile.lichess_username
):
user_player["color"] = "white"
if "aiLevel" in black_player.keys():
other_player["name_str"] = "aiLevel_" + str(
black_player.get("aiLevel", "")
)
else:
other_player["name_str"] = black_player.get("user", {}).get(
"name", ""
)
other_player["lichess_username"] = other_player["name_str"]
other_player["color"] = "black"
if winner == "white":
user_player["win"] = True
else:
other_player["win"] = True
log_data["players"].append(user_player)
log_data["players"].append(other_player)
scrobble_dict = {
"media_type": Scrobble.MediaType.BOARD_GAME,
"user_id": user.id,
"playback_position_seconds": (
game_dict.get("lastMoveAt") - game_dict.get("createdAt")
).seconds,
"in_progress": False,
"played_to_completion": True,
"timestamp": game_dict.get("createdAt"),
"stop_timestamp": game_dict.get("lastMoveAt"),
"board_game_id": chess.id,
"source": "Lichess",
"timezone": user.profile.timezone,
"log": log_data,
}
if commit:
Scrobble.objects.create(**scrobble_dict)
return scrobble_dict
def import_chess_games_for_all_users():
scrobbles_to_create = []
for user in User.objects.filter(profile__lichess_username__isnull=False):
scrobble_dict = import_chess_games_for_user_id(user.id)
scrobbles_to_create.append(Scrobble(**scrobble_dict))
if scrobbles_to_create:
created = Scrobble.objects.bulk_create(scrobbles_to_create)
for scrobble in created:
ScrobbleNtfyNotification(scrobble).send()
return scrobbles_to_create

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,16 +0,0 @@
from django.views import generic
from boardgames.models import BoardGame, BoardGamePublisher
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class BoardGameListView(ScrobbleableListView):
model = BoardGame
class BoardGameDetailView(ScrobbleableDetailView):
model = BoardGame
class BoardGamePublisherDetailView(generic.DetailView):
model = BoardGamePublisher
slug_field = "uuid"

View File

@ -1,54 +1,25 @@
from books.models import Author, Book, Paper
from django.contrib import admin
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",)
list_display = ("name", "openlibrary_id")
ordering = ("name",)
@admin.register(Book)
class BookAdmin(admin.ModelAdmin):
class ArtistAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"title",
"author",
"issue_or_volume",
"isbn_13",
"isbn",
"first_publish_year",
"pages",
"openlibrary_id",
)
search_fields = ("name",)
ordering = ("-created",)
inlines = [
ScrobbleInline,
]
def issue_or_volume(self, obj):
return obj.issue_number or obj.volume_number
@admin.register(Paper)
class BookAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"title",
"subtitle",
"arxiv_id",
"first_publish_year",
"pages",
)
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

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

View File

@ -1,9 +0,0 @@
#!/usr/bin/env python3
BOOKS_TITLES_TO_IGNORE = [
"KOReader Quickstart Guide",
"zb2rhkSwygt9vjkAEBj7tP5KVgFqejJqsJ2W3bYsrgiiKK8XL",
"zb2rhchGpo7P27mofV9hYjT63d9ZaQnbQ6LSfzmkvsYzvARif",
]
READCOMICSONLINE_URL = "https://readcomicsonline.ru"

View File

@ -1,432 +0,0 @@
import logging
import re
import sqlite3
from datetime import datetime, timedelta
from enum import Enum
from zoneinfo import ZoneInfo
import requests
from books.constants import BOOKS_TITLES_TO_IGNORE
from django.apps import apps
from django.contrib.auth import get_user_model
from scrobbles.notifications import ScrobbleNtfyNotification
from stream_sqlite import stream_sqlite
from webdav.client import get_webdav_client
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
"""
from books.models import Author
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)
# TODO Move these to async processes after importing
# author.fix_metadata()
logger.debug(f"Created author {author}")
author_list.append(author)
return author_list
def create_book_from_row(row: list):
from books.models import Book
# No KoReader book yet, create it
author_str = get_author_str_from_row(row).replace("\x00", "")
total_pages = row[KoReaderBookColumn.PAGES.value]
run_time = total_pages * Book.AVG_PAGE_READING_SECONDS
book_title = row[KoReaderBookColumn.TITLE.value].replace("\x00", "")
if " - " in book_title:
split_title = book_title.split(" - ")
book_title = split_title[0]
if (not author_str or author_str == "N/A") and len(split_title) > 1:
author_str = split_title[1].split("_")[0]
clean_row = []
for value in row:
if isinstance(value, str):
value = value.replace("\x00", "")
clean_row.append(value)
book = Book.objects.create(
title=book_title.replace("_", ":"),
pages=total_pages,
koreader_data_by_hash={
str(row[KoReaderBookColumn.MD5.value]): {
"title": book_title,
"author_str": author_str,
"book_id": row[KoReaderBookColumn.ID.value],
"raw_row_data": clean_row,
}
},
base_run_time_seconds=run_time,
)
# TODO Move these to async processes after importing
# 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.
"""
from books.models import Book
book_id_map = {}
for book_row in rows:
if book_row[KoReaderBookColumn.TITLE.value] in BOOKS_TITLES_TO_IGNORE:
logger.info(
"[build_book_map] Ignoring book title that is likely garbage",
extra={"book_row": book_row, "media_type": "Book"},
)
continue
book = Book.objects.filter(
koreader_data_by_hash__icontains=book_row[
KoReaderBookColumn.MD5.value
]
).first()
if not book:
title = (
book_row[KoReaderBookColumn.TITLE.value]
.split(" - ")[0]
.lower()
.replace("\x00", "")
)
book = Book.objects.filter(title=title).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
timestamp = user.profile.get_timestamp_with_tz(
datetime.fromtimestamp(int(first_page.get("start_ts")))
)
stop_timestamp = user.profile.get_timestamp_with_tz(
datetime.fromtimestamp(int(last_page.get("end_ts")))
)
# Adjust for Daylight Saving Time
#if timestamp.dst() == timedelta(
# 0
#) or stop_timestamp.dst() == timedelta(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=timestamp.tzinfo.name,
)
)
# 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 = ZoneInfo("UTC")
if user:
tz = user.profile.tzinfo
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)
if created:
ScrobbleNtfyNotification(created[-1]).send()
fix_long_play_stats_for_scrobbles(created)
logger.info(
f"Created {len(created)} scrobbles",
extra={"created_scrobbles": created},
)
return created
def fetch_file_from_webdav(user_id: int) -> str:
file_path = f"/tmp/{user_id}-koreader-import.sqlite3"
client = get_webdav_client(user_id)
if not client:
logger.warning("could not get webdav client for user")
# TODO maybe we raise an exception here?
return ""
client.download_sync(
remote_path="var/koreader/statistics.sqlite3",
local_path=file_path,
)
return file_path

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,40 +0,0 @@
from enum import Enum
from typing import Optional
import pendulum
YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v="
IMDB_VIDEO_URL = "https://www.imdb.com/title/tt"
class BookType:
...
class BookMetadata:
title: str
run_time_seconds: Optional[int]
authors = Optional[str]
goodreads_id = Optional[str]
koreader_data_by_hash = Optional[dict]
isbn = Optional[str]
# isbn_13 = Optional[str]
# isbn_10 = Optional[str]
pages = Optional[int]
language = Optional[str]
first_publish_year = Optional[int]
summary = Optional[str]
# General
cover_url: Optional[str]
genres: list[str]
def __init__(self, title: Optional[str] = ""):
self.title = title
def as_dict_with_authors_cover_and_genres(self) -> tuple:
book_dict = vars(self)
authors = book_dict.pop("authors")
cover = book_dict.pop("cover_url")
genres = book_dict.pop("genres")
return book_dict, authors, cover, genres

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,25 +0,0 @@
# Generated by Django 4.2.16 on 2024-10-17 20:08
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("books", "0020_author_comicvine_data_book_comicvine_data_and_more"),
]
operations = [
migrations.RemoveField(
model_name="book",
name="koreader_authors",
),
migrations.RemoveField(
model_name="book",
name="koreader_id",
),
migrations.RemoveField(
model_name="book",
name="koreader_md5",
),
]

View File

@ -1,21 +0,0 @@
# Generated by Django 4.2.16 on 2025-01-22 03:55
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
(
"books",
"0021_remove_book_koreader_authors_remove_book_koreader_id_and_more",
),
]
operations = [
migrations.AlterField(
model_name="book",
name="run_time_seconds",
field=models.IntegerField(default=900),
),
]

View File

@ -1,40 +0,0 @@
# Generated by Django 4.2.18 on 2025-01-27 04:20
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("books", "0022_alter_book_run_time_seconds"),
]
operations = [
migrations.RenameField(
model_name="book",
old_name="isbn",
new_name="isbn_13",
),
migrations.RemoveField(
model_name="book",
name="comicvine_data",
),
migrations.RemoveField(
model_name="book",
name="goodreads_id",
),
migrations.RemoveField(
model_name="book",
name="locg_slug",
),
migrations.AddField(
model_name="book",
name="isbn_10",
field=models.CharField(blank=True, max_length=255, null=True),
),
migrations.AddField(
model_name="book",
name="publish_date",
field=models.DateField(blank=True, null=True),
),
]

View File

@ -1,21 +0,0 @@
# Generated by Django 4.2.18 on 2025-01-27 04:31
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
(
"books",
"0023_rename_isbn_book_isbn_13_remove_book_comicvine_data_and_more",
),
]
operations = [
migrations.AddField(
model_name="book",
name="publisher",
field=models.CharField(blank=True, max_length=255, null=True),
),
]

View File

@ -1,136 +0,0 @@
# Generated by Django 4.2.19 on 2025-02-18 05:11
from django.db import migrations, models
import django_extensions.db.fields
import taggit.managers
import uuid
class Migration(migrations.Migration):
dependencies = [
("scrobbles", "0067_scrobble_food_alter_scrobble_media_type"),
("books", "0024_book_publisher"),
]
operations = [
migrations.RemoveField(
model_name="author",
name="amazon_id",
),
migrations.RemoveField(
model_name="author",
name="librarything_id",
),
migrations.RemoveField(
model_name="author",
name="locg_slug",
),
migrations.AddField(
model_name="author",
name="semantic_id",
field=models.CharField(blank=True, max_length=50, null=True),
),
migrations.CreateModel(
name="Paper",
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,
),
),
("run_time_seconds", models.IntegerField(default=900)),
(
"run_time_ticks",
models.PositiveBigIntegerField(blank=True, null=True),
),
("title", models.CharField(max_length=255)),
("semantic_title", models.CharField(max_length=255)),
(
"koreader_data_by_hash",
models.JSONField(blank=True, null=True),
),
(
"semantic_id",
models.CharField(blank=True, max_length=50, null=True),
),
(
"arxiv_id",
models.CharField(blank=True, max_length=50, null=True),
),
(
"corpus_id",
models.CharField(blank=True, max_length=50, null=True),
),
(
"doi_id",
models.CharField(blank=True, max_length=50, null=True),
),
("pages", models.IntegerField(blank=True, null=True)),
(
"language",
models.CharField(blank=True, max_length=4, null=True),
),
(
"first_publish_year",
models.IntegerField(blank=True, null=True),
),
("publish_date", models.DateField(blank=True, null=True)),
(
"journal",
models.CharField(blank=True, max_length=255, null=True),
),
(
"journal_volume",
models.CharField(blank=True, max_length=50, null=True),
),
("abstract", models.TextField(blank=True, null=True)),
("num_citations", models.IntegerField(blank=True, null=True)),
(
"openaccess_pdf_url",
models.CharField(blank=True, max_length=255, null=True),
),
(
"authors",
models.ManyToManyField(blank=True, to="books.author"),
),
(
"genre",
taggit.managers.TaggableManager(
help_text="A comma-separated list of tags.",
through="scrobbles.ObjectWithGenres",
to="scrobbles.Genre",
verbose_name="Tags",
),
),
],
options={
"abstract": False,
},
),
]

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