219 lines
9.5 KiB
Python
219 lines
9.5 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
|
|
|
|
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()
|
|
|
|
# Elimină partea de timp (0:00:00) dacă există
|
|
if " " in clean_str:
|
|
clean_str = clean_str.split(" ")[0]
|
|
|
|
# Încearcă să parseze formatul DD-M-YYYY (cu sau fără leading zeros)
|
|
try:
|
|
# Split după '-'
|
|
parts = clean_str.split("-")
|
|
if len(parts) == 3:
|
|
day = parts[0].zfill(2) # Adaugă zero în față dacă e necesar
|
|
month = parts[1].zfill(2) # Adaugă zero în față dacă e necesar
|
|
year = parts[2]
|
|
|
|
# Verifică dacă anul are 4 cifre
|
|
if len(year) == 2:
|
|
year = "20" + year
|
|
|
|
clean_str = f"{day}-{month}-{year}"
|
|
|
|
parsed = datetime.strptime(clean_str, "%d-%m-%Y")
|
|
return parsed.strftime("%Y-%m-%d")
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Încearcă și cu slash ca separator
|
|
try:
|
|
parts = clean_str.split("/")
|
|
if len(parts) == 3:
|
|
day = parts[0].zfill(2)
|
|
month = parts[1].zfill(2)
|
|
year = parts[2]
|
|
|
|
if len(year) == 2:
|
|
year = "20" + year
|
|
|
|
clean_str = f"{day}/{month}/{year}"
|
|
|
|
parsed = datetime.strptime(clean_str, "%d/%m/%Y")
|
|
return parsed.strftime("%Y-%m-%d")
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Dacă data există dar formatul e complet invalid, punem fallback
|
|
self.stdout.write(
|
|
self.style.WARNING(
|
|
f"Atenție: Data '{date_str}' e invalidă. Folosim {fallback_date}."
|
|
)
|
|
)
|
|
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!"))
|