cal/paper-dynasty-database
1
0
Fork 0
Code Issues 10 Pull Requests 5 Actions Packages Projects Releases Wiki Activity

feat: card-of-the-week endpoint for automated content pipeline #212

Open
cal wants to merge 2 commits from autonomous/feat-card-of-the-week-endpoint into main
pull from: autonomous/feat-card-of-the-week-endpoint
merge into: cal:main
Conversation 3 Commits 2 Files Changed 3 +600
3 changed files with 600 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/main.py
View File

@ -53,6 +53,7 @@ from .routers_v2 import ( # noqa: E402
scout_claims,
refractor,
season_stats,
featured,
)
@ -109,6 +110,7 @@ app.include_router(scout_opportunities.router)
app.include_router(scout_claims.router)
app.include_router(refractor.router)
app.include_router(season_stats.router)
app.include_router(featured.router)
@app.middleware("http")

129
app/routers_v2/featured.py Normal file
View File

@ -0,0 +1,129 @@
"""
Router: featured cards
Provides endpoints that surface curated "featured" card selections for use
in automated content pipelines (n8n, Discord scheduled posts, etc.).
Endpoints
---------
GET /api/v2/cards/featured/card-of-the-week
Returns the single highest-rated eligible card created in the past N days,
with all fields needed to build a Discord embed in one call. Deterministic
for a given calendar day (cache-friendly).
Design notes
------------
- "Card created" is approximated by Pack.open_time — the timestamp when the
pack containing the card was opened. The Card model has no created_at
column; open_time is the closest available proxy and is semantically correct
(the card enters a team's collection at pack-open time).
- Rating is Card.value (the integer rating field set at card issuance).
- Tiebreak: highest value desc, then newest pack open_time desc, then
highest card.id desc — fully deterministic with no random element.
- AI teams (Team.is_ai is not NULL and != 0) are excluded by default.
Pass ?include_ai=true to lift that filter.
"""
from datetime import datetime, timedelta
from typing import Optional
import pydantic
from fastapi import APIRouter, HTTPException, Query
from ..db_engine import Card, Cardset, Pack, Player, Rarity, Team
router = APIRouter(prefix="/api/v2/cards/featured", tags=["featured"])
class CardOfTheWeekResponse(pydantic.BaseModel):
"""Single-call Discord embed payload for the card-of-the-week feature.
All fields needed for a rich embed are present so the consumer (n8n,
Discord bot) never needs a follow-up API call.
"""
card_id: int
player_name: str
team_name: str
team_abbrev: str
rarity_name: str
rating: int
card_image_url: str
cardset_name: str
card_created_at: Optional[str] = (
None # ISO-8601 string; None when pack has no open_time
)
@router.get(
"/card-of-the-week",
response_model=CardOfTheWeekResponse,
summary="Card of the Week",
description=(
"Returns the highest-rated card created (pack opened) within the last N days. "
"Excludes AI-team cards by default. Deterministic within a calendar day."
),
)
async def card_of_the_week(
days: int = Query(default=7, ge=1, le=90, description="Look-back window in days"),
include_ai: bool = Query(
default=False,
description="Include cards owned by AI teams (excluded by default)",
),
):
"""Return the card-of-the-week for use in Discord embeds and n8n automation.
Selection algorithm:
1. Filter cards whose pack was opened within the last `days` days.
2. Exclude AI-team cards unless include_ai=true.
3. Order by Card.value DESC, Pack.open_time DESC, Card.id DESC.
4. Return the first row as a flat Discord-ready payload.
Returns 404 when no eligible card exists in the window.
"""
cutoff = datetime.utcnow() - timedelta(days=days)
query = (
Card.select(Card, Pack, Player, Team, Rarity, Cardset)
.join(Pack, on=(Card.pack == Pack.id))
.switch(Card)
.join(Player, on=(Card.player == Player.player_id))
.join(Rarity, on=(Player.rarity == Rarity.id))
.switch(Player)
.join(Cardset, on=(Player.cardset == Cardset.id))
.switch(Card)
.join(Team, on=(Card.team == Team.id))
.where(Pack.open_time >= cutoff)
.where(Card.player.is_null(False))
.where(Card.team.is_null(False))
.where(Card.pack.is_null(False))
)
if not include_ai:
# is_ai is an IntegerField(null=True); non-AI teams have is_ai=0 or NULL
query = query.where((Team.is_ai == 0) | (Team.is_ai.is_null(True)))
# Deterministic tiebreak: highest rating first, then newest pack, then card id
query = query.order_by(Card.value.desc(), Pack.open_time.desc(), Card.id.desc())
card = query.first()
if card is None:
raise HTTPException(status_code=404, detail="no featured card this week")
pack_open_time = None
if card.pack and card.pack.open_time:
ot = card.pack.open_time
pack_open_time = ot.isoformat() if hasattr(ot, "isoformat") else str(ot)
return CardOfTheWeekResponse(
card_id=card.id,
player_name=card.player.p_name,
team_name=card.team.sname,
team_abbrev=card.team.abbrev,
rarity_name=card.player.rarity.name,
rating=card.value,
card_image_url=card.player.image,
cardset_name=card.player.cardset.name,
card_created_at=pack_open_time,
)

