100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""
|
|
Manabox collection importer.
|
|
|
|
Manabox exports a single CSV format:
|
|
Name, Set code, Collector number, Foil, Quantity, ...
|
|
|
|
The "Foil" column is a boolean ("Yes"/"No" or "1"/"0") applied per-row.
|
|
Manabox creates separate rows for foil and non-foil copies of the same card,
|
|
so we merge them into a single RawCardRow (quantity + foil_quantity).
|
|
"""
|
|
import csv
|
|
import io
|
|
from collections import defaultdict
|
|
|
|
import chardet
|
|
|
|
from app.services.imports.models import RawCardRow
|
|
|
|
SOURCE = "manabox"
|
|
|
|
_NAME_COLS = {"name", "card name", "cardname", "card_name"}
|
|
_SET_COLS = {"set code", "set", "setcode", "set_code", "edition code"}
|
|
_COLLECTOR_COLS = {"collector number", "collector #", "collectornumber", "number"}
|
|
_FOIL_COLS = {"foil", "is foil", "isfoil"}
|
|
_QTY_COLS = {"quantity", "qty", "count", "amount"}
|
|
|
|
|
|
def parse(raw_bytes: bytes) -> list[RawCardRow]:
|
|
detected = chardet.detect(raw_bytes)
|
|
encoding = detected.get("encoding") or "utf-8"
|
|
text = raw_bytes.decode(encoding, errors="replace").strip()
|
|
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
if not reader.fieldnames:
|
|
raise ValueError("Manabox CSV appears to be empty or has no header row")
|
|
|
|
headers = {h.strip().lower(): h for h in reader.fieldnames if h}
|
|
|
|
name_col = _find_col(headers, _NAME_COLS, "Name")
|
|
set_col = _find_col(headers, _SET_COLS, None, required=False)
|
|
collect_col = _find_col(headers, _COLLECTOR_COLS, None, required=False)
|
|
foil_col = _find_col(headers, _FOIL_COLS, None, required=False)
|
|
qty_col = _find_col(headers, _QTY_COLS, "Quantity")
|
|
|
|
# Key: (card_name, set_code, collector_number) → merged row
|
|
merged: dict[tuple, RawCardRow] = defaultdict(lambda: RawCardRow(import_source=SOURCE))
|
|
|
|
for row in reader:
|
|
name = row.get(name_col, "").strip()
|
|
if not name:
|
|
continue
|
|
|
|
set_code = (row.get(set_col, "") or "").strip().lower() if set_col else ""
|
|
collector = (row.get(collect_col, "") or "").strip() if collect_col else ""
|
|
key = (name.lower(), set_code, collector)
|
|
|
|
try:
|
|
qty = int(float(row.get(qty_col, "1") or "1"))
|
|
except (ValueError, TypeError):
|
|
qty = 1
|
|
|
|
is_foil = _parse_bool(row.get(foil_col, "")) if foil_col else False
|
|
|
|
entry = merged[key]
|
|
entry.card_name = name
|
|
entry.set_code = set_code
|
|
entry.collector_number = collector
|
|
|
|
if is_foil:
|
|
entry.foil_quantity += qty
|
|
else:
|
|
entry.quantity += qty
|
|
|
|
if not merged:
|
|
raise ValueError("Manabox CSV contained no card rows")
|
|
|
|
return list(merged.values())
|
|
|
|
|
|
def _parse_bool(value: str) -> bool:
|
|
return str(value).strip().lower() in {"yes", "true", "1", "y"}
|
|
|
|
|
|
def _find_col(
|
|
headers: dict[str, str],
|
|
aliases: set[str],
|
|
fallback: str | None,
|
|
required: bool = True,
|
|
) -> str | None:
|
|
for alias in aliases:
|
|
if alias in headers:
|
|
return headers[alias]
|
|
if fallback and fallback.lower() in headers:
|
|
return headers[fallback.lower()]
|
|
if required:
|
|
raise ValueError(
|
|
f"Could not find required column. Expected one of: {sorted(aliases)}"
|
|
)
|
|
return None
|