67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
import csv
|
|
import enum
|
|
import logging
|
|
import typing
|
|
|
|
import psycopg
|
|
|
|
import tutor.database
|
|
import tutor.models
|
|
|
|
|
|
class ImportMode(enum.Enum):
|
|
add = "add"
|
|
replace = "replace"
|
|
|
|
|
|
async def load(
|
|
settings: dict, stream: typing.IO, mode: ImportMode = ImportMode.add
|
|
) -> typing.Iterable[typing.Tuple[tutor.models.Card, int]]:
|
|
"""Load cards from a CSV file.
|
|
|
|
Currently supports the following formats:
|
|
- Deckbox (set name match can fail)
|
|
- MTGStand (uses Scryfall ID)
|
|
"""
|
|
cards = []
|
|
async with await psycopg.AsyncConnection.connect(
|
|
settings["database"], autocommit=False
|
|
) as conn:
|
|
async with conn.cursor() as cursor:
|
|
if mode == ImportMode.replace:
|
|
await tutor.database.clear_copies(cursor)
|
|
reader = csv.DictReader(stream)
|
|
for row in reader:
|
|
is_foil = "Foil" in row and row["Foil"].lower() == "foil"
|
|
quantity = int(row.get("Quantity", row.get("Count", 1)))
|
|
if "Scryfall ID" in row:
|
|
found = await tutor.database.search(
|
|
cursor, scryfall_id=row["Scryfall ID"]
|
|
)
|
|
else:
|
|
found = await tutor.database.search(
|
|
cursor,
|
|
name=row["Name"],
|
|
set_name=row["Edition"],
|
|
collector_number=row["Card Number"],
|
|
foil=is_foil or None,
|
|
)
|
|
if not found:
|
|
logging.warning("Could not find card for row %s", row)
|
|
continue
|
|
else:
|
|
if len(found) > 1:
|
|
logging.warning(
|
|
"Found %s possibilities for row %s", len(found), row
|
|
)
|
|
card = tutor.models.CardCopy(
|
|
card=found[0],
|
|
foil=is_foil,
|
|
language=row["Language"] or "English",
|
|
collection=row.get("List name") or "Default",
|
|
)
|
|
logging.info((quantity, card))
|
|
for i in range(quantity):
|
|
await tutor.database.store_copy(cursor, card)
|
|
yield card, reader.line_num
|
|
await conn.commit()
|