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

[boardgames] Adding email scrobbler for BG Stats

Browse Source
This commit is contained in:
Colin Powell
2025-07-02 23:01:13 -04:00
parent 3548c29f97
commit 1590ce5f18
13 changed files with 469 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
vrobbler/apps/boardgames/admin.py
View File

@ -1,6 +1,11 @@
from django.contrib import admin
from boardgames.models import BoardGame, BoardGamePublisher
from boardgames.models import (
BoardGame,
BoardGameLocation,
BoardGamePublisher,
BoardGameDesigner,
)
from scrobbles.admin import ScrobbleInline
@ -15,8 +20,29 @@ class BoardGamePublisherAdmin(admin.ModelAdmin):
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 GameAdmin(admin.ModelAdmin):
class BoardGameAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = (
"bggeek_id",

167
vrobbler/apps/boardgames/migrations/0008_boardgamedesigner_and_more.py Normal file
View File

@ -0,0 +1,167 @@
# 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.IntegerField(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"
),
),
]

33
vrobbler/apps/boardgames/migrations/0009_alter_boardgame_cooperative_and_more.py Normal file
View File

@ -0,0 +1,33 @@
# 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),
),
]

18
vrobbler/apps/boardgames/migrations/0010_alter_boardgamelocation_bgstats_id.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 4.2.19 on 2025-07-03 02:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("boardgames", "0009_alter_boardgame_cooperative_and_more"),
]
operations = [
migrations.AlterField(
model_name="boardgamelocation",
name="bgstats_id",
field=models.UUIDField(blank=True, null=True),
),
]

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

@ -14,6 +14,7 @@ from imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFit
from scrobbles.dataclasses import BoardGameLogData
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
from locations.models import GeoLocation
logger = logging.getLogger(__name__)
BNULL = {"blank": True, "null": True}
@ -23,7 +24,7 @@ class BoardGamePublisher(TimeStampedModel):
name = models.CharField(max_length=255)
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
logo = models.ImageField(upload_to="games/platform-logos/", **BNULL)
igdb_id = models.IntegerField(**BNULL)
bgg_id = models.IntegerField(**BNULL)
def __str__(self):
return self.name
@ -34,6 +35,39 @@ class BoardGamePublisher(TimeStampedModel):
)
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
@ -53,6 +87,10 @@ class BoardGame(ScrobblableMixin):
publisher = models.ForeignKey(
BoardGamePublisher, **BNULL, on_delete=models.DO_NOTHING
)
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)
@ -87,6 +125,16 @@ class BoardGame(ScrobblableMixin):
published_date = models.DateField(**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

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

@ -1,13 +1,11 @@
from decimal import Decimal, getcontext
import logging
from typing import Dict
from uuid import uuid4
from django.contrib.auth import get_user_model
from django.conf import settings
from django.db import models
from django.urls import reverse
from django_extensions.db.models import TimeStampedModel
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
logger = logging.getLogger(__name__)

2
vrobbler/apps/people/admin.py
View File

@ -5,6 +5,6 @@ from people.models import Person
@admin.register(Person)
class PersonAdmin(admin.ModelAdmin):
date_hierarchy = "created"
list_display = ("name", "bgg_username", "bgstat_id")
list_display = ("name", "bgg_username", "bgstats_id")
ordering = ("-created",)
search_fields = ("name",)

22
vrobbler/apps/people/migrations/0002_remove_person_bgstat_id_person_bgstats_id.py Normal file
View File

@ -0,0 +1,22 @@
# Generated by Django 4.2.19 on 2025-07-03 02:25
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("people", "0001_initial"),
]
operations = [
migrations.RemoveField(
model_name="person",
name="bgstat_id",
),
migrations.AddField(
model_name="person",
name="bgstats_id",
field=models.UUIDField(blank=True, null=True),
),
]

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

@ -11,7 +11,7 @@ class Person(TimeStampedModel):
user = models.ForeignKey(User, on_delete=models.DO_NOTHING, **BNULL)
name = models.CharField(max_length=100, **BNULL)
bgstat_id = models.CharField(max_length=100, **BNULL)
bgstats_id = models.UUIDField(**BNULL)
bgg_username = models.CharField(max_length=100, **BNULL)
lichess_username = models.CharField(max_length=100, **BNULL)
bio = models.TextField(**BNULL)

21
vrobbler/apps/profiles/migrations/0025_rename_bgstat_id_userprofile_bgstats_id.py Normal file
View File

@ -0,0 +1,21 @@
# Generated by Django 4.2.19 on 2025-07-03 02:28
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
(
"profiles",
"0024_userprofile_bgstat_id_userprofile_imap_auto_import_and_more",
),
]
operations = [
migrations.RenameField(
model_name="userprofile",
old_name="bgstat_id",
new_name="bgstats_id",
),
]

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

