160 lines
7.3 KiB
Python
160 lines
7.3 KiB
Python
import csv
|
|
from datetime import datetime
|
|
from django.utils import timezone
|
|
from django.core.management.base import BaseCommand
|
|
from users.models import Employee
|
|
from equipment.models import Computer, Monitor, Peripheral
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Importă date din CSV-urile exportate din Microsoft Access"
|
|
|
|
def handle(self, *args, **kwargs):
|
|
# Setăm delimiterul corect (poate fi ',' sau ';' în funcție de setările tale regionale la export)
|
|
CSV_DELIMITER = ","
|
|
|
|
# Dicționar pentru a lega ID-ul vechi din Access de noul obiect Employee din Django
|
|
employee_map = {}
|
|
|
|
# Funcție ajutătoare pentru a evita erorile la date calendaristice goale
|
|
# Funcție ajutătoare pentru conversia datelor din DD-MM-YYYY în YYYY-MM-DD
|
|
def parse_date(date_str):
|
|
# Definim data de rezervă dacă rubrică e goală
|
|
fallback_date = "2000-01-01"
|
|
|
|
if not date_str or str(date_str).strip() == "":
|
|
return fallback_date
|
|
|
|
clean_str = str(date_str).strip()
|
|
|
|
try:
|
|
parsed = datetime.strptime(clean_str, "%d-%m-%Y")
|
|
return parsed.strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
try:
|
|
parsed = datetime.strptime(clean_str, "%d/%m/%Y")
|
|
return parsed.strftime("%Y-%m-%d")
|
|
except ValueError:
|
|
# Dacă data există dar formatul e complet invalid, punem tot fallback
|
|
self.stdout.write(
|
|
self.style.WARNING(
|
|
f"Atenție: Data '{clean_str}' e invalidă. Folosim data de azi."
|
|
)
|
|
)
|
|
return fallback_date
|
|
|
|
# 1. IMPORTUL ANGAJAȚILOR
|
|
self.stdout.write("Începe importul angajaților...")
|
|
try:
|
|
with open("Employee.csv", "r", encoding="utf-8-sig") as file:
|
|
reader = csv.DictReader(file, delimiter=CSV_DELIMITER)
|
|
for row in reader:
|
|
emp, created = Employee.objects.get_or_create(
|
|
username=row["username"],
|
|
defaults={
|
|
"name": row["name"],
|
|
"role": row.get("role", ""),
|
|
"office_number": row.get("office_number", ""),
|
|
"email": row.get("email", ""),
|
|
},
|
|
)
|
|
# Salvăm instanța generată în dicționar, folosind ID-ul vechi din CSV ca și cheie
|
|
employee_map[row["ID"]] = emp
|
|
if created:
|
|
self.stdout.write(f" Adăugat angajat: {emp.name}")
|
|
except FileNotFoundError:
|
|
self.stdout.write(
|
|
self.style.ERROR("Eroare: Nu am găsit fișierul Employee.csv")
|
|
)
|
|
|
|
# 2. IMPORTUL CALCULATOARELOR
|
|
self.stdout.write("\nÎncepe importul calculatoarelor...")
|
|
try:
|
|
with open("Computer.csv", "r", encoding="utf-8-sig") as file:
|
|
reader = csv.DictReader(file, delimiter=CSV_DELIMITER)
|
|
for row in reader:
|
|
# Folosim numele exact al coloanei specificat de tine (EmpolyeeID)
|
|
posesor = employee_map.get(row.get("EmpolyeeID"))
|
|
|
|
pc, created = Computer.objects.get_or_create(
|
|
equipment_code=row["equipment_code"],
|
|
defaults={
|
|
"entry_date": parse_date(row.get("entry_date")),
|
|
"oem_name": row.get("oem_name", ""),
|
|
"motherboard": row.get("motherboard", ""),
|
|
"cpu": row.get("cpu", ""),
|
|
"ram": row.get("ram", ""),
|
|
"storage": row.get("storage", ""),
|
|
"optical_drive": row.get("optical_drive", ""),
|
|
"gpu": row.get("gpu", ""),
|
|
"mac_address": row.get("mac_address", ""),
|
|
"os_license": row.get("os_license", ""),
|
|
"additional_licenses": row.get("additional_licenses", ""),
|
|
"observations": row.get("observations", ""),
|
|
"status": "activ",
|
|
"assigned_to": posesor,
|
|
},
|
|
)
|
|
if created:
|
|
self.stdout.write(f" Adăugat PC: {pc.equipment_code}")
|
|
except FileNotFoundError:
|
|
self.stdout.write(
|
|
self.style.ERROR("Eroare: Nu am găsit fișierul Computer.csv")
|
|
)
|
|
|
|
# 3. IMPORTUL MONITOARELOR
|
|
self.stdout.write("\nÎncepe importul monitoarelor...")
|
|
try:
|
|
with open("Monitor.csv", "r", encoding="utf-8-sig") as file:
|
|
reader = csv.DictReader(file, delimiter=CSV_DELIMITER)
|
|
for row in reader:
|
|
posesor = employee_map.get(row.get("EmployeeID"))
|
|
|
|
mon, created = Monitor.objects.get_or_create(
|
|
equipment_code=row["equipment_code"],
|
|
defaults={
|
|
"entry_date": parse_date(row.get("entry_date")),
|
|
"name": row.get("name", ""),
|
|
"serial_number": row.get("serial_number", ""),
|
|
"screen_diagonal": row.get("screen_diagonal", ""),
|
|
"observations": row.get("observations", ""),
|
|
"status": "activ",
|
|
"assigned_to": posesor,
|
|
},
|
|
)
|
|
if created:
|
|
self.stdout.write(f" Adăugat Monitor: {mon.equipment_code}")
|
|
except FileNotFoundError:
|
|
self.stdout.write(
|
|
self.style.ERROR("Eroare: Nu am găsit fișierul Monitor.csv")
|
|
)
|
|
|
|
# 4. IMPORTUL PERIFERICELOR
|
|
self.stdout.write("\nÎncepe importul perifericelor...")
|
|
try:
|
|
with open("Peripheral.csv", "r", encoding="utf-8-sig") as file:
|
|
reader = csv.DictReader(file, delimiter=CSV_DELIMITER)
|
|
for row in reader:
|
|
posesor = employee_map.get(row.get("EmployeeID"))
|
|
|
|
per, created = Peripheral.objects.get_or_create(
|
|
equipment_code=row["equipment_code"],
|
|
defaults={
|
|
"entry_date": parse_date(row.get("entry_date")),
|
|
"peripheral_type": row.get("peripheral_type", ""),
|
|
"name": row.get("name", ""),
|
|
"serial_number": row.get("serial_number", ""),
|
|
"observations": row.get("observations", ""),
|
|
"status": "activ",
|
|
"assigned_to": posesor,
|
|
},
|
|
)
|
|
if created:
|
|
self.stdout.write(f" Adăugat Periferic: {per.equipment_code}")
|
|
except FileNotFoundError:
|
|
self.stdout.write(
|
|
self.style.ERROR("Eroare: Nu am găsit fișierul Peripheral.csv")
|
|
)
|
|
|
|
self.stdout.write(self.style.SUCCESS("\nImport finalizat cu succes!"))
|