289 lines
14 KiB
Python
289 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import zipfile
|
|
from datetime import date, datetime
|
|
from decimal import Decimal, InvalidOperation
|
|
from io import BytesIO, StringIO
|
|
from typing import Any, Iterable
|
|
from xml.sax.saxutils import escape
|
|
|
|
from django.db import transaction
|
|
from django.http import HttpResponse
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.cell import WriteOnlyCell
|
|
|
|
from .models import AuditLog, ChoiceValue, Device, Employee, NumberSequence
|
|
from .device_types import TRISTATE_FIELDS, clear_inapplicable
|
|
from .services import add_audit
|
|
|
|
|
|
COLUMN_MAP = {
|
|
"kamera": "camera",
|
|
"sound": "sound",
|
|
"hdmi": "hdmi",
|
|
"vga": "vga",
|
|
"dvi": "dvi",
|
|
"dp": "displayport",
|
|
"displayport": "displayport", "seitenverhältnis": "aspect_ratio", "seitenverhaeltnis": "aspect_ratio",
|
|
"nr.": "number", "nr": "number", "nummer": "number", "regal": "shelf", "status": "status",
|
|
"datum": "date", "empfänger": "recipient", "empfaenger": "recipient", "ort": "location",
|
|
"ticket": "ticket", "typ": "device_type", "preis": "price", "akku": "battery", "wlan": "wifi",
|
|
"lan": "lan", "zoll": "display_inches", "ram": "ram_gb", "ssd": "ssd_gb", "hdd": "hdd_gb",
|
|
"cpu": "cpu", "hwid": "hwid", "anschlüsse": "ports", "anschluesse": "ports",
|
|
"auflösung": "resolution", "aufloesung": "resolution", "druckverfahren": "print_technology",
|
|
"duplex": "duplex", "farbdruck": "color_printing",
|
|
"modell": "model", "bearbeiter": "employee", "spender": "donor", "kommentar": "comment",
|
|
}
|
|
EXPORT_COLUMNS = [
|
|
("number", "Nr."), ("shelf", "Regal"), ("status", "Status"), ("date", "Datum"),
|
|
("recipient", "Empfänger"), ("location", "Ort"), ("ticket", "Ticket"), ("device_type", "Typ"),
|
|
("price", "Preis"), ("battery", "Akku"), ("wifi", "WLAN"), ("lan", "LAN"),
|
|
("display_inches", "Zoll"), ("ram_gb", "RAM"), ("ssd_gb", "SSD"), ("hdd_gb", "HDD"),
|
|
("cpu", "CPU"), ("hwid", "HWID"), ("ports", "Anschlüsse"), ("resolution", "Auflösung"),
|
|
("print_technology", "Druckverfahren"), ("duplex", "Duplex"), ("color_printing", "Farbdruck"),
|
|
("model", "Modell"), ("employee", "Bearbeiter"), ("donor", "Spender"),
|
|
("comment", "Kommentar"),
|
|
("camera", "Kamera"),
|
|
("sound", "Sound"),
|
|
("hdmi", "HDMI"),
|
|
("vga", "VGA"),
|
|
("dvi", "DVI"),
|
|
("displayport", "DP"),
|
|
("aspect_ratio", "Seitenverhältnis"),
|
|
]
|
|
|
|
|
|
def normalize_header(value: Any) -> str:
|
|
return str(value or "").strip().lower()
|
|
|
|
|
|
def analyze_workbook(file_obj) -> tuple[list[str], list[list[str]]]:
|
|
workbook = load_workbook(file_obj, read_only=True, data_only=True)
|
|
sheet = workbook.active
|
|
rows = sheet.iter_rows(values_only=True)
|
|
headers = [str(value or "").strip() for value in next(rows, [])]
|
|
preview = [[format_cell(value) for value in row[: len(headers)]] for _, row in zip(range(10), rows)]
|
|
if not headers or not any(headers):
|
|
raise ValueError("Die Datei enthält keine Kopfzeile.")
|
|
return headers, preview
|
|
|
|
|
|
def suggested_mapping(headers: list[str]) -> dict[int, str]:
|
|
return {index: COLUMN_MAP.get(normalize_header(header), "") for index, header in enumerate(headers)}
|
|
|
|
|
|
def format_cell(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, (date, datetime)):
|
|
return value.isoformat()
|
|
return str(value)
|
|
|
|
|
|
def parse_date(value: Any) -> date:
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
text = str(value).strip()
|
|
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y"):
|
|
try:
|
|
return datetime.strptime(text, fmt).date()
|
|
except ValueError:
|
|
continue
|
|
raise ValueError(f"Ungültiges Datum: {text}")
|
|
|
|
|
|
def decimal_value(value: Any) -> Decimal | None:
|
|
if value in (None, ""):
|
|
return None
|
|
try:
|
|
parsed = Decimal(str(value).replace("€", "").replace(" ", "").replace(",", "."))
|
|
if not parsed.is_finite():
|
|
raise InvalidOperation
|
|
return parsed
|
|
except InvalidOperation as exc:
|
|
raise ValueError(f"Ungültige Zahl: {value}") from exc
|
|
|
|
|
|
def integer_value(value: Any) -> int | None:
|
|
decimal = decimal_value(value)
|
|
if decimal is not None and decimal != decimal.to_integral_value():
|
|
raise ValueError(f"Eine ganze Zahl ist erforderlich: {value}")
|
|
return int(decimal) if decimal is not None else None
|
|
|
|
|
|
def tri_state(value: Any) -> str:
|
|
normalized = str(value or "").strip().upper()
|
|
return {"JA": Device.TriState.YES, "YES": Device.TriState.YES, "NEIN": Device.TriState.NO, "NO": Device.TriState.NO}.get(normalized, Device.TriState.UNKNOWN)
|
|
|
|
|
|
def resolve_choice(kind: str, value: Any, *, create: bool = True) -> ChoiceValue | None:
|
|
label = str(value or "").strip()
|
|
if not label:
|
|
return None
|
|
obj = ChoiceValue.objects.filter(kind=kind, label__iexact=label).first()
|
|
if obj:
|
|
return obj
|
|
if not create:
|
|
raise ValueError(f"Unbekannter Auswahlwert: {label}")
|
|
return ChoiceValue.objects.create(kind=kind, label=label, active=True)
|
|
|
|
|
|
@transaction.atomic
|
|
def import_workbook(file_obj, mapping: dict[int, str], user) -> dict[str, Any]:
|
|
workbook = load_workbook(file_obj, read_only=True, data_only=True)
|
|
sheet = workbook.active
|
|
next(sheet.iter_rows(values_only=True), None)
|
|
created = 0
|
|
errors: list[str] = []
|
|
maximum = 0
|
|
for row_number, row in enumerate(sheet.iter_rows(values_only=True, min_row=2), start=2):
|
|
if not any(value not in (None, "") for value in row):
|
|
continue
|
|
data = {field: row[index] if index < len(row) else None for index, field in mapping.items() if field}
|
|
try:
|
|
number = integer_value(data.get("number"))
|
|
if number and Device.objects.filter(number=number).exists():
|
|
raise ValueError(f"Gerätenummer {number} existiert bereits")
|
|
status = resolve_choice(ChoiceValue.Kind.STATUS, data.get("status") or "Eingang")
|
|
employee = None
|
|
employee_code = str(data.get("employee") or "").strip()
|
|
if employee_code:
|
|
employee = Employee.objects.filter(code__iexact=employee_code).first()
|
|
if not employee:
|
|
raise ValueError(f"Unbekanntes Mitarbeiterkürzel: {employee_code}")
|
|
device = Device(
|
|
number=number or 0, shelf=str(data.get("shelf") or "").strip(), status=status,
|
|
date=parse_date(data["date"]) if data.get("date") else date.today(),
|
|
recipient=resolve_choice(ChoiceValue.Kind.RECIPIENT, data.get("recipient")),
|
|
location=resolve_choice(ChoiceValue.Kind.LOCATION, data.get("location")),
|
|
ticket=str(data.get("ticket") or "").strip(),
|
|
device_type=resolve_choice(ChoiceValue.Kind.DEVICE_TYPE, data.get("device_type")),
|
|
price=decimal_value(data.get("price")), battery=tri_state(data.get("battery")),
|
|
wifi=tri_state(data.get("wifi")), lan=tri_state(data.get("lan")),
|
|
display_inches=decimal_value(data.get("display_inches")), ram_gb=integer_value(data.get("ram_gb")),
|
|
ssd_gb=integer_value(data.get("ssd_gb")), hdd_gb=integer_value(data.get("hdd_gb")),
|
|
cpu=str(data.get("cpu") or "").strip(), hwid=str(data.get("hwid") or "").strip(),
|
|
ports=str(data.get("ports") or "").strip(), resolution=str(data.get("resolution") or "").strip(),
|
|
print_technology=str(data.get("print_technology") or "").strip(),
|
|
duplex=tri_state(data.get("duplex")), color_printing=tri_state(data.get("color_printing")),
|
|
camera=tri_state(data.get("camera")),
|
|
sound=tri_state(data.get("sound")),
|
|
hdmi=tri_state(data.get("hdmi")),
|
|
vga=tri_state(data.get("vga")),
|
|
dvi=tri_state(data.get("dvi")),
|
|
displayport=tri_state(data.get("displayport")),
|
|
aspect_ratio=str(data.get("aspect_ratio") or "").strip(),
|
|
model=str(data.get("model") or "").strip(),
|
|
employee=employee, donor=str(data.get("donor") or "").strip(), comment=str(data.get("comment") or "").strip(),
|
|
)
|
|
clear_inapplicable(device)
|
|
device.full_clean(exclude=["number", "custom_values"])
|
|
device.save()
|
|
maximum = max(maximum, device.number)
|
|
add_audit(user, device, AuditLog.Action.IMPORT, new="Excel-Import")
|
|
created += 1
|
|
except Exception as exc:
|
|
errors.append(f"Zeile {row_number}: {exc}")
|
|
if errors:
|
|
raise ValueError("\n".join(errors[:50]))
|
|
NumberSequence.ensure_at_least(maximum)
|
|
return {"created": created, "errors": []}
|
|
|
|
|
|
def export_value(device: Device, field: str) -> Any:
|
|
value = getattr(device, field)
|
|
if field in TRISTATE_FIELDS:
|
|
return dict(Device.TriState.choices).get(value, value)
|
|
if hasattr(value, "label"):
|
|
return value.label
|
|
if hasattr(value, "code"):
|
|
return value.code
|
|
return value if value is not None else ""
|
|
|
|
|
|
def csv_safe_value(value: Any) -> Any:
|
|
if isinstance(value, str) and (value.lstrip().startswith(("=", "+", "-", "@")) or value.startswith(("\t", "\r", "\n"))):
|
|
return "'" + value
|
|
return value
|
|
|
|
|
|
def xlsx_safe_value(sheet, value: Any):
|
|
if isinstance(value, str):
|
|
cell = WriteOnlyCell(sheet, value=value)
|
|
cell.data_type = "s"
|
|
return cell
|
|
return value
|
|
|
|
|
|
def csv_response(devices: Iterable[Device]) -> HttpResponse:
|
|
response = HttpResponse(content_type="text/csv; charset=utf-8")
|
|
response["Content-Disposition"] = 'attachment; filename="geraete.csv"'
|
|
response.write("\ufeff")
|
|
writer = csv.writer(response, delimiter=";")
|
|
writer.writerow([label for _, label in EXPORT_COLUMNS])
|
|
for device in devices.iterator(chunk_size=1000):
|
|
writer.writerow([csv_safe_value(export_value(device, field)) for field, _ in EXPORT_COLUMNS])
|
|
return response
|
|
|
|
|
|
def xlsx_response(devices: Iterable[Device]) -> HttpResponse:
|
|
workbook = Workbook(write_only=True)
|
|
sheet = workbook.create_sheet("Geräte")
|
|
sheet.append([label for _, label in EXPORT_COLUMNS])
|
|
for device in devices.iterator(chunk_size=1000):
|
|
sheet.append([xlsx_safe_value(sheet, export_value(device, field)) for field, _ in EXPORT_COLUMNS])
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
response = HttpResponse(output.getvalue(), content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
|
response["Content-Disposition"] = 'attachment; filename="geraete.xlsx"'
|
|
return response
|
|
|
|
|
|
def _ods_cell(value: Any, *, header: bool = False) -> str:
|
|
style = ' table:style-name="Kopfzeile"' if header else ""
|
|
if isinstance(value, datetime):
|
|
value = value.date()
|
|
if isinstance(value, date):
|
|
return f'<table:table-cell{style} office:value-type="date" office:date-value="{value.isoformat()}"><text:p>{value.strftime("%d.%m.%Y")}</text:p></table:table-cell>'
|
|
if isinstance(value, (int, float, Decimal)) and not isinstance(value, bool):
|
|
number = str(value)
|
|
return f'<table:table-cell{style} office:value-type="float" office:value="{number}"><text:p>{escape(number)}</text:p></table:table-cell>'
|
|
text = escape(str(value if value is not None else ""))
|
|
return f'<table:table-cell{style} office:value-type="string"><text:p>{text}</text:p></table:table-cell>'
|
|
|
|
|
|
def ods_response(devices: Iterable[Device]) -> HttpResponse:
|
|
content = StringIO()
|
|
content.write('<?xml version="1.0" encoding="UTF-8"?>')
|
|
content.write('<office:document-content office:version="1.2" xmlns:office="urn:oasis:names:tc:opendocument:xmlns:office:1.0" xmlns:table="urn:oasis:names:tc:opendocument:xmlns:table:1.0" xmlns:text="urn:oasis:names:tc:opendocument:xmlns:text:1.0" xmlns:style="urn:oasis:names:tc:opendocument:xmlns:style:1.0" xmlns:fo="urn:oasis:names:tc:opendocument:xmlns:xsl-fo-compatible:1.0">')
|
|
content.write('<office:automatic-styles><style:style style:name="Kopfzeile" style:family="table-cell"><style:text-properties fo:font-weight="bold"/></style:style></office:automatic-styles>')
|
|
content.write('<office:body><office:spreadsheet><table:table table:name="Geräte">')
|
|
content.write('<table:table-row>')
|
|
for _, label in EXPORT_COLUMNS:
|
|
content.write(_ods_cell(label, header=True))
|
|
content.write('</table:table-row>')
|
|
for device in devices.iterator(chunk_size=1000):
|
|
content.write('<table:table-row>')
|
|
for field, _ in EXPORT_COLUMNS:
|
|
content.write(_ods_cell(export_value(device, field)))
|
|
content.write('</table:table-row>')
|
|
content.write('</table:table></office:spreadsheet></office:body></office:document-content>')
|
|
|
|
mime_type = "application/vnd.oasis.opendocument.spreadsheet"
|
|
manifest = f'''<?xml version="1.0" encoding="UTF-8"?>
|
|
<manifest:manifest manifest:version="1.2" xmlns:manifest="urn:oasis:names:tc:opendocument:xmlns:manifest:1.0">
|
|
<manifest:file-entry manifest:full-path="/" manifest:version="1.2" manifest:media-type="{mime_type}"/>
|
|
<manifest:file-entry manifest:full-path="content.xml" manifest:media-type="text/xml"/>
|
|
</manifest:manifest>'''
|
|
output = BytesIO()
|
|
with zipfile.ZipFile(output, "w") as archive:
|
|
archive.writestr("mimetype", mime_type, compress_type=zipfile.ZIP_STORED)
|
|
archive.writestr("content.xml", content.getvalue(), compress_type=zipfile.ZIP_DEFLATED)
|
|
archive.writestr("META-INF/manifest.xml", manifest, compress_type=zipfile.ZIP_DEFLATED)
|
|
response = HttpResponse(output.getvalue(), content_type=mime_type)
|
|
response["Content-Disposition"] = 'attachment; filename="geraete.ods"'
|
|
return response
|