Initial commit - steps 1-6 complete

manabox.py

@@ -0,0 +1,99 @@
+"""
+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