[boardgames] Fix and clean up BGG push integration
This commit is contained in:
29
PROJECT.org
29
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
|
||||
|
||||
@ -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
|
||||
|
||||
Reference in New Issue
Block a user