@ -31,7 +31,7 @@ class UserProfile(TimeStampedModel):
task_context_tags_str = models.CharField(max_length=255, **BNULL)
bgstat_id = models.CharField(max_length=255, **BNULL)
bgstats_id = models.CharField(max_length=255, **BNULL)
bgg_username = models.CharField(max_length=255, **BNULL)
lichess_username = models.CharField(max_length=255, **BNULL)

49
vrobbler/apps/scrobbles/imap.py
View File

@ -13,7 +13,7 @@ logger = logging.getLogger(__name__)
def process_scrobbles_from_imap() -> list[Scrobble] | None:
"""For all user profiles with IMAP creds, check inbox for scrobbleable email attachments."""
scrobbles_to_create: list[Scrobble] = []
scrobbles_created: list[Scrobble] = []
active_profiles = UserProfile.objects.filter(imap_auto_import=True)
for profile in active_profiles:
@ -23,11 +23,12 @@ def process_scrobbles_from_imap() -> list[Scrobble] | None:
)
mail = imaplib.IMAP4_SSL(profile.imap_url)
mail.login(profile.imap_user, profile.imap_pass)
mail.select("INBOX")
mail.select("INBOX") # TODO configure this in profile
# Search for unseen emails
status, messages = mail.search(None, "(UNSEEN)")
status, messages = mail.search(None, "UnSeen")
if status != "OK":
logger.info("IMAP status not OK", extra={"status": status})
return
for uid in messages[0].split():
@ -37,13 +38,15 @@ def process_scrobbles_from_imap() -> list[Scrobble] | None:
continue
try:
msg = email.message_from_bytes(msg_data[0][1])
logger.info("Processing email message", extra={"msg": msg})
message = email.message_from_bytes(msg_data[0][1])
logger.info(
"Processing email message", extra={"email_msg": message}
)
except IndexError:
logger.info("No email message data found")
return
for part in msg.walk():
for part in message.walk():
if part.get_content_disposition() == "attachment":
filename = part.get_filename()
if filename:
@ -57,42 +60,34 @@ def process_scrobbles_from_imap() -> list[Scrobble] | None:
parsed_json = ""
# Try parsing JSON if applicable
parsed_json = {}
if filename.lower().endswith(".bgsplay"):
try:
parsed_json = json.loads(
file_data.decode("utf-8")
)
scrobbles_to_create.append(
email_scrobble_board_game(
parsed_json, profile.user_id
)
)
except Exception as e:
# You might want to log this
print(
f"Failed to parse JSON from {filename}: {e}"
)
# Avoid duplicates
if not EmailAttachment.objects.filter(
email_uid=uid.decode(), filename=filename
).exists():
attachment = EmailAttachment(
email_uid=uid.decode(), filename=filename
if not parsed_json:
logger.info("No JSON found in BG Stats file")
continue
scrobble = email_scrobble_board_game(
parsed_json, profile.user_id
)
attachment.file.save(
filename, ContentFile(file_data)
)
attachment.save()
if scrobble:
scrobbles_created.append(scrobble)
mail.logout()
if scrobbles_to_create:
if scrobbles_created:
logger.info(
f"Creating {len(scrobbles_to_create)} new scrobbles",
extra={"scrobbles_to_create": scrobbles_to_create},
f"Creating {len(scrobbles_created)} new scrobbles",
extra={"scrobbles_created": scrobbles_created},
)
created_scrobbles = Scrobble.objects.bulk_create(scrobbles_to_create)
return created_scrobbles
return scrobbles_created
logger.info(f"No new scrobbles found in IMAP folders")

120
vrobbler/apps/scrobbles/scrobblers.py
View File

@ -1,12 +1,12 @@
import logging
import re
from datetime import datetime
from datetime import datetime, timedelta
from typing import Any, Optional
import pendulum
import pytz
from beers.models import Beer
from boardgames.models import BoardGame
from boardgames.models import BoardGame, BoardGameDesigner, BoardGameLocation
from books.models import Book
from bricksets.models import BrickSet
from dateutil.parser import parse
@ -15,6 +15,7 @@ from locations.constants import LOCATION_PROVIDERS
from locations.models import GeoLocation
from music.constants import JELLYFIN_POST_KEYS, MOPIDY_POST_KEYS
from music.models import Track
from people.models import Person
from podcasts.models import PodcastEpisode
from podcasts.utils import parse_mopidy_uri
from puzzles.models import Puzzle
@ -28,10 +29,10 @@ from scrobbles.utils import convert_to_seconds, extract_domain
from sports.models import SportEvent
from sports.thesportsdb import lookup_event_from_thesportsdb
from tasks.models import Task
from tasks.utils import get_title_from_labels
from videogames.howlongtobeat import lookup_game_from_hltb
from videogames.models import VideoGame
from videos.models import Video
from tasks.utils import get_title_from_labels
from webpages.models import WebPage
logger = logging.getLogger(__name__)
@ -311,26 +312,117 @@ def manual_scrobble_board_game(
return Scrobble.create_or_update(boardgame, user_id, scrobble_dict)
def find_and_enrich_board_game_data(game_dict: dict) -> BoardGame | None:
"""TODO Move this to a utility somewhere"""
game = BoardGame.find_or_create(game_dict.get("bggId"))
if game:
game.cooperative = game_dict.get("cooperative", False)
game.highest_wins = game_dict.get("highestWins", True)
game.no_points = game_dict.get("noPoints", False)
game.uses_teams = game_dict.get("useTeams", False)
if not game.rating:
game.rating = game_dict.get("rating") / 10
game.save()
if game_dict.get("designers"):
for designer_name in game_dict.get("designers", "").split(", "):
BoardGameDesigner.objects.get_or_create(name=designer_name)
return game
def email_scrobble_board_game(
bgstat_data: dict[str, Any], user_id: int
) -> Scrobble | None:
game_dict: dict[str, Any] = bgstat_data.get("games", [])[0]
if not game_dict.get("bggId", False):
player_dict = {}
for player in bgstat_data.get("players", []):
if player.get("isAnonymous"):
person, _created = Person.objects.get_or_create(name="Anonymous")
else:
person, _created = Person.objects.get_or_create(
bgstats_id=player.get("uuid")
)
if not person.name:
person.name = player.get("name", "")
person.save()
player_dict[player.get("id")] = person
game_list: list = bgstat_data.get("games", [])
if not game_list:
logger.info(
"Data from BG Stats JSON had not BGG ID, not scrobbling",
"No game data from BG Stats, not scrobbling",
extra={"bgstat_data": bgstat_data},
)
return
boardgame = BoardGame.find_or_create(game_dict.get("bggId"))
# TODO Enrich data from our bgstats data?
#
# TODO Build up board game meta data from the rest of the data
scrobble_dict = {}
players_list: list[dict[str, Any]] = bgstat_data.get("players", [])
location: dict[str, Any] = bgstat_data.get("locations", [])[0]
base_game = None
expansions = []
log_data = {}
for game in game_list:
logger.info(f"Finding and enriching {game.get('name')}")
enriched_game = find_and_enrich_board_game_data(game)
if game.get("isBaseGame"):
base_game = enriched_game
elif game.get("isExpansion"):
expansions.append(enriched_game)
return Scrobble.create_or_update(boardgame, user_id, scrobble_dict)
for expansion in expansions:
expansion.expansion_for_boardgame = base_game
expansion.save()
log_data["expansion_ids"] = [e.id for e in expansions]
location_dict: dict[str, Any] = bgstat_data.get("locations", [])[0]
location, _created = BoardGameLocation.objects.get_or_create(
bgstats_id=location_dict.get("uuid")
)
if not location.name:
location.name = location_dict.get("name")
geoloc = GeoLocation.objects.filter(
title__icontains=location.name
).first()
if geoloc:
location.geo_location = geoloc
location.save()
log_data["location_id"] = location.id
play_dict = bgstat_data.get("plays", [])[0]
if play_dict.get("rounds", False):
log_data["rounds"] = play_dict.get("rounds")
if play_dict.get("board", False):
log_data["board"] = play_dict.get("board")
log_data["players"] = []
for score_dict in play_dict.get("playerScores", []):
log_data["players"].append(
{
"person_id": player_dict[score_dict.get("playerRefId")].id,
"new": score_dict.get("newPlayer"),
"win": score_dict.get("winner"),
"score": score_dict.get("score"),
"rank": score_dict.get("rank"),
"seat_order": score_dict.get("seatOrder"),
"role": score_dict.get("role"),
}
)
start = parse(play_dict.get("playDate"))
duration_seconds = play_dict.get("durationMin") * 60
stop = start + timedelta(seconds=duration_seconds)
scrobble_dict = {
"user_id": user_id,
"timestamp": start,
"playback_position_seconds": duration_seconds,
"source": "BG Stats",
"log": log_data,
}
scrobble = Scrobble.create_or_update(base_game, user_id, scrobble_dict)
scrobble.stop_timestamp = stop
scrobble.in_progress = False
scrobble.played_to_completion = True
scrobble.save()
return scrobble
def manual_scrobble_from_url(