407 lines
15 KiB
Python
407 lines
15 KiB
Python
import csv
|
|
import io
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
|
|
from pydantic import EmailStr, HttpUrl, TypeAdapter, ValidationError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.customer import Customer, CustomerAddress, CustomerContact
|
|
from app.models.user import User
|
|
from app.repositories.customer_repository import CustomerRepository
|
|
from app.schemas.customer_import import (
|
|
CustomerImportCommitResponse,
|
|
CustomerImportIssue,
|
|
CustomerImportPreviewResponse,
|
|
CustomerImportPreviewRow,
|
|
CustomerImportSummary,
|
|
ImportMode,
|
|
)
|
|
from app.services.audit_service import write_audit_log
|
|
|
|
CSV_COLUMNS = [
|
|
"customer_number",
|
|
"company_name",
|
|
"legal_name",
|
|
"customer_type",
|
|
"status",
|
|
"industry",
|
|
"website",
|
|
"email",
|
|
"phone",
|
|
"tax_number",
|
|
"vat_id",
|
|
"notes",
|
|
"address_type",
|
|
"street",
|
|
"postal_code",
|
|
"city",
|
|
"state",
|
|
"country",
|
|
"address_is_primary",
|
|
"contact_first_name",
|
|
"contact_last_name",
|
|
"contact_position",
|
|
"contact_email",
|
|
"contact_phone",
|
|
"contact_mobile",
|
|
"contact_is_primary",
|
|
"contact_notes",
|
|
]
|
|
|
|
VALID_STATUSES = {"lead", "active", "inactive", "blocked", "archived"}
|
|
VALID_TYPES = {"company", "private", "public_sector", "partner", "supplier"}
|
|
VALID_ADDRESS_TYPES = {"billing", "shipping", "primary", "other"}
|
|
TRUE_VALUES = {"1", "true", "yes", "ja", "j", "x"}
|
|
FALSE_VALUES = {"0", "false", "no", "nein", "n", ""}
|
|
|
|
email_adapter = TypeAdapter(EmailStr)
|
|
url_adapter = TypeAdapter(HttpUrl)
|
|
|
|
|
|
@dataclass
|
|
class ParsedImportRow:
|
|
row_number: int
|
|
data: dict[str, str]
|
|
action: str
|
|
errors: list[CustomerImportIssue]
|
|
warnings: list[CustomerImportIssue]
|
|
existing_customer: Customer | None = None
|
|
|
|
|
|
def get_customer_import_template() -> str:
|
|
return ";".join(CSV_COLUMNS) + "\n"
|
|
|
|
|
|
def decode_csv(content: bytes) -> str:
|
|
try:
|
|
return content.decode("utf-8-sig")
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError("CSV-Datei muss UTF-8-kodiert sein") from exc
|
|
|
|
|
|
def parse_bool(value: str) -> bool:
|
|
normalized = value.strip().lower()
|
|
if normalized in TRUE_VALUES:
|
|
return True
|
|
if normalized in FALSE_VALUES:
|
|
return False
|
|
return False
|
|
|
|
|
|
def parse_bool_default(value: str, default: bool = False) -> bool:
|
|
if value.strip() == "":
|
|
return default
|
|
return parse_bool(value)
|
|
|
|
|
|
def sniff_dialect(text: str) -> csv.Dialect:
|
|
sample = text[:4096]
|
|
try:
|
|
return csv.Sniffer().sniff(sample, delimiters=",;")
|
|
except csv.Error:
|
|
return csv.excel
|
|
|
|
|
|
def read_csv_rows(content: bytes) -> list[tuple[int, dict[str, str]]]:
|
|
text = decode_csv(content)
|
|
reader = csv.DictReader(io.StringIO(text), dialect=sniff_dialect(text))
|
|
rows: list[tuple[int, dict[str, str]]] = []
|
|
|
|
for index, row in enumerate(reader, start=2):
|
|
normalized = {column: (row.get(column) or "").strip() for column in CSV_COLUMNS}
|
|
if not any(normalized.values()):
|
|
continue
|
|
rows.append((index, normalized))
|
|
|
|
return rows
|
|
|
|
|
|
def validate_email(value: str, row_number: int, field: str, errors: list[CustomerImportIssue]) -> str:
|
|
if not value:
|
|
return ""
|
|
try:
|
|
return str(email_adapter.validate_python(value))
|
|
except ValidationError:
|
|
errors.append(CustomerImportIssue(row=row_number, field=field, message="E-Mail ist ungültig"))
|
|
return value
|
|
|
|
|
|
def validate_url(value: str, row_number: int, errors: list[CustomerImportIssue]) -> str:
|
|
if not value:
|
|
return ""
|
|
candidate = value if value.startswith(("http://", "https://")) else f"https://{value}"
|
|
try:
|
|
return str(url_adapter.validate_python(candidate))
|
|
except ValidationError:
|
|
errors.append(CustomerImportIssue(row=row_number, field="website", message="Website ist ungültig"))
|
|
return value
|
|
|
|
|
|
def planned_action(mode: ImportMode, existing_customer: Customer | None) -> str:
|
|
if existing_customer is None:
|
|
return "create" if mode in {"create_only", "upsert"} else "skip"
|
|
if mode == "create_only":
|
|
return "skip"
|
|
return "update"
|
|
|
|
|
|
class CustomerImportService:
|
|
@staticmethod
|
|
def preview(db: Session, content: bytes, mode: ImportMode) -> CustomerImportPreviewResponse:
|
|
rows = CustomerImportService._parse_and_validate(db, content, mode)
|
|
return CustomerImportService._build_preview(mode, rows)
|
|
|
|
@staticmethod
|
|
def commit(
|
|
db: Session,
|
|
content: bytes,
|
|
mode: ImportMode,
|
|
*,
|
|
actor: User,
|
|
request,
|
|
) -> CustomerImportCommitResponse:
|
|
rows = CustomerImportService._parse_and_validate(db, content, mode)
|
|
preview = CustomerImportService._build_preview(mode, rows)
|
|
if preview.summary.valid_rows == 0:
|
|
return CustomerImportCommitResponse(
|
|
mode=mode,
|
|
summary=preview.summary,
|
|
created=0,
|
|
updated=0,
|
|
skipped=preview.summary.skip_count,
|
|
)
|
|
|
|
created = 0
|
|
updated = 0
|
|
skipped = 0
|
|
|
|
for row in rows:
|
|
if row.errors or row.action == "error":
|
|
continue
|
|
if row.action == "skip":
|
|
skipped += 1
|
|
continue
|
|
if row.action == "create":
|
|
CustomerImportService._create_customer(db, row.data)
|
|
created += 1
|
|
elif row.action == "update" and row.existing_customer is not None:
|
|
CustomerImportService._update_customer(db, row.existing_customer, row.data)
|
|
updated += 1
|
|
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="customers.import",
|
|
entity_type="customers",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={
|
|
"mode": mode,
|
|
"total_rows": preview.summary.total_rows,
|
|
"created": created,
|
|
"updated": updated,
|
|
"skipped": skipped,
|
|
"errors": preview.summary.error_count,
|
|
"warnings": preview.summary.warning_count,
|
|
},
|
|
)
|
|
summary = CustomerImportService._build_preview(mode, rows).summary
|
|
return CustomerImportCommitResponse(
|
|
mode=mode,
|
|
summary=summary,
|
|
created=created,
|
|
updated=updated,
|
|
skipped=skipped,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_and_validate(db: Session, content: bytes, mode: ImportMode) -> list[ParsedImportRow]:
|
|
csv_rows = read_csv_rows(content)
|
|
number_rows: dict[str, int] = {}
|
|
duplicates: set[str] = set()
|
|
|
|
for row_number, data in csv_rows:
|
|
customer_number = data["customer_number"]
|
|
if not customer_number:
|
|
continue
|
|
if customer_number in number_rows:
|
|
duplicates.add(customer_number)
|
|
else:
|
|
number_rows[customer_number] = row_number
|
|
|
|
parsed_rows: list[ParsedImportRow] = []
|
|
for row_number, data in csv_rows:
|
|
errors: list[CustomerImportIssue] = []
|
|
warnings: list[CustomerImportIssue] = []
|
|
|
|
if not data["company_name"]:
|
|
errors.append(CustomerImportIssue(row=row_number, field="company_name", message="Firmenname ist erforderlich"))
|
|
|
|
if not data["customer_type"]:
|
|
data["customer_type"] = "company"
|
|
warnings.append(CustomerImportIssue(row=row_number, field="customer_type", message="Typ fehlt und wird als company importiert"))
|
|
elif data["customer_type"] not in VALID_TYPES:
|
|
errors.append(CustomerImportIssue(row=row_number, field="customer_type", message="Typ ist ungültig"))
|
|
|
|
if not data["status"]:
|
|
data["status"] = "active"
|
|
warnings.append(CustomerImportIssue(row=row_number, field="status", message="Status fehlt und wird als active importiert"))
|
|
elif data["status"] not in VALID_STATUSES:
|
|
errors.append(CustomerImportIssue(row=row_number, field="status", message="Status ist ungültig"))
|
|
|
|
if not data["address_type"]:
|
|
data["address_type"] = "primary"
|
|
elif data["address_type"] not in VALID_ADDRESS_TYPES:
|
|
errors.append(CustomerImportIssue(row=row_number, field="address_type", message="Adresstyp ist ungültig"))
|
|
|
|
data["email"] = validate_email(data["email"], row_number, "email", errors)
|
|
data["contact_email"] = validate_email(data["contact_email"], row_number, "contact_email", errors)
|
|
data["website"] = validate_url(data["website"], row_number, errors)
|
|
|
|
if data["customer_number"] and data["customer_number"] in duplicates:
|
|
errors.append(CustomerImportIssue(row=row_number, field="customer_number", message="Kundennummer ist in der CSV mehrfach vorhanden"))
|
|
|
|
if not data["customer_number"] and mode == "update_existing":
|
|
errors.append(CustomerImportIssue(row=row_number, field="customer_number", message="Kundennummer ist für update_existing erforderlich"))
|
|
|
|
if not data["customer_number"]:
|
|
warnings.append(CustomerImportIssue(row=row_number, field="customer_number", message="Kundennummer fehlt und wird beim Erstellen generiert"))
|
|
|
|
existing_customer = (
|
|
CustomerRepository.get_by_number(db, data["customer_number"])
|
|
if data["customer_number"]
|
|
else None
|
|
)
|
|
action = planned_action(mode, existing_customer)
|
|
if action == "skip":
|
|
reason = "Kunde existiert bereits" if existing_customer is not None else "Kunde existiert nicht"
|
|
warnings.append(CustomerImportIssue(row=row_number, field="customer_number", message=f"Zeile wird übersprungen: {reason}"))
|
|
|
|
if errors:
|
|
action = "error"
|
|
|
|
parsed_rows.append(
|
|
ParsedImportRow(
|
|
row_number=row_number,
|
|
data=data,
|
|
action=action,
|
|
errors=errors,
|
|
warnings=warnings,
|
|
existing_customer=existing_customer,
|
|
)
|
|
)
|
|
|
|
return parsed_rows
|
|
|
|
@staticmethod
|
|
def _build_preview(mode: ImportMode, rows: list[ParsedImportRow]) -> CustomerImportPreviewResponse:
|
|
preview_rows = [
|
|
CustomerImportPreviewRow(
|
|
row=row.row_number,
|
|
customer_number=row.data["customer_number"],
|
|
company_name=row.data["company_name"],
|
|
action=row.action,
|
|
errors=row.errors,
|
|
warnings=row.warnings,
|
|
)
|
|
for row in rows
|
|
]
|
|
errors = [issue for row in rows for issue in row.errors]
|
|
warnings = [issue for row in rows for issue in row.warnings]
|
|
duplicates = [issue for issue in errors if issue.field == "customer_number" and "mehrfach" in issue.message]
|
|
summary = CustomerImportSummary(
|
|
total_rows=len(rows),
|
|
valid_rows=sum(1 for row in rows if not row.errors and row.action in {"create", "update"}),
|
|
error_count=len(errors),
|
|
warning_count=len(warnings),
|
|
duplicate_count=len(duplicates),
|
|
create_count=sum(1 for row in rows if not row.errors and row.action == "create"),
|
|
update_count=sum(1 for row in rows if not row.errors and row.action == "update"),
|
|
skip_count=sum(1 for row in rows if not row.errors and row.action == "skip"),
|
|
)
|
|
return CustomerImportPreviewResponse(
|
|
mode=mode,
|
|
summary=summary,
|
|
rows=preview_rows,
|
|
errors=errors,
|
|
warnings=warnings,
|
|
duplicates=duplicates,
|
|
)
|
|
|
|
@staticmethod
|
|
def _customer_number(data: dict[str, str]) -> str:
|
|
return data["customer_number"] or f"IMP-{uuid.uuid4().hex[:10].upper()}"
|
|
|
|
@staticmethod
|
|
def _address(data: dict[str, str]) -> CustomerAddress:
|
|
return CustomerAddress(
|
|
type=data["address_type"],
|
|
street=data["street"],
|
|
postal_code=data["postal_code"],
|
|
city=data["city"],
|
|
state=data["state"],
|
|
country=data["country"] or "Deutschland",
|
|
is_primary=parse_bool_default(data["address_is_primary"], default=True),
|
|
)
|
|
|
|
@staticmethod
|
|
def _contact(data: dict[str, str]) -> CustomerContact | None:
|
|
if not any(data[key] for key in ("contact_first_name", "contact_last_name", "contact_email", "contact_phone", "contact_mobile")):
|
|
return None
|
|
return CustomerContact(
|
|
first_name=data["contact_first_name"],
|
|
last_name=data["contact_last_name"],
|
|
position=data["contact_position"],
|
|
email=data["contact_email"],
|
|
phone=data["contact_phone"],
|
|
mobile=data["contact_mobile"],
|
|
is_primary=parse_bool(data["contact_is_primary"]),
|
|
notes=data["contact_notes"],
|
|
)
|
|
|
|
@staticmethod
|
|
def _create_customer(db: Session, data: dict[str, str]) -> Customer:
|
|
customer = Customer(
|
|
customer_number=CustomerImportService._customer_number(data),
|
|
company_name=data["company_name"],
|
|
legal_name=data["legal_name"],
|
|
customer_type=data["customer_type"],
|
|
status=data["status"],
|
|
industry=data["industry"],
|
|
website=data["website"],
|
|
email=data["email"],
|
|
phone=data["phone"],
|
|
tax_number=data["tax_number"],
|
|
vat_id=data["vat_id"],
|
|
notes=data["notes"],
|
|
addresses=[CustomerImportService._address(data)],
|
|
)
|
|
contact = CustomerImportService._contact(data)
|
|
if contact is not None:
|
|
customer.contacts = [contact]
|
|
db.add(customer)
|
|
return customer
|
|
|
|
@staticmethod
|
|
def _update_customer(db: Session, customer: Customer, data: dict[str, str]) -> Customer:
|
|
customer.company_name = data["company_name"]
|
|
customer.legal_name = data["legal_name"]
|
|
customer.customer_type = data["customer_type"]
|
|
customer.status = data["status"]
|
|
customer.industry = data["industry"]
|
|
customer.website = data["website"]
|
|
customer.email = data["email"]
|
|
customer.phone = data["phone"]
|
|
customer.tax_number = data["tax_number"]
|
|
customer.vat_id = data["vat_id"]
|
|
customer.notes = data["notes"]
|
|
customer.addresses = [CustomerImportService._address(data)]
|
|
contact = CustomerImportService._contact(data)
|
|
if contact is not None:
|
|
if contact.is_primary:
|
|
for existing_contact in customer.contacts:
|
|
existing_contact.is_primary = False
|
|
customer.contacts.append(contact)
|
|
db.add(customer)
|
|
return customer
|