14009332806

Committed 22 Mar 2025 02:11PM UTC coverage: 64.912% (+0.4%) from 64.5%

Build # 14009332806

Build Type

push

github

Committed by

web-flow

Commit Message

Scrape images (#92)

Run Details

29 of 38 new or added lines in 5 files covered. (76.32%)

1 existing line in 1 file now uncovered.

148 of 228 relevant lines covered (64.91%)

0.65 hits per line

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

60.27

/src/sabogaapi/api_v1/models.py

"""Beanie database models."""

import datetime
from typing import Annotated, List

from beanie import Document, Indexed
from pydantic import BaseModel


class BoardgameComparison(BaseModel):
    bgg_id: int
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    bgg_rank: int
    bgg_rank_change: int
    bgg_geek_rating: float
    bgg_geek_rating_change: float
    bgg_average_rating: float
    bgg_average_rating_change: float


class BoardgameWithHistoricalData(BaseModel):
    bgg_id: int
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    bgg_rank: int
    bgg_geek_rating: float
    bgg_average_rating: float
    bgg_rank_history: List["RankHistory"]


class RankHistory(BaseModel):
    date: datetime.datetime
    bgg_rank: int | None
    bgg_geek_rating: float | None
    bgg_average_rating: float | None


class Boardgame(Document):
    bgg_id: Annotated[int, Indexed(unique=True)]
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    bgg_rank_history: List["RankHistory"] = []

    @staticmethod
    async def get_top_ranked_boardgames(
        date: datetime.datetime,
        compare_to: datetime.datetime,
        page: int = 1,
        page_size: int = 10,
    ) -> List[BoardgameComparison]:
        pipeline = [
            {
                "$addFields": {
                    "current_rank_data": {
                        "$arrayElemAt": [
                            {
                                "$filter": {
                                    "input": "$bgg_rank_history",
                                    "as": "history",
                                    "cond": {"$lte": ["$$history.date", date]},
                                }
                            },
                            -1,
                        ]
                    }
                }
            },
            {"$match": {"current_rank_data.bgg_rank": {"$ne": None}}},
            {"$sort": {"current_rank_data.bgg_rank": 1}},
            {"$skip": (page - 1) * page_size},
            {"$limit": page_size},
            {
                "$addFields": {
                    "previous_rank_data": {
                        "$arrayElemAt": [
                            {
                                "$filter": {
                                    "input": "$bgg_rank_history",
                                    "as": "history",
                                    "cond": {"$lte": ["$$history.date", compare_to]},
                                }
                            },
                            -1,
                        ]
                    },
                }
            },
            {
                "$addFields": {
                    "bgg_rank": "$current_rank_data.bgg_rank",
                    "bgg_rank_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$previous_rank_data.bgg_rank",
                                    "$current_rank_data.bgg_rank",
                                ]
                            },
                            0,
                        ]
                    },
                    "bgg_geek_rating": "$current_rank_data.bgg_geek_rating",
                    "bgg_geek_rating_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$current_rank_data.bgg_geek_rating",
                                    "$previous_rank_data.bgg_geek_rating",
                                ]
                            },
                            0,
                        ]
                    },
                    "bgg_average_rating": "$current_rank_data.bgg_average_rating",
                    "bgg_average_rating_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$current_rank_data.bgg_average_rating",
                                    "$previous_rank_data.bgg_average_rating",
                                ]
                            },
                            0,
                        ]
                    },
                }
            },
        ]

        rank_data = await Boardgame.aggregate(
            aggregation_pipeline=pipeline, projection_model=BoardgameComparison
        ).to_list()

        return rank_data

    @staticmethod
    async def get_boardgame_with_historical_data(
        bgg_id: int,
        start_date: datetime.datetime,
        end_date: datetime.datetime,
        mode: str,
    ) -> BoardgameWithHistoricalData | None:
        date_diff = (end_date - start_date).days
        if mode == "auto":
            if date_diff <= 30:
                mode = "daily"
            elif date_diff <= 180:
                mode = "weekly"
            else:
                mode = "yearly"

        pipeline = [
            {"$match": {"bgg_id": bgg_id}},  # Match board game by ID
            {"$unwind": "$bgg_rank_history"},  # Unwind the bgg_rank_history array
            {
                "$match": {  # Filter rank history by date range
                    "bgg_rank_history.date": {"$gte": start_date, "$lte": end_date}
                }
            },
            {"$sort": {"bgg_rank_history.date": 1}},  # Sort by date (ascending)
            {
                "$group": {
                    "_id": "$bgg_id",
                    "latest_rank": {
                        "$last": "$bgg_rank_history"
                    },  # Get latest rank entry
                    "bgg_rank_history": {
                        "$push": "$bgg_rank_history"
                    },  # Keep all history
                    "doc": {"$first": "$$ROOT"},  # Store the full document
                }
            },
            {
                "$project": {
                    "_id": 0,
                    **{"bgg_id": "$_id"},  # Include the original _id as bgg_id
                    **{"bgg_rank": "$latest_rank.bgg_rank"},
                    **{"bgg_geek_rating": "$latest_rank.bgg_geek_rating"},
                    **{"bgg_average_rating": "$latest_rank.bgg_average_rating"},
                    **{"bgg_rank_history": 1},
                    **{"doc": 1},  # Include all other fields from the original document
                }
            },
            {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$doc", "$$ROOT"]}}},
        ]

        result = await Boardgame.aggregate(
            aggregation_pipeline=pipeline, projection_model=BoardgameWithHistoricalData
        ).to_list()

        if not result:
            return None

        boardgame_data = result[0]

        # Apply mode filtering to bgg_rank_history
        history = boardgame_data.bgg_rank_history
        if mode == "weekly":
            history = history[::7]  # Every 7th entry
        elif mode == "yearly":
            seen_years = set()
            yearly_history = []
            for entry in history:
                year = entry.date.year
                if year not in seen_years:
                    yearly_history.append(entry)
                    seen_years.add(year)
            history = yearly_history

        boardgame_data.bgg_rank_history = history
        return boardgame_data

    class Settings:
        name = "boardgames"

1	"""Beanie database models."""
2
3	import datetime	1✔
4	from typing import Annotated, List	1✔
5
6	from beanie import Document, Indexed	1✔
7	from pydantic import BaseModel	1✔
8
9
10	class BoardgameComparison(BaseModel):	1✔
11	bgg_id: int	1✔
12	name: str = ""	1✔
13	description: str \| None = None	1✔
14	image_url: str \| None = None	1✔
15	thumbnail_url: str \| None = None	1✔
16	bgg_rank: int	1✔
17	bgg_rank_change: int	1✔
18	bgg_geek_rating: float	1✔
19	bgg_geek_rating_change: float	1✔
20	bgg_average_rating: float	1✔
21	bgg_average_rating_change: float	1✔
22
23
24	class BoardgameWithHistoricalData(BaseModel):	1✔
25	bgg_id: int	1✔
26	name: str = ""	1✔
27	description: str \| None = None	1✔
28	image_url: str \| None = None	1✔
29	thumbnail_url: str \| None = None	1✔
30	bgg_rank: int	1✔
31	bgg_geek_rating: float	1✔
32	bgg_average_rating: float	1✔
33	bgg_rank_history: List["RankHistory"]	1✔
34
35
36	class RankHistory(BaseModel):	1✔
37	date: datetime.datetime	1✔
38	bgg_rank: int \| None	1✔
39	bgg_geek_rating: float \| None	1✔
40	bgg_average_rating: float \| None	1✔
41
42
43	class Boardgame(Document):	1✔
44	bgg_id: Annotated[int, Indexed(unique=True)]	1✔
45	name: str = ""	1✔
46	description: str \| None = None	1✔
47	image_url: str \| None = None	1✔
48	thumbnail_url: str \| None = None	1✔
49	bgg_rank_history: List["RankHistory"] = []	1✔
50
51	@staticmethod	1✔
52	async def get_top_ranked_boardgames(	1✔
53	date: datetime.datetime,
54	compare_to: datetime.datetime,
55	page: int = 1,
56	page_size: int = 10,
57	) -> List[BoardgameComparison]:
58	pipeline = [	×
59	{
60	"$addFields": {
61	"current_rank_data": {
62	"$arrayElemAt": [
63	{
64	"$filter": {
65	"input": "$bgg_rank_history",
66	"as": "history",
67	"cond": {"$lte": ["$$history.date", date]},
68	}
69	},
70	-1,
71	]
72	}
73	}
74	},
75	{"$match": {"current_rank_data.bgg_rank": {"$ne": None}}},
76	{"$sort": {"current_rank_data.bgg_rank": 1}},
77	{"$skip": (page - 1) * page_size},
78	{"$limit": page_size},
79	{
80	"$addFields": {
81	"previous_rank_data": {
82	"$arrayElemAt": [
83	{
84	"$filter": {
85	"input": "$bgg_rank_history",
86	"as": "history",
87	"cond": {"$lte": ["$$history.date", compare_to]},
88	}
89	},
90	-1,
91	]
92	},
93	}
94	},
95	{
96	"$addFields": {
97	"bgg_rank": "$current_rank_data.bgg_rank",
98	"bgg_rank_change": {
99	"$ifNull": [
100	{
101	"$subtract": [
102	"$previous_rank_data.bgg_rank",
103	"$current_rank_data.bgg_rank",
104	]
105	},
106	0,
107	]
108	},
109	"bgg_geek_rating": "$current_rank_data.bgg_geek_rating",
110	"bgg_geek_rating_change": {
111	"$ifNull": [
112	{
113	"$subtract": [
114	"$current_rank_data.bgg_geek_rating",
115	"$previous_rank_data.bgg_geek_rating",
116	]
117	},
118	0,
119	]
120	},
121	"bgg_average_rating": "$current_rank_data.bgg_average_rating",
122	"bgg_average_rating_change": {
123	"$ifNull": [
124	{
125	"$subtract": [
126	"$current_rank_data.bgg_average_rating",
127	"$previous_rank_data.bgg_average_rating",
128	]
129	},
130	0,
131	]
132	},
133	}
134	},
135	]
136
137	rank_data = await Boardgame.aggregate(	×
138	aggregation_pipeline=pipeline, projection_model=BoardgameComparison
139	).to_list()
140
141	return rank_data	×
142
143	@staticmethod	1✔
144	async def get_boardgame_with_historical_data(	1✔
145	bgg_id: int,
146	start_date: datetime.datetime,
147	end_date: datetime.datetime,
148	mode: str,
149	) -> BoardgameWithHistoricalData \| None:
UNCOV 150	date_diff = (end_date - start_date).days	×
151	if mode == "auto":	×
152	if date_diff <= 30:	×
153	mode = "daily"	×
154	elif date_diff <= 180:	×
155	mode = "weekly"	×
156	else:
157	mode = "yearly"	×
158
159	pipeline = [	×
160	{"$match": {"bgg_id": bgg_id}}, # Match board game by ID
161	{"$unwind": "$bgg_rank_history"}, # Unwind the bgg_rank_history array
162	{
163	"$match": { # Filter rank history by date range
164	"bgg_rank_history.date": {"$gte": start_date, "$lte": end_date}
165	}
166	},
167	{"$sort": {"bgg_rank_history.date": 1}}, # Sort by date (ascending)
168	{
169	"$group": {
170	"_id": "$bgg_id",
171	"latest_rank": {
172	"$last": "$bgg_rank_history"
173	}, # Get latest rank entry
174	"bgg_rank_history": {
175	"$push": "$bgg_rank_history"
176	}, # Keep all history
177	"doc": {"$first": "$$ROOT"}, # Store the full document
178	}
179	},
180	{
181	"$project": {
182	"_id": 0,
183	**{"bgg_id": "$_id"}, # Include the original _id as bgg_id
184	**{"bgg_rank": "$latest_rank.bgg_rank"},
185	**{"bgg_geek_rating": "$latest_rank.bgg_geek_rating"},
186	**{"bgg_average_rating": "$latest_rank.bgg_average_rating"},
187	**{"bgg_rank_history": 1},
188	**{"doc": 1}, # Include all other fields from the original document
189	}
190	},
191	{"$replaceRoot": {"newRoot": {"$mergeObjects": ["$doc", "$$ROOT"]}}},
192	]
193
194	result = await Boardgame.aggregate(	×
195	aggregation_pipeline=pipeline, projection_model=BoardgameWithHistoricalData
196	).to_list()
197
198	if not result:	×
199	return None	×
200
201	boardgame_data = result[0]	×
202
203	# Apply mode filtering to bgg_rank_history
204	history = boardgame_data.bgg_rank_history	×
205	if mode == "weekly":	×
206	history = history[::7] # Every 7th entry	×
207	elif mode == "yearly":	×
208	seen_years = set()	×
209	yearly_history = []	×
210	for entry in history:	×
211	year = entry.date.year	×
212	if year not in seen_years:	×
213	yearly_history.append(entry)	×
214	seen_years.add(year)	×
215	history = yearly_history	×
216
217	boardgame_data.bgg_rank_history = history	×
218	return boardgame_data	×
219
220	class Settings:	1✔
221	name = "boardgames"	1✔

Ruberhauptmann / saboga-api / 14009332806

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