15831523521

Committed 23 Jun 2025 05:54PM UTC coverage: 56.215%. Remained the same

Build # 15831523521

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare release (#131)

Run Details

1 of 3 new or added lines in 2 files covered. (33.33%)

398 of 708 relevant lines covered (56.21%)

1.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

17.58

/src/sabogaapi/api_v1/scraper/_update.py

import time
from datetime import datetime
from xml.etree import ElementTree

import requests
from PIL import Image
from pydantic import BaseModel

from sabogaapi.api_v1.config import settings
from sabogaapi.api_v1.database import init_db
from sabogaapi.api_v1.models import (
    Boardgame,
    Category,
    Designer,
    Family,
    Mechanic,
    RankHistory,
)
from sabogaapi.api_v1.scraper._utilities import parse_boardgame_data, scrape_api
from sabogaapi.logger import configure_logger

logger = configure_logger()


class BoardgameBGGIDs(BaseModel):
    bgg_id: int


async def analyse_api_response(
    item: ElementTree.Element,
) -> tuple[None, None] | tuple[Boardgame, None] | tuple[Boardgame, RankHistory]:
    data = parse_boardgame_data(item)

    # Get boardgame from database or create a new one
    boardgame = await Boardgame.find_one(Boardgame.bgg_id == data["bgg_id"])
    if data["rank"] is None:
        if boardgame is not None:
            await boardgame.delete()
            logger.info(f"Deleted boardgame {data['bgg_id']} due to missing rank.")
        return None, None  # Don't process further

    if boardgame is None:
        boardgame = Boardgame(bgg_id=data["bgg_id"])

    # Simple fields
    boardgame.name = data["name"]
    boardgame.description = data["description"]
    boardgame.year_published = data["year_published"]
    boardgame.minplayers = data["minplayers"]
    boardgame.maxplayers = data["maxplayers"]
    boardgame.playingtime = data["playingtime"]
    boardgame.minplaytime = data["minplaytime"]
    boardgame.maxplaytime = data["maxplaytime"]

    # Process image
    if data["image_url"]:
        image_filename = f"{data['bgg_id']}.jpg"
        image_file = settings.img_dir / image_filename
        if not image_file.exists():
            img_request = requests.get(data["image_url"])
            if img_request.status_code == 200:
                with open(image_file, "wb") as handler:
                    handler.write(img_request.content)

        thumbnail_filename = f"{image_file.stem}-thumbnail.jpg"
        thumbnail_file = settings.img_dir / thumbnail_filename
        if not thumbnail_file.exists():
            im = Image.open(image_file).convert("RGB")
            im.thumbnail((128, 128))
            im.save(thumbnail_file)

        boardgame.image_url = f"/img/{image_filename}"
        boardgame.thumbnail_url = f"/img/{thumbnail_filename}"

    # Process categories
    category_names_ids = data["categories"]
    if category_names_ids:
        boardgame.categories = [
            Category(name=name, bgg_id=bgg_id) for bgg_id, name in category_names_ids
        ]

    # Process families
    family_names_ids = data["families"]
    if family_names_ids:
        boardgame.families = [
            Family(name=name, bgg_id=bgg_id) for bgg_id, name in family_names_ids
        ]

    # Process mechanics
    mechanic_names_ids = data["mechanics"]
    if mechanic_names_ids:
        boardgame.mechanics = [
            Mechanic(name=name, bgg_id=bgg_id) for bgg_id, name in mechanic_names_ids
        ]

    # Process designers
    designer_names_ids = data["designers"]
    if designer_names_ids:
        boardgame.designers = [
            Designer(name=name, bgg_id=bgg_id) for bgg_id, name in designer_names_ids
        ]

    # Process rank history
    last_rank_history = (
        await RankHistory.find({"bgg_id": boardgame.bgg_id})
        .sort(-RankHistory.date)
        .limit(1)
        .first_or_none()
    )
    if last_rank_history and (datetime.now() - last_rank_history.date).days < 1:
        return boardgame, None

    boardgame.bgg_rank = data["rank"]
    boardgame.bgg_geek_rating = data["geek_rating"]
    boardgame.bgg_average_rating = data["average_rating"]
    rank_history = RankHistory(
        bgg_id=boardgame.bgg_id,
        bgg_rank=data["rank"],
        bgg_geek_rating=data["geek_rating"],
        bgg_average_rating=data["average_rating"],
        date=datetime.today(),
    )

    return boardgame, rank_history


async def ascrape_update(step: int) -> None:
    await init_db()
    run_index = 0
    while True:
        ids = (
            await Boardgame.find_all()
            .project(BoardgameBGGIDs)
            .sort("-bgg_id")
            .skip(run_index * step)
            .limit(step)
            .to_list()
        )
        ids_int = [x.bgg_id for x in ids]

        if len(ids) == 0:
            break
        logger.info(f"Scraping {ids_int}.")
        parsed_xml = scrape_api(ids_int)
        if parsed_xml:
            items = parsed_xml.findall("item")
            rank_histories = []
            for item in items:
                boardgame, rank_history = await analyse_api_response(item)
                if boardgame:
                    await boardgame.save()
                if rank_history:
                    rank_histories.append(rank_history)
            if rank_histories:
                await RankHistory.insert_many(rank_histories)
        run_index += 1
        time.sleep(5)

1	import time	2✔
2	from datetime import datetime	2✔
3	from xml.etree import ElementTree	2✔
4
5	import requests	2✔
6	from PIL import Image	2✔
7	from pydantic import BaseModel	2✔
8
9	from sabogaapi.api_v1.config import settings	2✔
10	from sabogaapi.api_v1.database import init_db	2✔
11	from sabogaapi.api_v1.models import (	2✔
12	Boardgame,
13	Category,
14	Designer,
15	Family,
16	Mechanic,
17	RankHistory,
18	)
19	from sabogaapi.api_v1.scraper._utilities import parse_boardgame_data, scrape_api	2✔
20	from sabogaapi.logger import configure_logger	2✔
21
22	logger = configure_logger()	2✔
23
24
25	class BoardgameBGGIDs(BaseModel):	2✔
26	bgg_id: int	2✔
27
28
29	async def analyse_api_response(	2✔
30	item: ElementTree.Element,
31	) -> tuple[None, None] \| tuple[Boardgame, None] \| tuple[Boardgame, RankHistory]:
32	data = parse_boardgame_data(item)	×
33
34	# Get boardgame from database or create a new one
35	boardgame = await Boardgame.find_one(Boardgame.bgg_id == data["bgg_id"])	×
36	if data["rank"] is None:	×
37	if boardgame is not None:	×
38	await boardgame.delete()	×
39	logger.info(f"Deleted boardgame {data['bgg_id']} due to missing rank.")	×
40	return None, None # Don't process further	×
41
42	if boardgame is None:	×
43	boardgame = Boardgame(bgg_id=data["bgg_id"])	×
44
45	# Simple fields
46	boardgame.name = data["name"]	×
47	boardgame.description = data["description"]	×
48	boardgame.year_published = data["year_published"]	×
49	boardgame.minplayers = data["minplayers"]	×
50	boardgame.maxplayers = data["maxplayers"]	×
51	boardgame.playingtime = data["playingtime"]	×
52	boardgame.minplaytime = data["minplaytime"]	×
53	boardgame.maxplaytime = data["maxplaytime"]	×
54
55	# Process image
56	if data["image_url"]:	×
57	image_filename = f"{data['bgg_id']}.jpg"	×
NEW 58	image_file = settings.img_dir / image_filename	×
59	if not image_file.exists():	×
60	img_request = requests.get(data["image_url"])	×
61	if img_request.status_code == 200:	×
62	with open(image_file, "wb") as handler:	×
63	handler.write(img_request.content)	×
64
65	thumbnail_filename = f"{image_file.stem}-thumbnail.jpg"	×
NEW 66	thumbnail_file = settings.img_dir / thumbnail_filename	×
67	if not thumbnail_file.exists():	×
68	im = Image.open(image_file).convert("RGB")	×
69	im.thumbnail((128, 128))	×
70	im.save(thumbnail_file)	×
71
72	boardgame.image_url = f"/img/{image_filename}"	×
73	boardgame.thumbnail_url = f"/img/{thumbnail_filename}"	×
74
75	# Process categories
76	category_names_ids = data["categories"]	×
77	if category_names_ids:	×
78	boardgame.categories = [	×
79	Category(name=name, bgg_id=bgg_id) for bgg_id, name in category_names_ids
80	]
81
82	# Process families
83	family_names_ids = data["families"]	×
84	if family_names_ids:	×
85	boardgame.families = [	×
86	Family(name=name, bgg_id=bgg_id) for bgg_id, name in family_names_ids
87	]
88
89	# Process mechanics
90	mechanic_names_ids = data["mechanics"]	×
91	if mechanic_names_ids:	×
92	boardgame.mechanics = [	×
93	Mechanic(name=name, bgg_id=bgg_id) for bgg_id, name in mechanic_names_ids
94	]
95
96	# Process designers
97	designer_names_ids = data["designers"]	×
98	if designer_names_ids:	×
99	boardgame.designers = [	×
100	Designer(name=name, bgg_id=bgg_id) for bgg_id, name in designer_names_ids
101	]
102
103	# Process rank history
104	last_rank_history = (	×
105	await RankHistory.find({"bgg_id": boardgame.bgg_id})
106	.sort(-RankHistory.date)
107	.limit(1)
108	.first_or_none()
109	)
110	if last_rank_history and (datetime.now() - last_rank_history.date).days < 1:	×
111	return boardgame, None	×
112
113	boardgame.bgg_rank = data["rank"]	×
114	boardgame.bgg_geek_rating = data["geek_rating"]	×
115	boardgame.bgg_average_rating = data["average_rating"]	×
116	rank_history = RankHistory(	×
117	bgg_id=boardgame.bgg_id,
118	bgg_rank=data["rank"],
119	bgg_geek_rating=data["geek_rating"],
120	bgg_average_rating=data["average_rating"],
121	date=datetime.today(),
122	)
123
124	return boardgame, rank_history	×
125
126
127	async def ascrape_update(step: int) -> None:	2✔
128	await init_db()	×
129	run_index = 0	×
130	while True:	×
131	ids = (	×
132	await Boardgame.find_all()
133	.project(BoardgameBGGIDs)
134	.sort("-bgg_id")
135	.skip(run_index * step)
136	.limit(step)
137	.to_list()
138	)
139	ids_int = [x.bgg_id for x in ids]	×
140
141	if len(ids) == 0:	×
142	break	×
143	logger.info(f"Scraping {ids_int}.")	×
144	parsed_xml = scrape_api(ids_int)	×
145	if parsed_xml:	×
146	items = parsed_xml.findall("item")	×
147	rank_histories = []	×
148	for item in items:	×
149	boardgame, rank_history = await analyse_api_response(item)	×
150	if boardgame:	×
151	await boardgame.save()	×
152	if rank_history:	×
153	rank_histories.append(rank_history)	×
154	if rank_histories:	×
155	await RankHistory.insert_many(rank_histories)	×
156	run_index += 1	×
157	time.sleep(5)	×

Ruberhauptmann / saboga-api / 15831523521

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous