import json from dataclasses import dataclass from datetime import UTC, datetime from decimal import Decimal, ROUND_HALF_UP from urllib.error import HTTPError, URLError from urllib.request import Request as UrlRequest from urllib.request import urlopen from fastapi import HTTPException, status from sqlalchemy.orm import Session from starlette.requests import Request from app.core.config import settings from app.models.lexware import LexwareSyncRecord from app.models.repair import Repair from app.models.repair_estimate import RepairEstimate from app.models.user import User from app.repositories.system_settings_repository import SystemSettingsRepository from app.schemas.lexware import ( AccountingTransferUpdate, LexwareCustomerMapping, LexwareInvoicePreparationResponse, LexwareLineItemMapping, LexwareSettingsResponse, LexwareSettingsUpdate, LexwareTestConnectionResponse, ) from app.schemas.system_setting import SettingsSource from app.services.audit_service import write_audit_log from app.services.system_settings_service import parse_bool LEXWARE_KEYS = ( "lexware.enabled", "lexware.api_base_url", "lexware.api_key", "lexware.organization_name", "lexware.default_tax_rate", "lexware.default_payment_terms_days", ) LEXWARE_SECRET_KEYS = {"lexware.api_key"} DEFAULT_API_BASE_URL = "https://api.lexware.io" @dataclass(frozen=True) class LexwareRuntimeConfig: enabled: bool api_base_url: str api_key: str organization_name: str default_tax_rate: Decimal default_payment_terms_days: int source: SettingsSource @property def api_key_is_set(self) -> bool: return bool(self.api_key) @property def is_configured(self) -> bool: return self.enabled and bool(self.api_base_url and self.api_key) def _decimal(value: object, *, default: Decimal) -> Decimal: try: return Decimal(str(value or "").replace(",", ".")) except Exception: return default def _int(value: object, *, default: int) -> int: try: parsed = int(str(value or "").strip()) except ValueError: return default return parsed if 0 <= parsed <= 365 else default def _euros(cents: int) -> Decimal: return (Decimal(cents) / Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def _safe_error_message(error: Exception) -> str: if isinstance(error, HTTPError): if error.code in {401, 403}: return "Lexware hat den API-Key abgelehnt." if error.code == 404: return "Lexware-Endpunkt wurde nicht gefunden." return "Lexware hat die Anfrage nicht erfolgreich beantwortet." if isinstance(error, URLError): return "Lexware ist momentan nicht erreichbar." return "Lexware-Verbindungstest konnte nicht abgeschlossen werden." class LexwareService: @staticmethod def get_settings(db: Session) -> LexwareSettingsResponse: return LexwareService.settings_response(LexwareService.get_runtime_config(db)) @staticmethod def get_runtime_config(db: Session) -> LexwareRuntimeConfig: values = {key: setting.value.strip() for key, setting in SystemSettingsRepository.get_many(db, LEXWARE_KEYS).items()} db_enabled = parse_bool(values.get("lexware.enabled"), default=False) db_base_url = (values.get("lexware.api_base_url") or "").rstrip("/") db_api_key = values.get("lexware.api_key") or "" db_has_config = db_enabled or db_base_url or db_api_key if db_has_config: return LexwareRuntimeConfig( enabled=db_enabled, api_base_url=db_base_url or DEFAULT_API_BASE_URL, api_key=db_api_key, organization_name=values.get("lexware.organization_name", ""), default_tax_rate=_decimal(values.get("lexware.default_tax_rate"), default=Decimal("19.00")), default_payment_terms_days=_int(values.get("lexware.default_payment_terms_days"), default=14), source="database", ) if settings.lexware_enabled or settings.lexware_api_key: return LexwareRuntimeConfig( enabled=settings.lexware_enabled, api_base_url=(settings.lexware_api_base_url or DEFAULT_API_BASE_URL).rstrip("/"), api_key=settings.lexware_api_key or "", organization_name="", default_tax_rate=Decimal("19.00"), default_payment_terms_days=14, source="environment", ) return LexwareRuntimeConfig( enabled=False, api_base_url=DEFAULT_API_BASE_URL, api_key="", organization_name="", default_tax_rate=Decimal("19.00"), default_payment_terms_days=14, source="missing", ) @staticmethod def update_settings(db: Session, payload: LexwareSettingsUpdate, *, actor: User, request: Request) -> LexwareSettingsResponse: current_values = {key: setting.value.strip() for key, setting in SystemSettingsRepository.get_many(db, LEXWARE_KEYS).items()} api_key = payload.api_key if payload.api_key else current_values.get("lexware.api_key", "") updates = { "lexware.enabled": "true" if payload.enabled else "false", "lexware.api_base_url": payload.api_base_url.rstrip("/") or DEFAULT_API_BASE_URL, "lexware.api_key": api_key, "lexware.organization_name": payload.organization_name, "lexware.default_tax_rate": str(payload.default_tax_rate), "lexware.default_payment_terms_days": str(payload.default_payment_terms_days), } for key, value in updates.items(): SystemSettingsRepository.upsert(db, key=key, value=value, is_secret=key in LEXWARE_SECRET_KEYS) db.commit() write_audit_log( db, action="lexware.settings.update", entity_type="system_settings", entity_label="Lexware Office", actor=actor, request=request, metadata={ "enabled": payload.enabled, "api_base_url": payload.api_base_url, "organization_name": payload.organization_name, "default_tax_rate": str(payload.default_tax_rate), "default_payment_terms_days": payload.default_payment_terms_days, "api_key_changed": bool(payload.api_key), }, ) return LexwareService.get_settings(db) @staticmethod def test_connection(db: Session, *, actor: User, request: Request) -> LexwareTestConnectionResponse: config = LexwareService.get_runtime_config(db) if not config.is_configured: write_audit_log( db, action="lexware.connection.test_failed", entity_type="system_settings", entity_label="Lexware Office", actor=actor, request=request, metadata={"reason": "lexware_not_configured", "source": config.source}, ) return LexwareTestConnectionResponse( success=False, message="Lexware ist nicht vollständig konfiguriert.", source=config.source, api_base_url=config.api_base_url, organization_name=config.organization_name, ) try: profile = LexwareService._get_profile(config) except Exception as exc: write_audit_log( db, action="lexware.connection.test_failed", entity_type="system_settings", entity_label="Lexware Office", actor=actor, request=request, metadata={"reason": exc.__class__.__name__, "source": config.source, "api_base_url": config.api_base_url}, ) return LexwareTestConnectionResponse( success=False, message=_safe_error_message(exc), source=config.source, api_base_url=config.api_base_url, organization_name=config.organization_name, ) organization_name = config.organization_name or str(profile.get("organizationName") or profile.get("companyName") or "") write_audit_log( db, action="lexware.connection.test_success", entity_type="system_settings", entity_label="Lexware Office", actor=actor, request=request, metadata={"source": config.source, "api_base_url": config.api_base_url, "organization_name": organization_name}, ) return LexwareTestConnectionResponse( success=True, message="Lexware-Verbindung erfolgreich geprüft.", source=config.source, api_base_url=config.api_base_url, organization_name=organization_name, ) @staticmethod def prepare_invoice( db: Session, repair: Repair, estimate: RepairEstimate, *, actor: User, request: Request, ) -> LexwareInvoicePreparationResponse: config = LexwareService.get_runtime_config(db) if estimate.status != "approved": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Nur freigegebene Kostenvoranschläge können für Lexware vorbereitet werden") warnings: list[str] = [] if not config.enabled: warnings.append("Lexware ist noch nicht aktiviert. Der Export ist nur vorbereitet.") if not config.api_key_is_set: warnings.append("Lexware API-Key ist noch nicht gesetzt.") if not repair.customer_email: warnings.append("Beim Kunden ist keine E-Mail-Adresse hinterlegt.") if not estimate.items: warnings.append("Der Kostenvoranschlag enthält keine Positionen.") customer_payload = { "roles": {"customer": {}}, "company": {"name": repair.customer_name}, "emailAddresses": {"business": [repair.customer_email]} if repair.customer_email else {}, "phoneNumbers": {"business": [repair.customer_phone]} if repair.customer_phone else {}, } customer_mapping = LexwareCustomerMapping( name=repair.customer_name, email=repair.customer_email, phone=repair.customer_phone, search_strategy="email" if repair.customer_email else "name", create_payload=customer_payload, ) line_items = [ LexwareLineItemMapping( title=item.title, description=item.description, quantity=item.quantity, unit=item.unit, unit_price=_euros(item.unit_price_cents), tax_rate=estimate.tax_rate_percent, total=_euros(item.total_cents), ) for item in estimate.items ] payload_summary = LexwareService._invoice_payload_summary(config, repair, estimate, line_items) record = LexwareSyncRecord( entity_type="repair_estimate", entity_id=estimate.id, lexware_resource_type="invoice", status="pending" if not warnings else "skipped", direction="push", export_status="prepared", accounting_note=estimate.accounting_note, payload_summary=json.dumps(payload_summary, ensure_ascii=True), error_message="; ".join(warnings) if warnings else None, ) estimate.accounting_export_status = "prepared" db.add(record) db.commit() db.refresh(record) db.refresh(estimate) write_audit_log( db, action="accounting.invoice.handoff", entity_type="repair_estimates", entity_id=estimate.id, entity_label=f"{estimate.estimate_number} · {estimate.title}", actor=actor, request=request, metadata={ "repair_id": repair.id, "repair_number": repair.repair_number, "ready_for_export": not warnings, "sync_record_id": record.id, "export_status": record.export_status, }, ) return LexwareService._invoice_preparation_response( record=record, payload_summary=payload_summary, customer_mapping=customer_mapping, line_items=line_items, estimate=estimate, config=config, warnings=warnings, ) @staticmethod def mark_transferred( db: Session, repair: Repair, estimate: RepairEstimate, payload: AccountingTransferUpdate, *, actor: User, request: Request, ) -> LexwareInvoicePreparationResponse: record = LexwareService._latest_invoice_record(db, estimate.id) if record is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Keine Rechnungsvorbereitung gefunden") if record.export_status not in {"prepared", "transferred"}: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Diese Rechnungsvorbereitung kann nicht als übertragen markiert werden") note_changed = payload.accounting_note is not None and payload.accounting_note != (estimate.accounting_note or "") now = datetime.now(UTC) record.export_status = "transferred" record.status = "success" record.accounting_note = payload.accounting_note if payload.accounting_note is not None else record.accounting_note record.transferred_at = now record.transferred_by_user_id = actor.id record.synced_at = now estimate.accounting_export_status = "transferred" estimate.accounting_note = record.accounting_note estimate.accounting_transferred_at = now estimate.accounting_transferred_by_user_id = actor.id db.commit() db.refresh(record) db.refresh(estimate) write_audit_log( db, action="accounting.invoice.mark_transferred", entity_type="repair_estimates", entity_id=estimate.id, entity_label=f"{estimate.estimate_number} · {estimate.title}", actor=actor, request=request, metadata={ "repair_id": repair.id, "repair_number": repair.repair_number, "sync_record_id": record.id, "export_status": record.export_status, "transferred_at": record.transferred_at, }, ) if note_changed: write_audit_log( db, action="accounting.invoice.note_update", entity_type="repair_estimates", entity_id=estimate.id, entity_label=f"{estimate.estimate_number} · {estimate.title}", actor=actor, request=request, metadata={"repair_id": repair.id, "repair_number": repair.repair_number, "sync_record_id": record.id}, ) config = LexwareService.get_runtime_config(db) customer_mapping, line_items = LexwareService._invoice_mapping(repair, estimate) warnings: list[str] = [] payload_summary = LexwareService._invoice_payload_summary(config, repair, estimate, line_items) return LexwareService._invoice_preparation_response( record=record, payload_summary=payload_summary, customer_mapping=customer_mapping, line_items=line_items, estimate=estimate, config=config, warnings=warnings, ) @staticmethod def _invoice_mapping(repair: Repair, estimate: RepairEstimate) -> tuple[LexwareCustomerMapping, list[LexwareLineItemMapping]]: customer_payload = { "roles": {"customer": {}}, "company": {"name": repair.customer_name}, "emailAddresses": {"business": [repair.customer_email]} if repair.customer_email else {}, "phoneNumbers": {"business": [repair.customer_phone]} if repair.customer_phone else {}, } customer_mapping = LexwareCustomerMapping( name=repair.customer_name, email=repair.customer_email, phone=repair.customer_phone, search_strategy="email" if repair.customer_email else "name", create_payload=customer_payload, ) line_items = [ LexwareLineItemMapping( title=item.title, description=item.description, quantity=item.quantity, unit=item.unit, unit_price=_euros(item.unit_price_cents), tax_rate=estimate.tax_rate_percent, total=_euros(item.total_cents), ) for item in estimate.items ] return customer_mapping, line_items @staticmethod def _invoice_payload_summary( config: LexwareRuntimeConfig, repair: Repair, estimate: RepairEstimate, line_items: list[LexwareLineItemMapping], ) -> dict: return { "type": "invoice", "title": f"Rechnung zu Reparatur {repair.repair_number}", "introduction": f"Rechnung zu Reparatur {repair.repair_number} gemäß Kostenvoranschlag {estimate.estimate_number}.", "repair_number": repair.repair_number, "estimate_number": estimate.estimate_number, "currency": estimate.currency, "payment_terms_days": config.default_payment_terms_days, "subtotal": str(_euros(estimate.subtotal_cents)), "tax": str(_euros(estimate.tax_cents)), "total": str(_euros(estimate.total_cents)), "line_item_count": len(line_items), } @staticmethod def _invoice_preparation_response( *, record: LexwareSyncRecord, payload_summary: dict, customer_mapping: LexwareCustomerMapping, line_items: list[LexwareLineItemMapping], estimate: RepairEstimate, config: LexwareRuntimeConfig, warnings: list[str], ) -> LexwareInvoicePreparationResponse: return LexwareInvoicePreparationResponse( ready_for_export=not warnings, export_status=record.export_status, payload_summary=payload_summary, customer_mapping=customer_mapping, line_item_mapping=line_items, tax_mapping={ "source": "repair_estimate", "tax_rate": str(estimate.tax_rate_percent or config.default_tax_rate), "default_tax_rate": str(config.default_tax_rate), "tax_amount": str(_euros(estimate.tax_cents)), }, warnings=warnings, sync_record_id=record.id, accounting_note=record.accounting_note or "", transferred_at=record.transferred_at.isoformat() if record.transferred_at else None, transferred_by_user_id=record.transferred_by_user_id, ) @staticmethod def _latest_invoice_record(db: Session, estimate_id: int) -> LexwareSyncRecord | None: from sqlalchemy import select return db.scalar( select(LexwareSyncRecord) .where(LexwareSyncRecord.entity_type == "repair_estimate") .where(LexwareSyncRecord.entity_id == estimate_id) .where(LexwareSyncRecord.lexware_resource_type == "invoice") .order_by(LexwareSyncRecord.created_at.desc(), LexwareSyncRecord.id.desc()) .limit(1) ) @staticmethod def settings_response(config: LexwareRuntimeConfig) -> LexwareSettingsResponse: return LexwareSettingsResponse( enabled=config.enabled, api_base_url=config.api_base_url, api_key_is_set=config.api_key_is_set, organization_name=config.organization_name, default_tax_rate=config.default_tax_rate, default_payment_terms_days=config.default_payment_terms_days, source=config.source, ) @staticmethod def _get_profile(config: LexwareRuntimeConfig) -> dict: request = UrlRequest( f"{config.api_base_url}/v1/profile", headers={ "Authorization": f"Bearer {config.api_key}", "Accept": "application/json", }, method="GET", ) with urlopen(request, timeout=15) as response: body = response.read().decode("utf-8") if not body: return {} data = json.loads(body) return data if isinstance(data, dict) else {}