Add requests dependency to project
- Included 'requests' in the dependencies list of uv.lock. - Specified minimum version of 'requests' as 2.32.0 in requires-dist.
This commit is contained in:
bsiggel
266
services/adressen_mapper.py
Normal file
266
services/adressen_mapper.py
Normal file
@@ -0,0 +1,266 @@
|
||||
"""
|
||||
Adressen Mapper: EspoCRM CAdressen ↔ Advoware Adressen
|
||||
|
||||
Transformiert Adressen zwischen den beiden Systemen.
|
||||
Basierend auf ADRESSEN_SYNC_ANALYSE.md Abschnitt 12.
|
||||
"""
|
||||
|
||||
from typing import Dict, Any, Optional
|
||||
from datetime import datetime
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AdressenMapper:
|
||||
"""Mapper für CAdressen (EspoCRM) ↔ Adressen (Advoware)"""
|
||||
|
||||
@staticmethod
|
||||
def map_cadressen_to_advoware_create(espo_addr: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Transformiert EspoCRM CAdressen → Advoware Adressen Format (CREATE/POST)
|
||||
|
||||
Für CREATE werden ALLE 11 Felder gemappt (inkl. READ-ONLY bei PUT).
|
||||
|
||||
Args:
|
||||
espo_addr: CAdressen Entity von EspoCRM
|
||||
|
||||
Returns:
|
||||
Dict für Advoware POST /api/v1/advonet/Beteiligte/{betnr}/Adressen
|
||||
"""
|
||||
logger.debug(f"Mapping EspoCRM → Advoware (CREATE): {espo_addr.get('id')}")
|
||||
|
||||
# Formatiere Anschrift (mehrzeilig)
|
||||
anschrift = AdressenMapper._format_anschrift(espo_addr)
|
||||
|
||||
advo_data = {
|
||||
# R/W Felder (via PUT änderbar)
|
||||
'strasse': espo_addr.get('adresseStreet') or '',
|
||||
'plz': espo_addr.get('adressePostalCode') or '',
|
||||
'ort': espo_addr.get('adresseCity') or '',
|
||||
'anschrift': anschrift,
|
||||
|
||||
# READ-ONLY Felder (nur bei CREATE!)
|
||||
'land': espo_addr.get('adresseCountry') or 'DE',
|
||||
'postfach': espo_addr.get('postfach'),
|
||||
'postfachPLZ': espo_addr.get('postfachPLZ'),
|
||||
'standardAnschrift': bool(espo_addr.get('isPrimary', False)),
|
||||
'bemerkung': f"EspoCRM-ID: {espo_addr['id']}", # WICHTIG für Matching!
|
||||
'gueltigVon': AdressenMapper._format_datetime(espo_addr.get('validFrom')),
|
||||
'gueltigBis': AdressenMapper._format_datetime(espo_addr.get('validUntil'))
|
||||
}
|
||||
|
||||
return advo_data
|
||||
|
||||
@staticmethod
|
||||
def map_cadressen_to_advoware_update(espo_addr: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Transformiert EspoCRM CAdressen → Advoware Adressen Format (UPDATE/PUT)
|
||||
|
||||
Für UPDATE werden NUR die 4 R/W Felder gemappt!
|
||||
Alle anderen Änderungen müssen über Notifications gehandelt werden.
|
||||
|
||||
Args:
|
||||
espo_addr: CAdressen Entity von EspoCRM
|
||||
|
||||
Returns:
|
||||
Dict für Advoware PUT /api/v1/advonet/Beteiligte/{betnr}/Adressen/{index}
|
||||
"""
|
||||
logger.debug(f"Mapping EspoCRM → Advoware (UPDATE): {espo_addr.get('id')}")
|
||||
|
||||
# NUR R/W Felder!
|
||||
advo_data = {
|
||||
'strasse': espo_addr.get('adresseStreet') or '',
|
||||
'plz': espo_addr.get('adressePostalCode') or '',
|
||||
'ort': espo_addr.get('adresseCity') or '',
|
||||
'anschrift': AdressenMapper._format_anschrift(espo_addr)
|
||||
}
|
||||
|
||||
return advo_data
|
||||
|
||||
@staticmethod
|
||||
def map_advoware_to_cadressen(advo_addr: Dict[str, Any],
|
||||
beteiligte_id: str,
|
||||
existing_espo_addr: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Transformiert Advoware Adressen → EspoCRM CAdressen Format
|
||||
|
||||
Args:
|
||||
advo_addr: Adresse von Advoware GET
|
||||
beteiligte_id: EspoCRM CBeteiligte ID (für Relation)
|
||||
existing_espo_addr: Existierende EspoCRM Entity (für Update)
|
||||
|
||||
Returns:
|
||||
Dict für EspoCRM API
|
||||
"""
|
||||
logger.debug(f"Mapping Advoware → EspoCRM: Index {advo_addr.get('reihenfolgeIndex')}")
|
||||
|
||||
espo_data = {
|
||||
# Core Adressfelder
|
||||
'adresseStreet': advo_addr.get('strasse'),
|
||||
'adressePostalCode': advo_addr.get('plz'),
|
||||
'adresseCity': advo_addr.get('ort'),
|
||||
'adresseCountry': advo_addr.get('land') or 'DE',
|
||||
|
||||
# Zusatzfelder
|
||||
'postfach': advo_addr.get('postfach'),
|
||||
'postfachPLZ': advo_addr.get('postfachPLZ'),
|
||||
'description': advo_addr.get('bemerkung'),
|
||||
|
||||
# Status-Felder
|
||||
'isPrimary': bool(advo_addr.get('standardAnschrift', False)),
|
||||
'validFrom': advo_addr.get('gueltigVon'),
|
||||
'validUntil': advo_addr.get('gueltigBis'),
|
||||
|
||||
# Sync-Felder
|
||||
'advowareRowId': advo_addr.get('rowId'),
|
||||
'advowareLastSync': datetime.now().isoformat(),
|
||||
'syncStatus': 'synced',
|
||||
|
||||
# Relation
|
||||
'beteiligteId': beteiligte_id
|
||||
}
|
||||
|
||||
# Preserve existing fields when updating
|
||||
if existing_espo_addr:
|
||||
espo_data['id'] = existing_espo_addr['id']
|
||||
# Keep existing isActive if not changed
|
||||
if 'isActive' in existing_espo_addr:
|
||||
espo_data['isActive'] = existing_espo_addr['isActive']
|
||||
else:
|
||||
# New address
|
||||
espo_data['isActive'] = True
|
||||
|
||||
return espo_data
|
||||
|
||||
@staticmethod
|
||||
def detect_readonly_changes(espo_addr: Dict[str, Any],
|
||||
advo_addr: Dict[str, Any]) -> list[Dict[str, Any]]:
|
||||
"""
|
||||
Erkenne Änderungen an READ-ONLY Feldern (nicht via PUT änderbar)
|
||||
|
||||
Args:
|
||||
espo_addr: EspoCRM CAdressen Entity
|
||||
advo_addr: Advoware Adresse
|
||||
|
||||
Returns:
|
||||
Liste von Änderungen mit Feldnamen und Werten
|
||||
"""
|
||||
changes = []
|
||||
|
||||
# Mapping: EspoCRM-Feld → (Advoware-Feld, Label)
|
||||
readonly_mappings = {
|
||||
'adresseCountry': ('land', 'Land'),
|
||||
'postfach': ('postfach', 'Postfach'),
|
||||
'postfachPLZ': ('postfachPLZ', 'Postfach PLZ'),
|
||||
'isPrimary': ('standardAnschrift', 'Hauptadresse'),
|
||||
'validFrom': ('gueltigVon', 'Gültig von'),
|
||||
'validUntil': ('gueltigBis', 'Gültig bis')
|
||||
}
|
||||
|
||||
for espo_field, (advo_field, label) in readonly_mappings.items():
|
||||
espo_value = espo_addr.get(espo_field)
|
||||
advo_value = advo_addr.get(advo_field)
|
||||
|
||||
# Normalisiere Werte für Vergleich
|
||||
if espo_field == 'isPrimary':
|
||||
espo_value = bool(espo_value)
|
||||
advo_value = bool(advo_value)
|
||||
elif espo_field in ['validFrom', 'validUntil']:
|
||||
# Datetime-Vergleich (nur Datum)
|
||||
espo_value = AdressenMapper._normalize_date(espo_value)
|
||||
advo_value = AdressenMapper._normalize_date(advo_value)
|
||||
|
||||
# Vergleiche
|
||||
if espo_value != advo_value:
|
||||
changes.append({
|
||||
'field': label,
|
||||
'espoField': espo_field,
|
||||
'advoField': advo_field,
|
||||
'espoCRM_value': espo_value,
|
||||
'advoware_value': advo_value
|
||||
})
|
||||
|
||||
return changes
|
||||
|
||||
@staticmethod
|
||||
def _format_anschrift(espo_addr: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Formatiert mehrzeilige Anschrift für Advoware
|
||||
|
||||
Format:
|
||||
{Firmenname oder Name}
|
||||
{Strasse}
|
||||
{PLZ} {Ort}
|
||||
"""
|
||||
parts = []
|
||||
|
||||
# Zeile 1: Name
|
||||
if espo_addr.get('firmenname'):
|
||||
parts.append(espo_addr['firmenname'])
|
||||
elif espo_addr.get('firstName') or espo_addr.get('lastName'):
|
||||
name = f"{espo_addr.get('firstName', '')} {espo_addr.get('lastName', '')}".strip()
|
||||
if name:
|
||||
parts.append(name)
|
||||
|
||||
# Zeile 2: Straße
|
||||
if espo_addr.get('adresseStreet'):
|
||||
parts.append(espo_addr['adresseStreet'])
|
||||
|
||||
# Zeile 3: PLZ + Ort
|
||||
plz = espo_addr.get('adressePostalCode', '').strip()
|
||||
ort = espo_addr.get('adresseCity', '').strip()
|
||||
if plz or ort:
|
||||
parts.append(f"{plz} {ort}".strip())
|
||||
|
||||
return '\n'.join(parts)
|
||||
|
||||
@staticmethod
|
||||
def _format_datetime(dt: Any) -> Optional[str]:
|
||||
"""
|
||||
Formatiert Datetime für Advoware API (ISO 8601)
|
||||
|
||||
Args:
|
||||
dt: datetime object, ISO string, oder None
|
||||
|
||||
Returns:
|
||||
ISO 8601 string oder None
|
||||
"""
|
||||
if not dt:
|
||||
return None
|
||||
|
||||
if isinstance(dt, str):
|
||||
# Bereits String - prüfe ob gültig
|
||||
try:
|
||||
datetime.fromisoformat(dt.replace('Z', '+00:00'))
|
||||
return dt
|
||||
except:
|
||||
return None
|
||||
|
||||
if isinstance(dt, datetime):
|
||||
return dt.isoformat()
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _normalize_date(dt: Any) -> Optional[str]:
|
||||
"""
|
||||
Normalisiert Datum für Vergleich (nur Datum, keine Zeit)
|
||||
|
||||
Returns:
|
||||
YYYY-MM-DD string oder None
|
||||
"""
|
||||
if not dt:
|
||||
return None
|
||||
|
||||
if isinstance(dt, str):
|
||||
try:
|
||||
dt_obj = datetime.fromisoformat(dt.replace('Z', '+00:00'))
|
||||
return dt_obj.strftime('%Y-%m-%d')
|
||||
except:
|
||||
return None
|
||||
|
||||
if isinstance(dt, datetime):
|
||||
return dt.strftime('%Y-%m-%d')
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user