import csv import io from datetime import date, datetime from decimal import Decimal, InvalidOperation from django.core.exceptions import ValidationError from django.db import transaction from openpyxl import load_workbook from shop_management.models.data.inventory import InventoryBatch from shop_management.models.data.product import Product from shop_management.models.data.stock_adjustment import StockAdjustment, StockAdjustmentItem from shop_management.models.data.utils import Currency from shop_management.services.stock_adjustments import post_stock_adjustment REQUIRED_COLUMNS = { "product_code", "batch_number", "manufacture_date", "expiry_date", "quantity", "unit_cost", "currency", } MAX_ROWS = 5000 def _normalize_header(value): return str(value or "").strip().lower().replace(" ", "_").replace("-", "_") def _read_rows(uploaded_file): name = str(uploaded_file.name or "").lower() uploaded_file.seek(0) if name.endswith(".csv"): text = io.TextIOWrapper(uploaded_file.file, encoding="utf-8-sig", newline="") reader = csv.DictReader(text) headers = [_normalize_header(value) for value in (reader.fieldnames or [])] rows = [dict(zip(headers, [value for value in row.values()])) for row in reader] text.detach() elif name.endswith(".xlsx"): workbook = load_workbook(uploaded_file, read_only=True, data_only=True) sheet = workbook.active iterator = sheet.iter_rows(values_only=True) first = next(iterator, None) headers = [_normalize_header(value) for value in (first or [])] rows = [dict(zip(headers, values)) for values in iterator] workbook.close() else: raise ValidationError({"file": "Only .csv and .xlsx files are supported."}) missing = sorted(REQUIRED_COLUMNS - set(headers)) if missing: raise ValidationError({"file": f"Missing required columns: {', '.join(missing)}"}) if len(rows) > MAX_ROWS: raise ValidationError({"file": f"Maximum {MAX_ROWS} data rows are allowed."}) return rows def _date_value(value, field, errors): if value in (None, ""): return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value text = str(value).strip() try: return date.fromisoformat(text) except ValueError: errors.append(f"{field} must use YYYY-MM-DD format.") return None def _positive_decimal(value, field, errors, *, allow_zero=False): try: result = Decimal(str(value).strip()) except (InvalidOperation, AttributeError, ValueError): errors.append(f"{field} must be a valid number.") return None if result < 0 or (result == 0 and not allow_zero): errors.append(f"{field} must be {'zero or greater' if allow_zero else 'greater than zero'}.") return None return result def validate_opening_inventory(uploaded_file, warehouse): raw_rows = _read_rows(uploaded_file) results, errors, seen = [], [], set() for index, raw in enumerate(raw_rows, start=2): if not any(value not in (None, "") for value in raw.values()): continue row_errors = [] code = str(raw.get("product_code") or "").strip() batch_number = str(raw.get("batch_number") or "").strip() currency_code = str(raw.get("currency") or "").strip() products = Product.objects.filter(shop=warehouse.shop, code__iexact=code, is_active=True) product = products.first() if products.count() == 1 else None if not code: row_errors.append("product_code is required.") elif products.count() == 0: row_errors.append(f"Active product code '{code}' was not found in this branch.") elif products.count() > 1: row_errors.append(f"Product code '{code}' is duplicated in this branch.") if product and product.requires_batch and not batch_number: row_errors.append("batch_number is required for this product.") if product and not batch_number: batch_number = "NO-BATCH" manufacture = _date_value(raw.get("manufacture_date"), "manufacture_date", row_errors) expiry = _date_value(raw.get("expiry_date"), "expiry_date", row_errors) if manufacture and expiry and expiry < manufacture: row_errors.append("expiry_date cannot be before manufacture_date.") if product and product.requires_expiry and not expiry: row_errors.append("expiry_date is required for this product.") quantity = _positive_decimal(raw.get("quantity"), "quantity", row_errors) cost = _positive_decimal(raw.get("unit_cost"), "unit_cost", row_errors, allow_zero=True) currencies = Currency.objects.filter(shop=warehouse.shop, title__iexact=currency_code) currency = currencies.first() if currencies.count() == 1 else None if not currency_code: row_errors.append("currency is required.") elif currencies.count() == 0: row_errors.append(f"Currency '{currency_code}' was not found in this branch.") elif currencies.count() > 1: row_errors.append(f"Currency '{currency_code}' is duplicated in this branch.") key = (product.id if product else code.lower(), batch_number.lower()) if key in seen: row_errors.append("Duplicate product and batch in this file; combine the quantities into one row.") seen.add(key) existing = InventoryBatch.objects.filter(product=product, batch_number=batch_number).first() if product else None if existing and (existing.manufacture_date != manufacture or existing.expiry_date != expiry): row_errors.append("Existing batch dates do not match the import row.") normalized = { "row": index, "product_id": product.id if product else None, "product_code": code, "product_name": product.name if product else "", "batch_number": batch_number, "manufacture_date": manufacture, "expiry_date": expiry, "quantity": quantity, "unit_cost": cost, "currency_id": currency.id if currency else None, "currency": currency.title if currency else currency_code, } if row_errors: errors.append({"row": index, "messages": row_errors}) results.append(normalized) if not results: errors.append({"row": None, "messages": ["The file contains no data rows."]}) return results, errors @transaction.atomic def post_opening_inventory(uploaded_file, warehouse, *, user, reference=""): rows, errors = validate_opening_inventory(uploaded_file, warehouse) if errors: raise ValidationError({"rows": errors}) adjustment = StockAdjustment.objects.create( adjustment_kind=StockAdjustment.Kind.OPENING, warehouse=warehouse, reason="Opening Inventory", notes=str(reference or "").strip(), created_by=user, ) for row in rows: product = Product.objects.get(pk=row["product_id"]) batch, _ = InventoryBatch.objects.get_or_create( product=product, batch_number=row["batch_number"], defaults={ "manufacture_date": row["manufacture_date"], "expiry_date": row["expiry_date"], "notes": "Created by opening inventory import", "created_by": user, }, ) StockAdjustmentItem.objects.create( adjustment=adjustment, product=product, batch=batch, direction=StockAdjustmentItem.Direction.IN, quantity=row["quantity"], unit_cost=row["unit_cost"], currency_id=row["currency_id"], remarks="Opening inventory", created_by=user, ) return post_stock_adjustment(adjustment.pk, user=user)