469
tests/test_card_of_the_week.py Normal file
View File

@ -0,0 +1,469 @@
"""Tests for GET /api/v2/cards/featured/card-of-the-week.
What: Verifies the card-of-the-week endpoint returns the highest-rated card
opened (via pack) within the last N days, with all fields needed for a Discord
embed in a single call.
Why: This endpoint feeds the automated content pipeline (n8n / Discord
scheduled posts). Correctness of tiebreak logic, AI-exclusion, and the 404
empty-window case are all contract-level guarantees that must not regress.
Test matrix
-----------
test_happy_path_single_eligible_card
One eligible card in the window — endpoint returns its fields.
test_empty_window_returns_404
No cards in the window — endpoint returns 404 with "no featured card
this week".
test_ai_team_excluded_by_default
AI-team card is present but excluded by default; non-AI card is
returned.
test_include_ai_true_includes_ai_team_card
?include_ai=true lifts the AI exclusion and returns the AI card when it
has the highest rating.
test_tiebreak_highest_rating_wins
Two cards in the window; the one with higher value is returned.
test_tiebreak_same_rating_newest_pack_wins
Two cards with equal value; the one with the newer pack open_time is
returned, proving the secondary sort is deterministic.
All tests use an in-memory SQLite database (no PostgreSQL required) via the
shared TestClient pattern established in test_refractor_image_url.py.
"""
import os
os.environ.setdefault("API_TOKEN", "test-token")
os.environ.setdefault("DATABASE_TYPE", "postgresql")
os.environ.setdefault("POSTGRES_PASSWORD", "test-dummy")
from datetime import datetime, timedelta
import pytest
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from peewee import SqliteDatabase
from app.db_engine import (
BattingSeasonStats,
Card,
Cardset,
Decision,
Event,
MlbPlayer,
Pack,
PackType,
PitchingSeasonStats,
Player,
ProcessedGame,
Rarity,
RefractorBoostAudit,
RefractorCardState,
RefractorTrack,
Roster,
RosterSlot,
ScoutClaim,
ScoutOpportunity,
StratGame,
StratPlay,
Team,
BattingCard,
PitchingCard,
)
# ---------------------------------------------------------------------------
# In-memory SQLite setup — shared named database so TestClient and fixtures
# share the same connection within a test.
# ---------------------------------------------------------------------------
_cotw_db = SqliteDatabase(
"file:cotwtest?mode=memory&cache=shared",
uri=True,
pragmas={"foreign_keys": 1},
)
_COTW_MODELS = [
Rarity,
Event,
Cardset,
MlbPlayer,
Player,
BattingCard,
PitchingCard,
Team,
PackType,
Pack,
Card,
Roster,
RosterSlot,
StratGame,
StratPlay,
Decision,
BattingSeasonStats,
PitchingSeasonStats,
ProcessedGame,
ScoutOpportunity,
ScoutClaim,
RefractorTrack,
RefractorCardState,
RefractorBoostAudit,
]
@pytest.fixture(autouse=False)
def cotw_db():
"""Bind all models to the shared-memory SQLite database.
What: Creates all tables before each test and drops them after, ensuring
test isolation within the same process.
Why: SQLite shared-memory databases persist across tests in the same
process if tables are not explicitly dropped. The autouse=False means
tests must request this fixture (or the cotw_client fixture that depends
on it) explicitly, avoiding interference with the conftest autouse fixture.
"""
_cotw_db.bind(_COTW_MODELS)
_cotw_db.connect(reuse_if_open=True)
_cotw_db.create_tables(_COTW_MODELS)
yield _cotw_db
_cotw_db.drop_tables(list(reversed(_COTW_MODELS)), safe=True)
def _build_cotw_app() -> FastAPI:
"""Minimal FastAPI app with featured router and SQLite middleware."""
from app.routers_v2.featured import router as featured_router
app = FastAPI()
@app.middleware("http")
async def db_middleware(request: Request, call_next):
_cotw_db.connect(reuse_if_open=True)
return await call_next(request)
app.include_router(featured_router)
return app
@pytest.fixture
def cotw_client(cotw_db):
"""FastAPI TestClient backed by in-memory SQLite for COTW endpoint tests."""
with TestClient(_build_cotw_app()) as c:
yield c
# ---------------------------------------------------------------------------
# Seed helpers
# ---------------------------------------------------------------------------
_RARITY_COUNTER = [0]
_TEAM_COUNTER = [0]
def _make_rarity(name: str = "Common") -> Rarity:
_RARITY_COUNTER[0] += 1
unique_name = f"COTW_{name}_{_RARITY_COUNTER[0]}"
return Rarity.create(value=_RARITY_COUNTER[0], name=unique_name, color="#ffffff")
def _make_cardset(name: str = "COTW Set") -> Cardset:
return Cardset.create(name=name, description="cotw test", total_cards=10)
def _make_player(
rarity: Rarity,
cardset: Cardset,
name: str = "Test Player",
image: str = "https://example.com/card.png",
) -> Player:
return Player.create(
p_name=name,
rarity=rarity,
cardset=cardset,
set_num=1,
pos_1="1B",
image=image,
mlbclub="TST",
franchise="TST",
description="cotw test player",
)
def _make_team(is_ai: bool = False, suffix: str = "") -> Team:
_TEAM_COUNTER[0] += 1
idx = _TEAM_COUNTER[0]
abbrev = f"CT{idx}{suffix}"[:20]
return Team.create(
abbrev=abbrev,
sname=f"COTW{idx}",
lname=f"COTW Team {idx}",
gmid=9_000_000 + idx,
gmname=f"cotw_gm_{idx}",
gsheet="https://docs.google.com/cotw",
wallet=500,
team_value=1000,
collection_value=1000,
season=11,
is_ai=1 if is_ai else 0,
)
def _make_pack_type() -> PackType:
return PackType.create(
name="COTW Pack Type",
card_count=5,
description="cotw test pack type",
cost=100,
)
def _make_pack(team: Team, pack_type: PackType, open_time: datetime) -> Pack:
return Pack.create(
team=team,
pack_type=pack_type,
open_time=open_time,
)
def _make_card(player: Player, team: Team, pack: Pack, value: int = 50) -> Card:
return Card.create(
player=player,
team=team,
pack=pack,
value=value,
)
def _recent(days_ago: float = 1.0) -> datetime:
"""Return a datetime that is `days_ago` days before now (within the 7-day window)."""
return datetime.utcnow() - timedelta(days=days_ago)
def _old(days_ago: float = 14.0) -> datetime:
"""Return a datetime that is `days_ago` days before now (outside the 7-day window)."""
return datetime.utcnow() - timedelta(days=days_ago)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_happy_path_single_eligible_card(cotw_db, cotw_client):
"""GET /api/v2/cards/featured/card-of-the-week returns the eligible card.
What: Seeds one card with pack.open_time within the 7-day window. Calls
the endpoint and asserts that all required response fields are present and
match the seeded data.
Why: This is the primary happy-path test. If any field is missing, the
n8n integration will fail silently at embed-render time.
"""
rarity = _make_rarity("MVP")
cardset = _make_cardset("COTW Happy Set")
player = _make_player(
rarity,
cardset,
name="Happy Gilmore",
image="https://s3.example.com/happy.png",
)
team = _make_team()
pt = _make_pack_type()
pack = _make_pack(team, pt, open_time=_recent(1))
card = _make_card(player, team, pack, value=85)
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["card_id"] == card.id
assert data["player_name"] == "Happy Gilmore"
assert data["team_name"] == team.sname
assert data["team_abbrev"] == team.abbrev
assert data["rarity_name"] == rarity.name
assert data["rating"] == 85
assert data["card_image_url"] == "https://s3.example.com/happy.png"
assert data["cardset_name"] == "COTW Happy Set"
assert data["card_created_at"] is not None # open_time is set
def test_empty_window_returns_404(cotw_db, cotw_client):
"""GET /api/v2/cards/featured/card-of-the-week returns 404 when no card is in window.
What: Seeds a card whose pack was opened 14 days ago (outside the 7-day
window). Asserts the endpoint returns 404.
Why: The n8n workflow must gracefully handle weeks with no new activity.
The 404 is the agreed contract; a 200 with empty data would require
different handling on the consumer side.
"""
rarity = _make_rarity("Common")
cardset = _make_cardset("COTW Old Set")
player = _make_player(rarity, cardset, name="Old Timer")
team = _make_team()
pt = _make_pack_type()
pack = _make_pack(team, pt, open_time=_old(14))
_make_card(player, team, pack, value=70)
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week")
assert resp.status_code == 404
assert "no featured card this week" in resp.json()["detail"]
def test_ai_team_excluded_by_default(cotw_db, cotw_client):
"""AI-team cards are excluded by default; non-AI card is returned instead.
What: Seeds two cards in the window — one on an AI team (high rating) and
one on a human team (lower rating). The default request must return the
human-team card.
Why: AI cards artificially inflate ratings and should not appear in the
public content feed. The default must be exclusion-safe without callers
needing to remember the ?include_ai=false param.
"""
rarity = _make_rarity("MVP")
cardset = _make_cardset("AI Excl Set")
pt = _make_pack_type()
ai_team = _make_team(is_ai=True, suffix="AI")
human_team = _make_team(is_ai=False, suffix="HU")
ai_player = _make_player(
rarity, cardset, name="Robot Arm", image="https://s3.example.com/robot.png"
)
human_player = _make_player(
rarity, cardset, name="Human Arm", image="https://s3.example.com/human.png"
)
ai_pack = _make_pack(ai_team, pt, open_time=_recent(1))
human_pack = _make_pack(human_team, pt, open_time=_recent(1))
_make_card(ai_player, ai_team, ai_pack, value=99) # higher rating, AI team
human_card = _make_card(
human_player, human_team, human_pack, value=80
) # lower rating, human
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["card_id"] == human_card.id, (
"Expected the human-team card; AI card should be excluded by default"
)
assert data["player_name"] == "Human Arm"
def test_include_ai_true_includes_ai_team_card(cotw_db, cotw_client):
"""?include_ai=true lifts AI exclusion and returns the AI card when it has top rating.
What: Same setup as test_ai_team_excluded_by_default, but calls with
?include_ai=true. The AI card (higher rating) must be returned.
Why: Admin tooling and diagnostics may need to inspect AI-team cards. The
opt-in parameter must work correctly or the exclusion logic is broken.
"""
rarity = _make_rarity("MVP")
cardset = _make_cardset("AI Incl Set")
pt = _make_pack_type()
ai_team = _make_team(is_ai=True, suffix="AI2")
human_team = _make_team(is_ai=False, suffix="HU2")
ai_player = _make_player(
rarity, cardset, name="Cyborg King", image="https://s3.example.com/cyborg.png"
)
human_player = _make_player(
rarity, cardset, name="Regular Joe", image="https://s3.example.com/joe.png"
)
ai_pack = _make_pack(ai_team, pt, open_time=_recent(1))
human_pack = _make_pack(human_team, pt, open_time=_recent(1))
ai_card = _make_card(ai_player, ai_team, ai_pack, value=99)
_make_card(human_player, human_team, human_pack, value=80)
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week?include_ai=true")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["card_id"] == ai_card.id, (
"Expected the AI-team card when include_ai=true and it has the highest rating"
)
assert data["player_name"] == "Cyborg King"
def test_tiebreak_highest_rating_wins(cotw_db, cotw_client):
"""The card with the highest rating is returned when two cards are in the window.
What: Seeds two cards opened at the same time with different values (80 and
90). Asserts the card with value=90 is returned.
Why: Primary sort must be rating descending. If the sort order is wrong the
weekly content feed could highlight mediocre cards instead of standouts.
"""
rarity = _make_rarity("HoF")
cardset = _make_cardset("Tiebreak Rating Set")
pt = _make_pack_type()
team = _make_team()
open_time = _recent(2)
p1 = _make_player(
rarity, cardset, name="Low Rater", image="https://s3.example.com/low.png"
)
p2 = _make_player(
rarity, cardset, name="High Rater", image="https://s3.example.com/high.png"
)
pack1 = _make_pack(team, pt, open_time=open_time)
pack2 = _make_pack(team, pt, open_time=open_time)
_make_card(p1, team, pack1, value=80)
card_high = _make_card(p2, team, pack2, value=90)
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["card_id"] == card_high.id, (
"Highest-rating card must win the primary sort"
)
assert data["rating"] == 90
def test_tiebreak_same_rating_newest_pack_wins(cotw_db, cotw_client):
"""When two cards have identical ratings, the one from the newer pack is returned.
What: Seeds two cards with the same value (75) but opened at different times
(3 days ago vs 1 day ago). Asserts the newer card is returned.
Why: The secondary sort (open_time DESC) must be deterministic so the same
card is returned on every call within a calendar day — important for
cache-friendliness and avoiding flip-flop behavior in the Discord post.
"""
rarity = _make_rarity("All-Star")
cardset = _make_cardset("Tiebreak Newest Set")
pt = _make_pack_type()
team = _make_team()
p_old = _make_player(
rarity,
cardset,
name="Old Pack Player",
image="https://s3.example.com/old_pack.png",
)
p_new = _make_player(
rarity,
cardset,
name="New Pack Player",
image="https://s3.example.com/new_pack.png",
)
pack_old = _make_pack(team, pt, open_time=_recent(3))
pack_new = _make_pack(team, pt, open_time=_recent(1))
_make_card(p_old, team, pack_old, value=75)
card_new = _make_card(p_new, team, pack_new, value=75)
resp = cotw_client.get("/api/v2/cards/featured/card-of-the-week")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["card_id"] == card_new.id, (
"Newer pack must win the secondary tiebreak when ratings are equal"
)
assert data["player_name"] == "New Pack Player"