361 lines
14 KiB
Python
361 lines
14 KiB
Python
import json
|
|
from dataclasses import dataclass
|
|
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 (
|
|
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 = {
|
|
"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),
|
|
}
|
|
record = LexwareSyncRecord(
|
|
entity_type="repair_estimate",
|
|
entity_id=estimate.id,
|
|
lexware_resource_type="invoice",
|
|
status="pending" if not warnings else "skipped",
|
|
direction="push",
|
|
payload_summary=json.dumps(payload_summary, ensure_ascii=True),
|
|
error_message="; ".join(warnings) if warnings else None,
|
|
)
|
|
db.add(record)
|
|
db.commit()
|
|
db.refresh(record)
|
|
write_audit_log(
|
|
db,
|
|
action="lexware.invoice.prepare",
|
|
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,
|
|
},
|
|
)
|
|
return LexwareInvoicePreparationResponse(
|
|
ready_for_export=not warnings,
|
|
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,
|
|
)
|
|
|
|
@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 {}
|