feat(customers): add initial admin bootstrap and csv import
This commit is contained in:
DS | Schubert
parent
c816e9869d
commit
ecab0fe6b6
27 changed files with 1197 additions and 34 deletions
407
backend/hermes/app/services/customer_import_service.py
Normal file
407
backend/hermes/app/services/customer_import_service.py
Normal file
|
|
@ -0,0 +1,407 @@
|
|||
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
|
||||
Loading…
Add table
Add a link
Reference in a new issue