248 lines
8.9 KiB
Python
248 lines
8.9 KiB
Python
import datetime
|
|
import decimal
|
|
import logging
|
|
import typing
|
|
import uuid
|
|
|
|
import aiosqlite
|
|
|
|
import tutor.models
|
|
import tutor.search
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def search(
|
|
db: aiosqlite.Connection,
|
|
name: typing.Optional[str] = None,
|
|
collector_number: typing.Optional[str] = None,
|
|
set_code: typing.Optional[str] = None,
|
|
set_name: typing.Optional[str] = None,
|
|
foil: typing.Optional[bool] = None,
|
|
alternate_art: typing.Optional[bool] = None,
|
|
scryfall_id: typing.Optional[str] = None,
|
|
limit: int = 10,
|
|
distinct: bool = True,
|
|
in_collection: typing.Optional[bool] = None,
|
|
) -> typing.List[tutor.models.Card]:
|
|
db.row_factory = aiosqlite.Row
|
|
joins = []
|
|
constraints = []
|
|
params = {}
|
|
if name is not None:
|
|
constraints.append("cards.name LIKE :name")
|
|
params["name"] = name
|
|
if collector_number is not None:
|
|
constraints.append("cards.collector_number LIKE :number")
|
|
params["number"] = collector_number
|
|
if set_code is not None:
|
|
constraints.append("cards.set_code LIKE :set_code")
|
|
params["set_code"] = set_code.upper()
|
|
if set_name is not None:
|
|
constraints.append("sets.name LIKE :set_name")
|
|
params["set_name"] = set_name
|
|
if foil is not None:
|
|
constraints.append("cards.foil IS :foil")
|
|
params["foil"] = foil
|
|
if alternate_art is not None:
|
|
constraints.append("cards.variation IS :alternative")
|
|
params["alternative"] = alternate_art
|
|
if scryfall_id is not None:
|
|
constraints.append("cards.scryfall_id LIKE :scryfall_id")
|
|
params["scryfall_id"] = scryfall_id
|
|
if in_collection is not None:
|
|
if in_collection:
|
|
joins.append("JOIN copies USING (scryfall_id)")
|
|
else:
|
|
joins.append("LEFT JOIN copies USING (scryfall_id)")
|
|
constraints.append("copies.id IS NULL")
|
|
joins.append("JOIN sets USING (set_code)")
|
|
query = " ".join(
|
|
[
|
|
"SELECT cards.* FROM cards",
|
|
" ".join(joins),
|
|
"WHERE" if constraints else "",
|
|
" AND ".join(constraints),
|
|
f"LIMIT {limit}",
|
|
]
|
|
)
|
|
cursor = await db.execute(query, params)
|
|
rows = await cursor.fetchall()
|
|
return [
|
|
tutor.models.Card(
|
|
scryfall_id=uuid.UUID(row["scryfall_id"]),
|
|
name=row["name"],
|
|
set_code=row["set_code"],
|
|
collector_number=row["collector_number"],
|
|
rarity=tutor.models.Rarity.from_string(row["rarity"]),
|
|
color_identity=tutor.models.Color.from_string(row["color_identity"]),
|
|
cmc=decimal.Decimal(row["cmc"]),
|
|
type_line=row["type_line"],
|
|
release_date=datetime.date.fromisoformat(row["release_date"]),
|
|
games=set(),
|
|
legalities={},
|
|
edhrec_rank=row["edhrec_rank"],
|
|
oracle_text=row["oracle_text"],
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
|
|
async def advanced_search(
|
|
db: aiosqlite.Connection,
|
|
search: tutor.search.Search,
|
|
limit: int = 10,
|
|
offset: int = 0,
|
|
in_collection: typing.Optional[bool] = None,
|
|
) -> typing.List[tutor.models.Card]:
|
|
db.row_factory = aiosqlite.Row
|
|
joins = []
|
|
constraints = []
|
|
params = {}
|
|
sets = []
|
|
|
|
logging.debug("Performing search for: %s", search)
|
|
for i, criterion in enumerate(search.criteria):
|
|
param = f"param_{i}"
|
|
if isinstance(criterion, tutor.search.Name):
|
|
constraints.append(f"cards.name LIKE :{param}")
|
|
params[param] = f"%{criterion.name}%"
|
|
if isinstance(criterion, tutor.search.Type):
|
|
constraints.append(f"cards.type_line LIKE :{param}")
|
|
params[param] = f"%{criterion.name}%"
|
|
if isinstance(criterion, tutor.search.IsExpansion):
|
|
constraints.append(f"cards.set_code LIKE :{param}")
|
|
params[param] = criterion.set_code
|
|
if isinstance(criterion, tutor.search.InExpansion):
|
|
sets.append(criterion.set_code)
|
|
if isinstance(criterion, tutor.search.IsColor):
|
|
constraints.append(f"cards.color_identity LIKE :{param}")
|
|
params[param] = tutor.models.Color.to_string(criterion.colors)
|
|
if isinstance(criterion, tutor.search.IsRarity):
|
|
constraints.append(f"cards.rarity LIKE :{param}")
|
|
params[param] = str(criterion.rarity)
|
|
if isinstance(criterion, tutor.search.Oracle):
|
|
constraints.append(f"cards.oracle_text LIKE :{param}")
|
|
params[param] = f"%{criterion.text}%"
|
|
|
|
if sets:
|
|
set_params = {f"set_{i}": set_code for i, set_code in enumerate(sets)}
|
|
constraints.append(
|
|
"cards.set_code IN ({})".format(
|
|
", ".join([f":{key}" for key in set_params.keys()])
|
|
)
|
|
)
|
|
params.update(set_params)
|
|
if in_collection is not None:
|
|
if in_collection:
|
|
joins.append("JOIN copies ON (cards.scryfall_id = copies.scryfall_id)")
|
|
else:
|
|
joins.append("LEFT JOIN copies ON (cards.scryfall_id = copies.scryfall_id)")
|
|
constraints.append("copies.id IS NULL")
|
|
joins.append("JOIN sets ON (cards.set_code = sets.set_code)")
|
|
query = " ".join(
|
|
[
|
|
"SELECT cards.* FROM cards",
|
|
" ".join(joins),
|
|
"WHERE" if constraints else "",
|
|
" AND ".join(constraints),
|
|
f"LIMIT {offset},{limit}",
|
|
]
|
|
)
|
|
cursor = await db.execute(query, params)
|
|
rows = await cursor.fetchall()
|
|
return [
|
|
tutor.models.Card(
|
|
scryfall_id=uuid.UUID(row["scryfall_id"]),
|
|
name=row["name"],
|
|
set_code=row["set_code"],
|
|
collector_number=row["collector_number"],
|
|
rarity=tutor.models.Rarity.from_string(row["rarity"]),
|
|
color_identity=tutor.models.Color.from_string(row["color_identity"]),
|
|
cmc=decimal.Decimal(row["cmc"]),
|
|
type_line=row["type_line"],
|
|
release_date=datetime.date.fromisoformat(row["release_date"]),
|
|
games=set(),
|
|
legalities={},
|
|
edhrec_rank=row["edhrec_rank"],
|
|
oracle_text=row["oracle_text"],
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
|
|
async def store_card(db: aiosqlite.Connection, card: tutor.models.Card) -> None:
|
|
await db.execute(
|
|
"INSERT INTO cards "
|
|
"(`scryfall_id`, `name`, `set_code`, `collector_number`, `rarity`,"
|
|
" `color_identity`, `cmc`, `type_line`, `release_date`, `edhrec_rank`,"
|
|
" `oracle_text`) "
|
|
"VALUES (:scryfall_id, :name, :set_code, :collector_number, :rarity,"
|
|
" :color_identity, :cmc, :type_line, :release_date, :edhrec_rank,"
|
|
" :oracle_text) "
|
|
"ON CONFLICT (scryfall_id) DO UPDATE "
|
|
"SET `name` = :name"
|
|
" , `set_code` = :set_code"
|
|
" , `collector_number` = :collector_number"
|
|
" , `rarity` = :rarity"
|
|
" , `color_identity` = :color_identity"
|
|
" , `cmc` = :cmc"
|
|
" , `type_line` = :type_line"
|
|
" , `release_date` = :release_date"
|
|
" , `edhrec_rank` = :edhrec_rank"
|
|
" , `oracle_text` = :oracle_text",
|
|
{
|
|
"scryfall_id": str(card.scryfall_id),
|
|
"name": card.name,
|
|
"set_code": card.set_code,
|
|
"collector_number": card.collector_number,
|
|
"rarity": str(card.rarity),
|
|
"color_identity": tutor.models.Color.to_string(card.color_identity),
|
|
"cmc": str(card.cmc),
|
|
"type_line": card.type_line,
|
|
"release_date": str(card.release_date),
|
|
"edhrec_rank": card.edhrec_rank,
|
|
"oracle_text": card.oracle_text,
|
|
},
|
|
)
|
|
await db.execute(
|
|
"DELETE FROM `games` WHERE `scryfall_id` = ?",
|
|
[str(card.scryfall_id)],
|
|
)
|
|
for game in card.games:
|
|
await db.execute(
|
|
"INSERT INTO `games` (`scryfall_id`, `game`) VALUES (?, ?)",
|
|
(str(card.scryfall_id), game.value),
|
|
)
|
|
await db.execute(
|
|
"DELETE FROM `legalities` WHERE `scryfall_id` = ?",
|
|
[str(card.scryfall_id)],
|
|
)
|
|
for game_format, legality in card.legalities.items():
|
|
await db.execute(
|
|
"INSERT INTO `legalities` (`scryfall_id`, `format`, `legality`) "
|
|
"VALUES (?, ?, ?)",
|
|
(str(card.scryfall_id), game_format, legality.value),
|
|
)
|
|
|
|
|
|
async def store_set(db: aiosqlite.Connection, set_code: str, name: str) -> None:
|
|
await db.execute(
|
|
"INSERT INTO `sets` (`set_code`, `name`) "
|
|
"VALUES (:set_code, :name) "
|
|
"ON CONFLICT (`set_code`) DO NOTHING",
|
|
{"set_code": set_code, "name": name},
|
|
)
|
|
|
|
|
|
async def store_copy(db: aiosqlite.Connection, copy: tutor.models.CardCopy) -> None:
|
|
await db.execute(
|
|
"INSERT INTO copies (scryfall_id, isFoil, condition)"
|
|
"VALUES (:scryfall_id, :foil, :condition)",
|
|
{
|
|
"scryfall_id": str(copy.card.scryfall_id),
|
|
"foil": copy.foil,
|
|
"condition": copy.condition,
|
|
},
|
|
)
|