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

[boardgames] Clean up email parser to work with many plays

Browse Source
This commit is contained in:
Colin Powell
2025-07-03 00:34:51 -04:00
parent 1590ce5f18
commit 1fd325823b
5 changed files with 134 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -47,7 +47,7 @@ class BoardGameAdmin(admin.ModelAdmin):
list_display = (
"bggeek_id",
"title",
"published_date",
"published_year",
)
search_fields = ("title",)
ordering = ("-created",)

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

@ -100,8 +100,8 @@ def lookup_boardgame_from_bgg(lookup_id: str) -> dict:
def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]:
bgg_username = user.profile.bgg_username
bgg_password = user.profile.bgg_password
bgg_username = "secstate" # user.profile.bgg_username
bgg_password = "yYFCKnfo8AK89lc68q0S"
if not bgg_username or bgg_password:
return
@ -119,24 +119,22 @@ def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]:
data=json.dumps(login_payload),
headers=headers,
)
print(p)
players = []
if scrobble.metadata:
for player in scrobble.metadata.players:
if player["user_id"]:
player_user = User.objects.filter(
id=player["user_id"]
).first()
if player_user:
if player_user.bgg_username:
player["username"] = player_user.bgg_username
else:
player["name"] = player_user.username
player["win"] = player.get("win")
player["color"] = player.get("color")
player["new"] = player.get("new")
player["score"] = player.get("score")
players.append(player)
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"),
@ -150,3 +148,9 @@ def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]:
"objecttype": "thing",
"ajax": 1,
}
r = s.post(
"https://boardgamegeek.com/geekplay.php",
data=json.dumps(play_payload),
headers=headers,
)
print(r)

18
vrobbler/apps/boardgames/migrations/0011_boardgame_published_year.py Normal file
View File

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

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

@ -123,6 +123,7 @@ class BoardGame(ScrobblableMixin):
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)
@ -176,7 +177,7 @@ class BoardGame(ScrobblableMixin):
publisher_name = data.pop("publisher_name")
if year:
data["published_date"] = datetime(int(year), 1, 1)
data["published_year"] = int(year)
if not data["min_players"]:
data.pop("min_players")

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

@ -333,7 +333,15 @@ def find_and_enrich_board_game_data(game_dict: dict) -> BoardGame | None:
def email_scrobble_board_game(
bgstat_data: dict[str, Any], user_id: int
) -> Scrobble | None:
) -> list[Scrobble]:
game_list: list = bgstat_data.get("games", [])
if not game_list:
logger.info(
"No game data from BG Stats, not scrobbling",
extra={"bgstat_data": bgstat_data},
)
return []
player_dict = {}
for player in bgstat_data.get("players", []):
if player.get("isAnonymous"):
@ -347,82 +355,105 @@ def email_scrobble_board_game(
person.save()
player_dict[player.get("id")] = person
game_list: list = bgstat_data.get("games", [])
if not game_list:
logger.info(
"No game data from BG Stats, not scrobbling",
extra={"bgstat_data": bgstat_data},
)
return
base_game = None
expansions = []
base_games = {}
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
base_games[game.get("id")] = enriched_game
elif game.get("isExpansion"):
expansions.append(enriched_game)
expansions[game.get("id")] = enriched_game
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")
locations = {}
for location_dict in bgstat_data.get("locations", []):
location, _created = BoardGameLocation.objects.get_or_create(
bgstats_id=location_dict.get("uuid")
)
update_fields = []
if not location.name:
location.name = location_dict.get("name")
update_fields.append("name")
geoloc = GeoLocation.objects.filter(
title__icontains=location.name
).first()
if geoloc:
location.geo_location = geoloc
location.save()
log_data["location_id"] = location.id
update_fields.append("geo_location")
if update_fields:
location.save(update_fields=update_fields)
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")
locations[location_dict.get("id")] = location
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"),
}
)
scrobbles_created = []
for play_dict in bgstat_data.get("plays", []):
log_data["expansion_ids"] = []
try:
base_game = base_games[play_dict.get("gameRefId")]
except KeyError:
try:
base_game = expansions[play_dict.get("gameRefId")]
except KeyError:
print(play_dict)
logger.info(
"Skipping scrobble of play, can't find game",
extra={"play_dict": play_dict},
)
continue
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,
}
for eplay in play_dict.get("expansionPlays", []):
expansion = expansions[eplay.get("gameRefId")]
expansion.expansion_for_boardgame = base_game
expansion.save()
log_data["expansion_ids"].append(expansion.id)
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()
if play_dict.get("locationRefId", False):
log_data["location_id"] = locations[
play_dict.get("locationRefId")
].id
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")
return scrobble
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"))
if play_dict.get("durationMin") > 0:
duration_seconds = play_dict.get("durationMin") * 60
else:
duration_seconds = base_game.run_time_seconds
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,
}
print(scrobble_dict)
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()
scrobbles_created.append(scrobble)
return scrobbles_created
def manual_scrobble_from_url(