From 00249e3d3b4092578d075408ee5d0a133072a299 Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Wed, 15 Jul 2026 16:41:12 -0400 Subject: [PATCH] [boardgames] Fix and clean up BGG push integration --- PROJECT.org | 29 ++++---- vrobbler/apps/boardgames/bgg.py | 122 ++++++++++++++++++++++++++------ 2 files changed, 114 insertions(+), 37 deletions(-) diff --git a/PROJECT.org b/PROJECT.org index e1580da..b474e67 100644 --- a/PROJECT.org +++ b/PROJECT.org @@ -88,7 +88,7 @@ fetching and simple saving. *** Metadata sources **** Scraper -* Backlog [0/25] :vrobbler:project:personal: +* Backlog [1/25] :vrobbler:project:personal: ** TODO [#C] After transition to linux add curl_cffi as webpage scrapper again :webpages:metadata: ** TODO [#C] Create small utility to clean up tracks scrobbled with wonky playback times :bug:music:scrobbles: :PROPERTIES: @@ -580,18 +580,6 @@ named constants for maintainability. - ~vrobbler/apps/scrobbles/importers/tsv.py~ (line 55) -- ="S"= completion status -** TODO [#A] Deduplicate BGG plays before posting :boardgames:bgg:duplication: -:PROPERTIES: -:ID: e9b842bf-0049-42e7-a060-f3ebd0067d2f -:END: - -*** Description - -No check for existing BGG plays before posting, which can create duplicates. -Should look up past plays by =bggeek_id= first. - -File: ~vrobbler/apps/boardgames/bgg.py~ (line 117) - ** TODO [#B] Is there way to create unique slugs for media instances :media_types: ** TODO [#A] Update how board game scrobbles work :boardgames: @@ -607,8 +595,21 @@ The Edit log form should have from top to bottom: - People (which should be similar to the Bird widget on BirdLocation and allow setting per user score, win true/false, rank, new true/false, seat_ordrer) - Expansion ids (which should a multi-select widget of expansions for this game) - Location (which should be a drop down of BoardGameLocations for this user) - +** DONE [#A] Add ability to push one or sync all board game scrobbles to BGG :boardgames: +:PROPERTIES: +:ID: 1f306552-35e8-4a75-a661-b0956e8de967 +:END: + +*** Description + +We have a half-baked BGG sync module. I don't even know it if works. Can we +clean it up, add some tests and then add UI features to either sync a single +board game scrobble (via the detail page) to BGG, along with a "sync all" button +on the board game list page. Additionally, the sync should merge data between +Vrobbler and BGG based on timestamp and bgg_id. Priority should be given to data +from Vrobbler. + ** DONE [#C] Clean up naming of =bgsplay= parsing :importers:refactoring: :PROPERTIES: :ID: c751dbbc-464a-4e63-9fe3-e034303f7b54 diff --git a/vrobbler/apps/boardgames/bgg.py b/vrobbler/apps/boardgames/bgg.py index 94802fc..42be4bd 100644 --- a/vrobbler/apps/boardgames/bgg.py +++ b/vrobbler/apps/boardgames/bgg.py @@ -1,12 +1,11 @@ -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 +from django.contrib.auth import get_user_model User = get_user_model() if TYPE_CHECKING: @@ -101,48 +100,124 @@ def lookup_boardgame_from_bgg(lookup_id: str) -> dict: return game_dict -def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]: - bgg_username = "secstate" # user.profile.bgg_username - bgg_password = "yYFCKnfo8AK89lc68q0S" +def fetch_existing_bgg_plays(bgg_username: str) -> list[dict]: + """Fetch recent plays from BGG for a user to check for duplicates.""" + url = f"https://boardgamegeek.com/xmlapi2/plays?username={bgg_username}&page=1" + r = requests.get(url, headers=BASE_HEADERS) + if r.status_code != 200: + logger.error( + "Failed to fetch BGG plays", + extra={"status_code": r.status_code, "username": bgg_username}, + ) + return [] - if not bgg_username or bgg_password: - return + soup = BeautifulSoup(r.text, "xml") + plays = [] + for play_elem in soup.findAll("play"): + item = play_elem.find("item") + if item: + plays.append( + { + "objectid": item.get("objectid"), + "date": play_elem.get("date"), + "quantity": play_elem.get("quantity"), + "length": play_elem.get("length"), + } + ) + return plays + + +def is_duplicate_bgg_play( + existing_plays: list[dict], bgg_id: str, play_date: str +) -> bool: + """Check if a play already exists on BGG matching game ID and date.""" + for play in existing_plays: + if play["objectid"] == str(bgg_id) and play["date"] == play_date: + return True + return False + + +def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]: + from boardgames.models import BoardGameLocation + from people.models import Person + + bgg_username = user.profile.bgg_username if hasattr(user, "profile") else None + bgg_password = getattr(settings, "BGG_PASSWORD", "") + + if not bgg_username or not bgg_password: + logger.warning( + "Cannot push to BGG: missing credentials", + extra={"user_id": user.id, "bgg_username": bgg_username}, + ) + return None + + play_date = scrobble.timestamp.date().strftime("%Y-%m-%d") + bgg_id = scrobble.media_obj.bggeek_id + + existing_plays = fetch_existing_bgg_plays(bgg_username) + if is_duplicate_bgg_play(existing_plays, bgg_id, play_date): + logger.info( + "Skipping duplicate BGG play", + extra={ + "bgg_id": bgg_id, + "play_date": play_date, + "scrobble_id": scrobble.id, + }, + ) + return None login_payload = { "credentials": {"username": bgg_username, "password": bgg_password} } - headers = BASE_HEADERS + headers = BASE_HEADERS.copy() 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, ) + if p.status_code != 200: + logger.error( + "BGG login failed", + extra={"status_code": p.status_code, "response": p.text}, + ) + return None + players = [] if scrobble.log: - for player in scrobble.log.get("players"): + 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) + if not player_person: + continue + + player_data = { + "name": player_person.name, + "win": player.get("win", False), + "new": player.get("new", False), + "score": player.get("score"), + } + if player_person.bgg_username: + player_data["username"] = player_person.bgg_username + players.append(player_data) + + location = None + if scrobble.log and scrobble.log.get("location_id"): + boardgame_location = BoardGameLocation.objects.filter( + id=scrobble.log["location_id"] + ).first() + if boardgame_location: + location = boardgame_location.name play_payload = { - "playdate": scrobble.timestamp.date.strftime("%Y-%m-%d"), - "length": scrobble.playback_position_seconds / 60, + "playdate": play_date, + "length": (scrobble.playback_position_seconds or 0) / 60, "comments": "Uploaded from Vrobbler", - "location": scrobble.log.location or None, - "objectid": scrobble.media_obj.bggeek_id, + "location": location, + "objectid": bgg_id, "quantity": "1", "action": "save", "players": players, @@ -154,3 +229,4 @@ def push_scrobble_to_bgg(scrobble: "Scrobble", user: User) -> Optional[bool]: data=json.dumps(play_payload), headers=headers, ) + return r.status_code == 200