""" core/importer.py ---------------- CSV draw importer for LottoSight. Supported formats (auto-detected): 1. LottoSight export — Game, Date, "N1,N2,...", Bonus, Multiplier, Source 2. Wide format — Date, N1, N2, ..., Nn [, Bonus] 3. Packed format — Date, "N1,N2,...", [Bonus] Header rows are auto-detected and skipped. Dates accepted: YYYY-MM-DD, MM/DD/YYYY, M/D/YYYY, MM-DD-YYYY. Duplicate draws (same game + date already in DB) are skipped, not errored. """ import csv from datetime import datetime from db.models import get_game_by_id, insert_draw # ── Date parsing ────────────────────────────────────────────────────────────── _DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%m-%d-%Y", "%d/%m/%Y", "%-m/%-d/%Y") def _parse_date(text: str) -> str | None: """Return ISO date string (YYYY-MM-DD) or None.""" text = text.strip() for fmt in _DATE_FORMATS: try: return datetime.strptime(text, fmt).strftime("%Y-%m-%d") except ValueError: pass return None def _is_header(row: list[str]) -> bool: """Heuristic: first cell is not a recognisable date → row is a header.""" return bool(row) and _parse_date(row[0]) is None # ── Row parsing ─────────────────────────────────────────────────────────────── def _parse_row_lottosight(cells: list[str]) -> dict | None: """Parse LottoSight export row: Game, Date, Numbers, Bonus, ...""" if len(cells) < 3: return None date_str = _parse_date(cells[1]) if date_str is None: return None try: numbers = [int(n.strip()) for n in cells[2].split(",") if n.strip()] bonus = int(cells[3]) if len(cells) > 3 and cells[3].strip().isdigit() else None except (ValueError, IndexError): return None return {"date": date_str, "numbers": numbers, "bonus": bonus} def _parse_row_generic(cells: list[str], game: dict) -> dict | None: """Parse wide or packed format: Date, [nums...]""" date_str = _parse_date(cells[0]) if date_str is None: return None rest = cells[1:] if not rest: return None # Packed: second cell contains comma-separated numbers if "," in rest[0]: try: numbers = [int(n.strip()) for n in rest[0].split(",") if n.strip()] bonus = int(rest[1]) if len(rest) > 1 and rest[1].strip().isdigit() else None return {"date": date_str, "numbers": numbers, "bonus": bonus} except ValueError: pass # Wide: each number in its own column nums: list[int] = [] for col in rest: col = col.strip() if col.isdigit(): nums.append(int(col)) else: break # stop at first non-numeric cell (e.g. multiplier string) if len(nums) < game["main_count"]: return None numbers = nums[: game["main_count"]] bonus = (nums[game["main_count"]] if game["bonus_count"] > 0 and len(nums) > game["main_count"] else None) return {"date": date_str, "numbers": numbers, "bonus": bonus} # ── Public API ──────────────────────────────────────────────────────────────── def import_draws_csv(game_id: int, filepath: str) -> dict: """ Import draw records from *filepath* into the database for *game_id*. Returns: {"added": int, "skipped": int, "errors": list[str]} """ game = get_game_by_id(game_id) if game is None: return {"added": 0, "skipped": 0, "errors": [f"Unknown game_id {game_id}"]} added = skipped = 0 errors: list[str] = [] # Read file try: with open(filepath, newline="", encoding="utf-8-sig") as f: sample = f.read(4096) f.seek(0) try: dialect = csv.Sniffer().sniff(sample, delimiters=",\t;") except csv.Error: dialect = csv.excel rows = list(csv.reader(f, dialect)) except FileNotFoundError: return {"added": 0, "skipped": 0, "errors": [f"File not found: {filepath}"]} except Exception as e: return {"added": 0, "skipped": 0, "errors": [f"Could not read file: {e}"]} if not rows: return {"added": 0, "skipped": 0, "errors": ["File is empty"]} # Detect LottoSight export format by header lottosight_fmt = ( len(rows[0]) >= 3 and rows[0][0].strip().lower() == "game" and rows[0][1].strip().lower() == "date" ) # Determine start row (skip header if present) start = 1 if (lottosight_fmt or _is_header(rows[0])) else 0 for line_num, row in enumerate(rows[start:], start=start + 1): cells = [c.strip() for c in row] if not any(cells): continue # blank line if lottosight_fmt: parsed = _parse_row_lottosight(cells) else: parsed = _parse_row_generic(cells, game) if parsed is None: errors.append(f"Line {line_num}: could not parse — {row}") continue numbers = parsed["numbers"] # Validate count if len(numbers) != game["main_count"]: errors.append( f"Line {line_num}: expected {game['main_count']} numbers, " f"got {len(numbers)}" ) continue # Validate range bad = [n for n in numbers if not (1 <= n <= game["main_max"])] if bad: errors.append( f"Line {line_num}: numbers out of range 1–{game['main_max']}: {bad}" ) continue result = insert_draw( game_id, parsed["date"], numbers, bonus=parsed["bonus"], source="csv_import", ) if result == "inserted": added += 1 else: skipped += 1 return {"added": added, "skipped": skipped, "errors": errors}