bgg_api/bgg_connection.py

import requests
import xml.etree.ElementTree as ET
from pydantic import HttpUrl
import requests

from classes.boardgame import BoardGame, BoardGameExpansion
import auth_manager

authenticated_session: requests.Session = requests.Session()

def url_to_xml_object(url: HttpUrl) -> ET.Element:
    r = authenticated_session.get(url)
    assert r.status_code == 200
    root = ET.fromstring(r.content)
    return root

def get_boardgame(boardgame_id: int) -> BoardGame:
    url : str = "https://boardgamegeek.com/xmlapi2/thing?id={}&stats=true".format(boardgame_id)
    boardgame_xml_object : ET.Element = url_to_xml_object(url)

    requested_boardgame = convert_xml_to_boardgame(boardgame_xml_object.find('item'))

    return requested_boardgame

def get_multiple_boardgames(boardgame_ids: list[int]) -> list[BoardGame]:

    def divide_list_in_chunks(list_to_divide: list[int], chunk_size: int = 20): 
        for i in range(0, len(list_to_divide), chunk_size):  
            yield list_to_divide[i:i + chunk_size] 

    
    boardgame_list_to_return: list[BoardGame] = []

    #Boardgamegeek only allows chunks of 20 boardgames at a time
    boardgame_ids_divided = list(divide_list_in_chunks(boardgame_ids))

    for boardgame_id_list_20 in boardgame_ids_divided:
        boardgame_id_list_commas: str = ','.join(boardgame_id_list_20)
        url : str = "https://boardgamegeek.com/xmlapi2/thing?id={}&stats=true".format(boardgame_id_list_commas)
        print(url)
        boardgame_xml_object : ET.Element = url_to_xml_object(url)

        for boardgame in boardgame_xml_object:
            requested_boardgame = convert_xml_to_boardgame(boardgame_xml_object.find('item'))
            boardgame_list_to_return.append(requested_boardgame)

    return boardgame_list_to_return



#Requires single boardgame XML 'item' from bgg api on /thing
def convert_xml_to_boardgame(boardgame_xml: ET.Element) -> BoardGame:

    boardgame_type = boardgame_xml.get('type')

    boardgame_dict = {
        "id" : int(boardgame_xml.get('id')),
        "name" : boardgame_xml.find('name').get('value'),
        "description" : boardgame_xml.find('description').text,
        "image_url" : boardgame_xml.find('image').text,
        "thumbnail_url" : boardgame_xml.find('thumbnail').text,
        "year_published" : int(boardgame_xml.find('yearpublished').get('value')),
        "min_players" : int(boardgame_xml.find('minplayers').get('value')),
        "max_players" : int(boardgame_xml.find('maxplayers').get('value')),
        "min_playing_time" : int(boardgame_xml.find('minplaytime').get('value')),
        "max_playing_time" : int(boardgame_xml.find('maxplaytime').get('value')),
        "min_age" : int(boardgame_xml.find('minage').get('value')),
        "all_expansion_ids" : [0,1,2,3]
    }

    match boardgame_type:
        case "boardgame":
            boardgame = BoardGame(**boardgame_dict)
        case "boardgameexpansion":
            boardgame = BoardGameExpansion(**boardgame_dict)

    

    return boardgame

def convert_collection_xml_to_boardgame(boardgame_xml: ET.Element) -> BoardGame:

    boardgame_extra: BoardGame = get_boardgame(boardgame_xml.get('objectid'))

    boardgame_type = boardgame_xml.get('subtype')

    boardgame_dict = {
        "id" : boardgame_extra.id,
        "name" : boardgame_extra.name,
        "description" : boardgame_extra.description,
        "image_url" : boardgame_extra.image_url,
        "thumbnail_url" : boardgame_extra.thumbnail_url,
        "year_published" : boardgame_extra.year_published,
        "min_players" : boardgame_extra.min_players,
        "max_players" : boardgame_extra.max_players,
        "min_playing_time" : boardgame_extra.min_playing_time,
        "max_playing_time" : boardgame_extra.max_playing_time,
        "min_age" : boardgame_extra.min_age,
        "all_expansion_ids" : [0,1,2,3]
    }

    match boardgame_type:
        case "boardgame":
            boardgame = BoardGame(**boardgame_dict)
        case "boardgameexpansion":
            boardgame = BoardGameExpansion(**boardgame_dict)

    

    return boardgame

def get_user_owned_collection() -> list[BoardGame]:
    url = 'https://boardgamegeek.com/xmlapi2/collection?username={}&own=1&stats=1&excludesubtype=boardgameexpansion&showprivate=1&version=1'.format(auth_manager.username)
    collection_xml = url_to_xml_object(url)

    print(collection_xml.tag)

    owned_collection_list: list[BoardGame] = []

    for boardgame_item in collection_xml:
        boardgame = convert_collection_xml_to_boardgame(boardgame_item)
        owned_collection_list.append(boardgame)

    return owned_collection_list


def load_authenticated_bgg_session(username: str, password: str) -> requests.Session:
    global authenticated_session

    login_url = "https://boardgamegeek.com/login/api/v1"

    post_data = {
        "credentials":{
            "username": username,
            "password": password
        }
    }


    assert len(authenticated_session.cookies) == 0, 'Session already exists'

    login_response = authenticated_session.post(login_url, json=post_data)

    assert login_response.status_code == 204, "Login failed!"
   
load_authenticated_bgg_session(auth_manager.username, auth_manager.password)