From d2e7e7141037377d293b00e691fa835694783f26 Mon Sep 17 00:00:00 2001 From: spl3g Date: Sat, 30 May 2026 15:07:50 +0300 Subject: add an import script --- core/management/commands/import_data.py | 101 ++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 core/management/commands/import_data.py diff --git a/core/management/commands/import_data.py b/core/management/commands/import_data.py new file mode 100644 index 0000000..f78f27e --- /dev/null +++ b/core/management/commands/import_data.py @@ -0,0 +1,101 @@ +import csv +import os +from typing import Any +from datetime import datetime + +from django.core.files.images import ImageFile +from django.core.management.base import BaseCommand + +from core.models import Order, OrderItem, PickupPoint, Supplier, User, Product, Role + +class Command(BaseCommand): + def handle(self, *args: Any, **kwargs: Any) -> str | None: + base_path = "attachments/" + + with open(os.path.join(base_path, "pickup_point.csv"), encoding="utf-8-sig") as f: + reader = csv.DictReader(f, ["address"]) + for row in reader: + print(row) + PickupPoint.objects.get_or_create(address=row["address"]) + + with open(os.path.join(base_path, "product.csv"), encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + for row in reader: + print(row) + if not row: + continue + + photo_name = row.pop("photo_name", None) + + sup_obj, _ = Supplier.objects.get_or_create(name=row.pop("supplier")) + product, _ = Product.objects.get_or_create(article=row["article"], supplier=sup_obj, defaults=row) + + if photo_name: + photo_path = os.path.join(base_path, photo_name) + if os.path.exists(photo_path): + with open(photo_path, "rb") as img_file: + image = ImageFile(img_file) + print(image.width, image.height) + product.photo.save( + os.path.basename(photo_path), + image, + save=True + ) + + with open(os.path.join(base_path, "user.csv"), encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + for row in reader: + print(row) + if not row: + continue + + role_obj, _ = Role.objects.get_or_create(name=row["role"]) + if not User.objects.filter(username=row["login"]).exists(): + user = User.objects.create( + username=row["login"], full_name=row["name"], role=role_obj + ) + user.set_password(str(row["password"]).strip()) + user.save() + + with open(os.path.join(base_path, "order.csv"), encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + for row in reader: + print(row) + if not row: + continue + + pp = row.pop("pp", 1) + items_string = row.pop("items", None) + + delivery_date_str = row.pop("delivery_date", None) + delivery_date = datetime.strptime(delivery_date_str, "%m-%d-%Y").date() if delivery_date_str else None + + order_date_str = row.pop("order_date", None) + order_date = datetime.strptime(order_date_str, "%m-%d-%Y").date() if order_date_str else None + + pp_obj = ( + PickupPoint.objects.get(id=pp) + if PickupPoint.objects.filter(id=pp).exists() + else PickupPoint.objects.first() + ) + + user_obj = User.objects.get(full_name=row["client_name"]) + order, created = Order.objects.get_or_create( + id=row["id"], + pickup_point=pp_obj, + user=user_obj, + delivery_date=delivery_date, + order_date=order_date, + defaults=row, + ) + if created and items_string: + items = items_string.split(", ") + for i in range(0, len(items), 2): + art = items[i].strip() + try: + prod = Products.get(article=art) + OrderItem.objects.create( + order=order, product=prod, count=int(items[i+1]) + ) + except Exception: + pass -- cgit v1.2.3