Files
motia-iii/services/kommunikation_sync_utils.py

997 lines
46 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Kommunikation Sync Utilities
Bidirektionale Synchronisation: Advoware ↔ EspoCRM
Strategie:
- Emails: emailAddressData[] ↔ Advoware Kommunikationen (kommKz: 4,8,11,12)
- Phones: phoneNumberData[] ↔ Advoware Kommunikationen (kommKz: 1,2,3,6,7,9,10)
- Matching: Hash-basiert via bemerkung-Marker
- Type Detection: Marker > Top-Level > Value Pattern > Default
"""
import logging
from typing import Dict, List, Optional, Tuple, Any
from services.kommunikation_mapper import (
parse_marker, create_marker, create_slot_marker,
detect_kommkz, encode_value, decode_value,
is_email_type, is_phone_type,
advoware_to_espocrm_email, advoware_to_espocrm_phone,
find_matching_advoware, find_empty_slot,
should_sync_to_espocrm, get_user_bemerkung,
calculate_hash,
EMAIL_KOMMKZ, PHONE_KOMMKZ
)
from services.advoware_service import AdvowareService
from services.espocrm import EspoCRMAPI
class KommunikationSyncManager:
"""Manager für Kommunikation-Synchronisation"""
def __init__(self, advoware: AdvowareService, espocrm: EspoCRMAPI, context=None):
self.advoware = advoware
self.espocrm = espocrm
self.context = context
self.logger = context.logger if context else logger
# ========== BIDIRECTIONAL SYNC ==========
async def sync_bidirectional(self, beteiligte_id: str, betnr: int,
direction: str = 'both', force_espo_wins: bool = False) -> Dict[str, Any]:
"""
Bidirektionale Synchronisation mit intelligentem Diffing
Optimiert:
- Lädt Daten nur 1x von jeder Seite (kein doppelter API-Call)
- Echtes 3-Way Diffing (Advoware, EspoCRM, Marker)
- Handhabt alle 6 Szenarien korrekt (Var1-6)
- Initial Sync: Value-Matching verhindert Duplikate (BUG-3 Fix)
- Hash nur bei Änderung schreiben (Performance)
- Lock-Release garantiert via try/finally
Args:
direction: 'both', 'to_espocrm', 'to_advoware'
force_espo_wins: Erzwingt EspoCRM-wins Konfliktlösung (für Stammdaten-Konflikte)
Returns:
Combined results mit detaillierten Änderungen
"""
result = {
'advoware_to_espocrm': {'emails_synced': 0, 'phones_synced': 0, 'errors': []},
'espocrm_to_advoware': {'created': 0, 'updated': 0, 'deleted': 0, 'errors': []},
'summary': {'total_changes': 0}
}
# NOTE: Lock-Management erfolgt außerhalb dieser Methode (in Event/Cron Handler)
# Diese Methode ist für die reine Sync-Logik zuständig
try:
# ========== LADE DATEN NUR 1X ==========
self.logger.info(f"[KOMM] Bidirectional Sync: betnr={betnr}, bet_id={beteiligte_id}")
# Advoware Daten
advo_result = await self.advoware.get_beteiligter(betnr)
if isinstance(advo_result, list):
advo_bet = advo_result[0] if advo_result else None
else:
advo_bet = advo_result
if not advo_bet:
result['advoware_to_espocrm']['errors'].append("Advoware Beteiligte nicht gefunden")
result['espocrm_to_advoware']['errors'].append("Advoware Beteiligte nicht gefunden")
return result
# EspoCRM Daten
espo_bet = await self.espocrm.get_entity('CBeteiligte', beteiligte_id)
if not espo_bet:
result['advoware_to_espocrm']['errors'].append("EspoCRM Beteiligte nicht gefunden")
result['espocrm_to_advoware']['errors'].append("EspoCRM Beteiligte nicht gefunden")
return result
advo_kommunikationen = advo_bet.get('kommunikation', [])
espo_emails = espo_bet.get('emailAddressData', [])
espo_phones = espo_bet.get('phoneNumberData', [])
self.logger.info(f"[KOMM] Geladen: {len(advo_kommunikationen)} Advoware, {len(espo_emails)} EspoCRM emails, {len(espo_phones)} EspoCRM phones")
# Check ob initialer Sync
stored_komm_hash = espo_bet.get('kommunikationHash')
is_initial_sync = not stored_komm_hash
# ========== 3-WAY DIFFING MIT HASH-BASIERTER KONFLIKT-ERKENNUNG ==========
diff = self._compute_diff(advo_kommunikationen, espo_emails, espo_phones, advo_bet, espo_bet)
# WICHTIG: force_espo_wins überschreibt den Hash-basierten Konflikt-Check
if force_espo_wins:
diff['espo_wins'] = True
self.logger.info(f"[KOMM] ⚠️ force_espo_wins=True → EspoCRM WINS (override)")
# Konvertiere Var3 (advo_deleted) → Var1 (espo_new)
# Bei Konflikt müssen gelöschte Advoware-Einträge wiederhergestellt werden
if diff['advo_deleted']:
self.logger.info(f"[KOMM] 🔄 Converting {len(diff['advo_deleted'])} Var3→Var1 (force EspoCRM wins)")
for value, espo_item in diff['advo_deleted']:
diff['espo_new'].append((value, espo_item))
diff['advo_deleted'] = [] # Leeren, da jetzt als Var1 behandelt
espo_wins = diff.get('espo_wins', False)
self.logger.info(f"[KOMM] ===== DIFF RESULTS =====")
self.logger.info(f"[KOMM] Diff: {len(diff['advo_changed'])} Advoware changed, {len(diff['espo_changed'])} EspoCRM changed, "
f"{len(diff['advo_new'])} Advoware new, {len(diff['espo_new'])} EspoCRM new, "
f"{len(diff['advo_deleted'])} Advoware deleted, {len(diff['espo_deleted'])} EspoCRM deleted")
force_status = " (force=True)" if force_espo_wins else ""
self.logger.info(f"[KOMM] ===== CONFLICT STATUS: espo_wins={espo_wins}{force_status} =====")
# ========== APPLY CHANGES ==========
# Bestimme Sync-Richtungen und Konflikt-Handling
sync_to_espocrm = direction in ['both', 'to_espocrm']
sync_to_advoware = direction in ['both', 'to_advoware']
should_revert_advoware_changes = (sync_to_espocrm and espo_wins) or (direction == 'to_advoware')
# 1. Advoware → EspoCRM (Var4: Neu in Advoware, Var6: Geändert in Advoware)
if sync_to_espocrm and not espo_wins:
self.logger.info(f"[KOMM] ✅ Applying Advoware→EspoCRM changes...")
espo_result = await self._apply_advoware_to_espocrm(
beteiligte_id, diff, advo_bet
)
result['advoware_to_espocrm'] = espo_result
# Bei Konflikt oder direction='to_advoware': Revert Advoware-Änderungen
if should_revert_advoware_changes:
if espo_wins:
self.logger.info(f"[KOMM] ⚠️ CONFLICT: EspoCRM wins - reverting Advoware changes")
else:
self.logger.info(f"[KOMM] Direction={direction}: reverting Advoware changes")
# Var6: Revert Änderungen
if len(diff['advo_changed']) > 0:
self.logger.info(f"[KOMM] 🔄 Reverting {len(diff['advo_changed'])} Var6 entries to EspoCRM values...")
for komm, old_value, new_value in diff['advo_changed']:
await self._revert_advoware_change(betnr, komm, old_value, new_value, advo_bet)
result['espocrm_to_advoware']['updated'] += 1
# Var4: Convert to Empty Slots
if len(diff['advo_new']) > 0:
self.logger.info(f"[KOMM] 🔄 Converting {len(diff['advo_new'])} Var4 entries to Empty Slots...")
for komm in diff['advo_new']:
await self._create_empty_slot(betnr, komm)
result['espocrm_to_advoware']['deleted'] += 1
# Var3: Wiederherstellung gelöschter Einträge (kein separater Code nötig)
# → Wird über Var1 in _apply_espocrm_to_advoware behandelt
# Die gelöschten Einträge sind noch in EspoCRM vorhanden und werden als "espo_new" erkannt
if len(diff['advo_deleted']) > 0:
self.logger.info(f"[KOMM] {len(diff['advo_deleted'])} Var3 entries (deleted in Advoware) will be restored via espo_new")
# 2. EspoCRM → Advoware (Var1: Neu in EspoCRM, Var2: Gelöscht in EspoCRM, Var5: Geändert in EspoCRM)
if sync_to_advoware:
advo_result = await self._apply_espocrm_to_advoware(
betnr, diff, advo_bet
)
# Merge results (Var6/Var4 Counts aus Konflikt-Handling behalten)
result['espocrm_to_advoware']['created'] += advo_result['created']
result['espocrm_to_advoware']['updated'] += advo_result['updated']
result['espocrm_to_advoware']['deleted'] += advo_result['deleted']
result['espocrm_to_advoware']['errors'].extend(advo_result['errors'])
# 3. Initial Sync Matches: Nur Marker setzen (keine CREATE/UPDATE)
if is_initial_sync and 'initial_sync_matches' in diff:
self.logger.info(f"[KOMM] ✓ Processing {len(diff['initial_sync_matches'])} initial sync matches...")
for value, matched_komm, espo_item in diff['initial_sync_matches']:
# Erkenne kommKz
espo_type = espo_item.get('type', 'email' if '@' in value else None)
kommkz = detect_kommkz(value, advo_bet, espo_type=espo_type)
# Setze Marker in Advoware
await self.advoware.update_kommunikation(betnr, matched_komm['id'], {
'bemerkung': create_marker(value, kommkz),
'online': espo_item.get('primary', False)
})
result['espocrm_to_advoware']['updated'] += 1
total_changes = (
result['advoware_to_espocrm']['emails_synced'] +
result['advoware_to_espocrm']['phones_synced'] +
result['espocrm_to_advoware']['created'] +
result['espocrm_to_advoware']['updated'] +
result['espocrm_to_advoware']['deleted']
)
result['summary']['total_changes'] = total_changes
# Hash-Update: Immer berechnen, aber nur schreiben wenn geändert
import hashlib
# FIX: Nur neu laden wenn Änderungen gemacht wurden
if total_changes > 0:
advo_result_final = await self.advoware.get_beteiligter(betnr)
if isinstance(advo_result_final, list):
advo_bet_final = advo_result_final[0]
else:
advo_bet_final = advo_result_final
final_kommunikationen = advo_bet_final.get('kommunikation', [])
else:
# Keine Änderungen: Verwende cached data (keine doppelte API-Call)
final_kommunikationen = advo_bet.get('kommunikation', [])
# Berechne neuen Hash
sync_relevant_komm = [
k for k in final_kommunikationen
if should_sync_to_espocrm(k)
]
komm_rowids = sorted([k.get('rowId', '') for k in sync_relevant_komm if k.get('rowId')])
new_komm_hash = hashlib.md5(''.join(komm_rowids).encode()).hexdigest()[:16]
# Nur schreiben wenn Hash sich geändert hat oder Initial Sync
if new_komm_hash != stored_komm_hash:
await self.espocrm.update_entity('CBeteiligte', beteiligte_id, {
'kommunikationHash': new_komm_hash
})
self.logger.info(f"[KOMM] ✅ Updated kommunikationHash: {stored_komm_hash}{new_komm_hash} (based on {len(sync_relevant_komm)} sync-relevant of {len(final_kommunikationen)} total)")
else:
self.logger.info(f"[KOMM] Hash unchanged: {new_komm_hash} - no EspoCRM update needed")
self.logger.info(f"[KOMM] ✅ Bidirectional Sync complete: {total_changes} total changes")
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler bei Bidirectional Sync: {e}")
self.logger.error(traceback.format_exc())
result['advoware_to_espocrm']['errors'].append(str(e))
result['espocrm_to_advoware']['errors'].append(str(e))
return result
# ========== 3-WAY DIFFING ==========
def _compute_diff(self, advo_kommunikationen: List[Dict], espo_emails: List[Dict],
espo_phones: List[Dict], advo_bet: Dict, espo_bet: Dict) -> Dict[str, List]:
"""
Berechnet Diff zwischen Advoware und EspoCRM mit Hash-basierter Konflikt-Erkennung
Returns:
Dict mit Var1-6 Änderungen und Konflikt-Status
"""
diff = {
'advo_changed': [], # Var6
'advo_new': [], # Var4
'advo_deleted': [], # Var3
'espo_changed': [], # Var5
'espo_new': [], # Var1
'espo_deleted': [], # Var2
'no_change': [],
'espo_wins': False
}
# 1. Konflikt-Erkennung
is_initial_sync, espo_wins, espo_changed_since_sync, advo_changed_since_sync = \
self._detect_conflict(advo_kommunikationen, espo_bet)
diff['espo_wins'] = espo_wins
# 2. Baue Value-Maps
espo_values = self._build_espocrm_value_map(espo_emails, espo_phones)
advo_with_marker, advo_without_marker = self._build_advoware_maps(advo_kommunikationen)
# 3. Analysiere Advoware-Einträge MIT Marker
self._analyze_advoware_with_marker(advo_with_marker, espo_values, diff)
# 4. Analysiere Advoware-Einträge OHNE Marker (Var4) + Initial Sync Matching
self._analyze_advoware_without_marker(
advo_without_marker, espo_values, is_initial_sync, advo_bet, diff
)
# 5. Analysiere EspoCRM-Einträge die nicht in Advoware sind (Var1/Var3)
self._analyze_espocrm_only(
espo_values, advo_with_marker, espo_wins,
espo_changed_since_sync, advo_changed_since_sync, diff
)
return diff
def _detect_conflict(self, advo_kommunikationen: List[Dict], espo_bet: Dict) -> Tuple[bool, bool, bool, bool]:
"""
Erkennt Konflikte via Hash-Vergleich
Returns:
(is_initial_sync, espo_wins, espo_changed_since_sync, advo_changed_since_sync)
"""
espo_modified = espo_bet.get('modifiedAt')
last_sync = espo_bet.get('advowareLastSync')
stored_komm_hash = espo_bet.get('kommunikationHash')
# Berechne aktuellen Hash
import hashlib
sync_relevant_komm = [k for k in advo_kommunikationen if should_sync_to_espocrm(k)]
komm_rowids = sorted([k.get('rowId', '') for k in sync_relevant_komm if k.get('rowId')])
current_advo_hash = hashlib.md5(''.join(komm_rowids).encode()).hexdigest()[:16]
# Parse Timestamps
from services.beteiligte_sync_utils import BeteiligteSync
espo_modified_ts = BeteiligteSync.parse_timestamp(espo_modified)
last_sync_ts = BeteiligteSync.parse_timestamp(last_sync)
# Bestimme Änderungen
espo_changed_since_sync = espo_modified_ts and last_sync_ts and espo_modified_ts > last_sync_ts
advo_changed_since_sync = stored_komm_hash and current_advo_hash != stored_komm_hash
is_initial_sync = not stored_komm_hash
# Konflikt-Logik: Beide geändert → EspoCRM wins
espo_wins = espo_changed_since_sync and advo_changed_since_sync
self.logger.info(f"[KOMM] 🔍 Konflikt-Check:")
self.logger.info(f"[KOMM] - EspoCRM changed: {espo_changed_since_sync}, Advoware changed: {advo_changed_since_sync}")
self.logger.info(f"[KOMM] - Initial sync: {is_initial_sync}, Conflict: {espo_wins}")
self.logger.info(f"[KOMM] - Hash: stored={stored_komm_hash}, current={current_advo_hash}")
return is_initial_sync, espo_wins, espo_changed_since_sync, advo_changed_since_sync
def _build_espocrm_value_map(self, espo_emails: List[Dict], espo_phones: List[Dict]) -> Dict[str, Dict]:
"""Baut Map: value → {value, is_email, primary, type}"""
espo_values = {}
for email in espo_emails:
val = email.get('emailAddress', '').strip()
if val:
espo_values[val] = {
'value': val,
'is_email': True,
'primary': email.get('primary', False),
'type': 'email'
}
for phone in espo_phones:
val = phone.get('phoneNumber', '').strip()
if val:
espo_values[val] = {
'value': val,
'is_email': False,
'primary': phone.get('primary', False),
'type': phone.get('type', 'Office')
}
return espo_values
def _build_advoware_maps(self, advo_kommunikationen: List[Dict]) -> Tuple[Dict, List]:
"""
Trennt Advoware-Einträge in MIT Marker und OHNE Marker
Returns:
(advo_with_marker: {synced_value: (komm, current_value)}, advo_without_marker: [komm])
"""
advo_with_marker = {}
advo_without_marker = []
for komm in advo_kommunikationen:
if not should_sync_to_espocrm(komm):
continue
tlf = (komm.get('tlf') or '').strip()
if not tlf:
continue
marker = parse_marker(komm.get('bemerkung', ''))
if marker and not marker['is_slot']:
# Hat Marker → Von EspoCRM synchronisiert
advo_with_marker[marker['synced_value']] = (komm, tlf)
else:
# Kein Marker → Von Advoware angelegt (Var4)
advo_without_marker.append(komm)
return advo_with_marker, advo_without_marker
def _analyze_advoware_with_marker(self, advo_with_marker: Dict, espo_values: Dict, diff: Dict) -> None:
"""Analysiert Advoware-Einträge MIT Marker für Var6, Var5, Var2"""
for synced_value, (komm, current_value) in advo_with_marker.items():
if synced_value != current_value:
# Var6: In Advoware geändert
self.logger.info(f"[KOMM] ✏️ Var6: Changed in Advoware")
diff['advo_changed'].append((komm, synced_value, current_value))
elif synced_value in espo_values:
espo_item = espo_values[synced_value]
current_online = komm.get('online', False)
espo_primary = espo_item['primary']
if current_online != espo_primary:
# Var5: EspoCRM hat primary geändert
self.logger.info(f"[KOMM] 🔄 Var5: Primary changed in EspoCRM")
diff['espo_changed'].append((synced_value, komm, espo_item))
else:
# Keine Änderung
diff['no_change'].append((synced_value, komm, espo_item))
else:
# Var2: In EspoCRM gelöscht
self.logger.info(f"[KOMM] 🗑️ Var2: Deleted in EspoCRM")
diff['espo_deleted'].append(komm)
def _analyze_advoware_without_marker(
self, advo_without_marker: List[Dict], espo_values: Dict,
is_initial_sync: bool, advo_bet: Dict, diff: Dict
) -> None:
"""Analysiert Advoware-Einträge OHNE Marker für Var4 + Initial Sync Matching"""
# FIX BUG-3: Bei Initial Sync Value-Map erstellen
advo_values_without_marker = {}
if is_initial_sync:
advo_values_without_marker = {
(k.get('tlf') or '').strip(): k
for k in advo_without_marker
if (k.get('tlf') or '').strip()
}
# Sammle matched values für Initial Sync
matched_komm_ids = set()
# Prüfe ob EspoCRM-Werte bereits in Advoware existieren (Initial Sync)
if is_initial_sync:
for value in espo_values.keys():
if value in advo_values_without_marker:
matched_komm = advo_values_without_marker[value]
espo_item = espo_values[value]
# Match gefunden - setze nur Marker, kein Var1/Var4
if 'initial_sync_matches' not in diff:
diff['initial_sync_matches'] = []
diff['initial_sync_matches'].append((value, matched_komm, espo_item))
matched_komm_ids.add(matched_komm['id'])
self.logger.info(f"[KOMM] ✓ Initial Sync Match: '{value[:30]}...'")
# Var4: Neu in Advoware (nicht matched im Initial Sync)
for komm in advo_without_marker:
if komm['id'] not in matched_komm_ids:
tlf = (komm.get('tlf') or '').strip()
self.logger.info(f"[KOMM] Var4: New in Advoware - '{tlf[:30]}...'")
diff['advo_new'].append(komm)
def _analyze_espocrm_only(
self, espo_values: Dict, advo_with_marker: Dict,
espo_wins: bool, espo_changed_since_sync: bool,
advo_changed_since_sync: bool, diff: Dict
) -> None:
"""Analysiert EspoCRM-Einträge die nicht in Advoware sind für Var1/Var3"""
# Sammle bereits gematchte values aus Initial Sync
matched_values = set()
if 'initial_sync_matches' in diff:
matched_values = {v for v, k, e in diff['initial_sync_matches']}
for value, espo_item in espo_values.items():
# Skip wenn bereits im Initial Sync gematched
if value in matched_values:
continue
# Skip wenn in Advoware mit Marker
if value in advo_with_marker:
continue
# Hash-basierte Logik: Var1 vs Var3
if espo_wins or (espo_changed_since_sync and not advo_changed_since_sync):
# Var1: Neu in EspoCRM
self.logger.info(f"[KOMM] Var1: New in EspoCRM '{value[:30]}...'")
diff['espo_new'].append((value, espo_item))
elif advo_changed_since_sync and not espo_changed_since_sync:
# Var3: In Advoware gelöscht
self.logger.info(f"[KOMM] 🗑️ Var3: Deleted in Advoware '{value[:30]}...'")
diff['advo_deleted'].append((value, espo_item))
else:
# Default: Var1 (neu in EspoCRM)
self.logger.info(f"[KOMM] Var1 (default): '{value[:30]}...'")
diff['espo_new'].append((value, espo_item))
# ========== APPLY CHANGES ==========
async def _apply_advoware_to_espocrm(self, beteiligte_id: str, diff: Dict,
advo_bet: Dict) -> Dict[str, Any]:
"""
Wendet Advoware-Änderungen auf EspoCRM an (Var4, Var6)
"""
result = {'emails_synced': 0, 'phones_synced': 0, 'markers_updated': 0, 'errors': []}
try:
# Lade aktuelle EspoCRM Daten
espo_bet = await self.espocrm.get_entity('CBeteiligte', beteiligte_id)
espo_emails = list(espo_bet.get('emailAddressData', []))
espo_phones = list(espo_bet.get('phoneNumberData', []))
# Var6: Advoware-Änderungen → Update Marker + Sync zu EspoCRM
for komm, old_value, new_value in diff['advo_changed']:
self.logger.info(f"[KOMM] Var6: Advoware changed '{old_value}''{new_value}'")
# Update Marker in Advoware
bemerkung = komm.get('bemerkung') or ''
marker = parse_marker(bemerkung)
user_text = marker.get('user_text', '') if marker else ''
kommkz = marker['kommKz'] if marker else detect_kommkz(new_value, advo_bet)
new_marker = create_marker(new_value, kommkz, user_text)
await self.advoware.update_kommunikation(advo_bet['betNr'], komm['id'], {
'bemerkung': new_marker
})
result['markers_updated'] += 1
# Update in EspoCRM: Finde alten Wert und ersetze mit neuem
if is_email_type(kommkz):
for i, email in enumerate(espo_emails):
if email.get('emailAddress') == old_value:
espo_emails[i] = {
'emailAddress': new_value,
'lower': new_value.lower(),
'primary': komm.get('online', False),
'optOut': False,
'invalid': False
}
result['emails_synced'] += 1
break
else:
for i, phone in enumerate(espo_phones):
if phone.get('phoneNumber') == old_value:
type_map = {1: 'Office', 2: 'Fax', 3: 'Mobile', 6: 'Home', 7: 'Fax', 9: 'Mobile', 10: 'Other'}
espo_phones[i] = {
'phoneNumber': new_value,
'type': type_map.get(kommkz, 'Other'),
'primary': komm.get('online', False),
'optOut': False,
'invalid': False
}
result['phones_synced'] += 1
break
# Var4: Neu in Advoware → Zu EspoCRM hinzufügen + Marker setzen
for komm in diff['advo_new']:
tlf = (komm.get('tlf') or '').strip()
kommkz = detect_kommkz(tlf, advo_bet, komm.get('bemerkung'))
self.logger.info(f"[KOMM] Var4: New in Advoware '{tlf}', syncing to EspoCRM")
# Setze Marker in Advoware
new_marker = create_marker(tlf, kommkz)
await self.advoware.update_kommunikation(advo_bet['betNr'], komm['id'], {
'bemerkung': new_marker
})
# Zu EspoCRM hinzufügen
if is_email_type(kommkz):
espo_emails.append({
'emailAddress': tlf,
'lower': tlf.lower(),
'primary': komm.get('online', False),
'optOut': False,
'invalid': False
})
result['emails_synced'] += 1
else:
type_map = {1: 'Office', 2: 'Fax', 3: 'Mobile', 6: 'Home', 7: 'Fax', 9: 'Mobile', 10: 'Other'}
espo_phones.append({
'phoneNumber': tlf,
'type': type_map.get(kommkz, 'Other'),
'primary': komm.get('online', False),
'optOut': False,
'invalid': False
})
result['phones_synced'] += 1
# Var3: In Advoware gelöscht → Aus EspoCRM entfernen
for value, espo_item in diff.get('advo_deleted', []):
self.logger.info(f"[KOMM] Var3: Deleted in Advoware '{value}', removing from EspoCRM")
if espo_item['is_email']:
espo_emails = [e for e in espo_emails if e.get('emailAddress') != value]
result['emails_synced'] += 1 # Zählt als "synced" (gelöscht)
else:
espo_phones = [p for p in espo_phones if p.get('phoneNumber') != value]
result['phones_synced'] += 1
# Update EspoCRM wenn Änderungen
if result['emails_synced'] > 0 or result['phones_synced'] > 0:
await self.espocrm.update_entity('CBeteiligte', beteiligte_id, {
'emailAddressData': espo_emails,
'phoneNumberData': espo_phones
})
self.logger.info(f"[KOMM] ✅ Updated EspoCRM: {result['emails_synced']} emails, {result['phones_synced']} phones")
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler bei Advoware→EspoCRM Apply: {e}")
self.logger.error(traceback.format_exc())
result['errors'].append(str(e))
return result
async def _apply_espocrm_to_advoware(self, betnr: int, diff: Dict,
advo_bet: Dict) -> Dict[str, Any]:
"""
Wendet EspoCRM-Änderungen auf Advoware an (Var1, Var2, Var3, Var5)
"""
result = {'created': 0, 'updated': 0, 'deleted': 0, 'errors': []}
try:
advo_kommunikationen = advo_bet.get('kommunikation', [])
# OPTIMIERUNG: Matche Var2 (Delete) + Var1 (New) mit gleichem kommKz
# → Direkt UPDATE statt DELETE+RELOAD+CREATE
var2_by_kommkz = {} # kommKz → [komm, ...]
var1_by_kommkz = {} # kommKz → [(value, espo_item), ...]
# Gruppiere Var2 nach kommKz
for komm in diff['espo_deleted']:
bemerkung = komm.get('bemerkung') or ''
marker = parse_marker(bemerkung)
if marker:
kommkz = marker['kommKz']
if kommkz not in var2_by_kommkz:
var2_by_kommkz[kommkz] = []
var2_by_kommkz[kommkz].append(komm)
# Gruppiere Var1 nach kommKz
for value, espo_item in diff['espo_new']:
espo_type = espo_item.get('type', 'email' if '@' in value else None)
kommkz = detect_kommkz(value, advo_bet, espo_type=espo_type)
if kommkz not in var1_by_kommkz:
var1_by_kommkz[kommkz] = []
var1_by_kommkz[kommkz].append((value, espo_item))
# Matche und führe direkte Updates aus
matched_var2_ids = set()
matched_var1_indices = {} # kommkz → set of matched indices
for kommkz in var2_by_kommkz.keys():
if kommkz in var1_by_kommkz:
var2_list = var2_by_kommkz[kommkz]
var1_list = var1_by_kommkz[kommkz]
# Matche paarweise
for i, (value, espo_item) in enumerate(var1_list):
if i < len(var2_list):
komm = var2_list[i]
komm_id = komm['id']
self.logger.info(f"[KOMM] 🔄 Var2+Var1 Match: kommKz={kommkz}, updating slot {komm_id} with '{value[:30]}...'")
# Direktes UPDATE statt DELETE+CREATE
await self.advoware.update_kommunikation(betnr, komm_id, {
'tlf': value,
'online': espo_item['primary'],
'bemerkung': create_marker(value, kommkz)
})
matched_var2_ids.add(komm_id)
if kommkz not in matched_var1_indices:
matched_var1_indices[kommkz] = set()
matched_var1_indices[kommkz].add(i)
result['created'] += 1
self.logger.info(f"[KOMM] ✅ Slot updated (optimized merge)")
# Unmatched Var2: Erstelle Empty Slots
for komm in diff['espo_deleted']:
komm_id = komm.get('id')
if komm_id not in matched_var2_ids:
synced_value = komm.get('_synced_value', '')
self.logger.info(f"[KOMM] 🗑️ Var2: Deleted in EspoCRM - komm_id={komm_id}, synced_value='{synced_value[:30]}...'")
await self._create_empty_slot(betnr, komm, synced_value=synced_value)
result['deleted'] += 1
# Var5: In EspoCRM geändert (z.B. primary Flag)
for value, advo_komm, espo_item in diff['espo_changed']:
self.logger.info(f"[KOMM] ✏️ Var5: EspoCRM changed '{value[:30]}...', primary={espo_item.get('primary')}")
bemerkung = advo_komm.get('bemerkung') or ''
marker = parse_marker(bemerkung)
user_text = marker.get('user_text', '') if marker else ''
# Erkenne kommKz mit espo_type
if marker:
kommkz = marker['kommKz']
self.logger.info(f"[KOMM] kommKz from marker: {kommkz}")
else:
espo_type = espo_item.get('type', 'email' if '@' in value else None)
kommkz = detect_kommkz(value, advo_bet, espo_type=espo_type)
self.logger.info(f"[KOMM] kommKz detected: espo_type={espo_type}, kommKz={kommkz}")
# Update in Advoware
await self.advoware.update_kommunikation(betnr, advo_komm['id'], {
'tlf': value,
'online': espo_item['primary'],
'bemerkung': create_marker(value, kommkz, user_text)
})
self.logger.info(f"[KOMM] ✅ Updated komm_id={advo_komm['id']}, kommKz={kommkz}")
result['updated'] += 1
# Var1: Neu in EspoCRM → Create oder reuse Slot in Advoware
# Überspringe bereits gematchte Einträge (Var2+Var1 merged)
for idx, (value, espo_item) in enumerate(diff['espo_new']):
espo_type = espo_item.get('type', 'email' if '@' in value else None)
kommkz = detect_kommkz(value, advo_bet, espo_type=espo_type)
# Skip wenn bereits als Var2+Var1 Match verarbeitet
if kommkz in matched_var1_indices and idx in matched_var1_indices[kommkz]:
continue
self.logger.info(f"[KOMM] Var1: New in EspoCRM '{value[:30]}...', type={espo_item.get('type')}")
self.logger.info(f"[KOMM] 🔍 kommKz detected: espo_type={espo_type}, kommKz={kommkz}")
# Suche leeren Slot
empty_slot = find_empty_slot(kommkz, advo_kommunikationen)
if empty_slot:
# Reuse Slot
self.logger.info(f"[KOMM] ♻️ Reusing empty slot: slot_id={empty_slot['id']}, kommKz={kommkz}")
await self.advoware.update_kommunikation(betnr, empty_slot['id'], {
'tlf': value,
'online': espo_item['primary'],
'bemerkung': create_marker(value, kommkz)
})
self.logger.info(f"[KOMM] ✅ Slot reused successfully")
else:
# Create new
self.logger.info(f"[KOMM] Creating new kommunikation: kommKz={kommkz}")
await self.advoware.create_kommunikation(betnr, {
'tlf': value,
'kommKz': kommkz,
'online': espo_item['primary'],
'bemerkung': create_marker(value, kommkz)
})
self.logger.info(f"[KOMM] ✅ Created new kommunikation with kommKz={kommkz}")
result['created'] += 1
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler bei EspoCRM→Advoware Apply: {e}")
self.logger.error(traceback.format_exc())
result['errors'].append(str(e))
return result
# ========== HELPER METHODS ==========
async def _create_empty_slot(self, betnr: int, advo_komm: Dict, synced_value: str = None) -> None:
"""
Erstellt leeren Slot für gelöschten Eintrag
Args:
betnr: Beteiligten-Nummer
advo_komm: Kommunikations-Eintrag aus Advoware
synced_value: Optional - Original-Wert aus EspoCRM (nur für Logging)
Verwendet für:
- Var2: In EspoCRM gelöscht (hat Marker)
- Var4 bei Konflikt: Neu in Advoware aber EspoCRM wins (hat KEINEN Marker)
"""
try:
komm_id = advo_komm['id']
tlf = (advo_komm.get('tlf') or '').strip()
bemerkung = advo_komm.get('bemerkung') or ''
marker = parse_marker(bemerkung)
# Bestimme kommKz
if marker:
# Hat Marker (Var2)
kommkz = marker['kommKz']
else:
# Kein Marker (Var4 bei Konflikt) - erkenne kommKz aus Wert
from services.kommunikation_mapper import detect_kommkz
kommkz = detect_kommkz(tlf) if tlf else 1 # Default: TelGesch
self.logger.info(f"[KOMM] Var4 ohne Marker: erkenne kommKz={kommkz} aus Wert '{tlf[:20]}...'")
slot_marker = create_slot_marker(kommkz)
update_data = {
'tlf': '', # Empty Slot = leerer Wert
'bemerkung': slot_marker,
'online': False
}
log_value = synced_value if synced_value else tlf
await self.advoware.update_kommunikation(betnr, komm_id, update_data)
self.logger.info(f"[KOMM] ✅ Created empty slot: komm_id={komm_id}, kommKz={kommkz}, original_value='{log_value[:30]}...'")
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler beim Erstellen von Empty Slot: {e}")
self.logger.error(traceback.format_exc())
async def _revert_advoware_change(
self,
betnr: int,
advo_komm: Dict,
espo_synced_value: str,
advo_current_value: str,
advo_bet: Dict
) -> None:
"""
Revertiert Var6-Änderung in Advoware zurück auf EspoCRM-Wert
Verwendet bei direction='to_advoware' (EspoCRM wins):
- User hat in Advoware geändert
- Aber EspoCRM soll gewinnen
- → Setze Advoware zurück auf EspoCRM-Wert
Args:
advo_komm: Advoware Kommunikation mit Änderung
espo_synced_value: Der Wert der mit EspoCRM synchronisiert war (aus Marker)
advo_current_value: Der neue Wert in Advoware (User-Änderung)
"""
try:
komm_id = advo_komm['id']
bemerkung = advo_komm.get('bemerkung', '')
marker = parse_marker(bemerkung)
if not marker:
self.logger.error(f"[KOMM] Var6 ohne Marker - sollte nicht passieren! komm_id={komm_id}")
return
kommkz = marker['kommKz']
user_text = marker.get('user_text', '')
# Revert: Setze tlf zurück auf EspoCRM-Wert
new_marker = create_marker(espo_synced_value, kommkz, user_text)
update_data = {
'tlf': espo_synced_value,
'bemerkung': new_marker,
'online': advo_komm.get('online', False)
}
await self.advoware.update_kommunikation(betnr, komm_id, update_data)
self.logger.info(f"[KOMM] ✅ Reverted Var6: '{advo_current_value[:30]}...''{espo_synced_value[:30]}...' (komm_id={komm_id})")
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler beim Revert von Var6: {e}")
self.logger.error(traceback.format_exc())
def _needs_update(self, advo_komm: Dict, espo_item: Dict) -> bool:
"""Prüft ob Update nötig ist"""
current_value = (advo_komm.get('tlf') or '').strip()
new_value = espo_item['value'].strip()
current_online = advo_komm.get('online', False)
new_online = espo_item.get('primary', False)
return current_value != new_value or current_online != new_online
async def _update_kommunikation(self, betnr: int, advo_komm: Dict, espo_item: Dict) -> None:
"""Updated Advoware Kommunikation"""
try:
komm_id = advo_komm['id']
value = espo_item['value']
# Erkenne kommKz (sollte aus Marker kommen)
bemerkung = advo_komm.get('bemerkung') or ''
marker = parse_marker(bemerkung)
kommkz = marker['kommKz'] if marker else detect_kommkz(value, espo_type=espo_item.get('type'))
# Behalte User-Bemerkung
user_text = get_user_bemerkung(advo_komm)
new_marker = create_marker(value, kommkz, user_text)
update_data = {
'tlf': value,
'bemerkung': new_marker,
'online': espo_item.get('primary', False)
}
await self.advoware.update_kommunikation(betnr, komm_id, update_data)
self.logger.info(f"[KOMM] ✅ Updated: komm_id={komm_id}, value={value[:30]}...")
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler beim Update: {e}")
self.logger.error(traceback.format_exc())
async def _create_or_reuse_kommunikation(self, betnr: int, espo_item: Dict,
advo_kommunikationen: List[Dict]) -> bool:
"""
Erstellt neue Kommunikation oder nutzt leeren Slot
Returns:
True wenn erfolgreich erstellt/reused
"""
try:
value = espo_item['value']
# Erkenne kommKz mit EspoCRM type
espo_type = espo_item.get('type', 'email' if '@' in value else None)
kommkz = detect_kommkz(value, espo_type=espo_type)
self.logger.info(f"[KOMM] 🔍 kommKz detection: value='{value[:30]}...', espo_type={espo_type}, kommKz={kommkz}")
# Suche leeren Slot mit passendem kommKz
empty_slot = find_empty_slot(kommkz, advo_kommunikationen)
new_marker = create_marker(value, kommkz)
if empty_slot:
# ========== REUSE SLOT ==========
komm_id = empty_slot['id']
self.logger.info(f"[KOMM] ♻️ Reusing empty slot: komm_id={komm_id}, kommKz={kommkz}")
update_data = {
'tlf': value,
'bemerkung': new_marker,
'online': espo_item.get('primary', False)
}
await self.advoware.update_kommunikation(betnr, komm_id, update_data)
self.logger.info(f"[KOMM] ✅ Slot reused successfully: value='{value[:30]}...'")
else:
# ========== CREATE NEW ==========
self.logger.info(f"[KOMM] Creating new kommunikation entry: kommKz={kommkz}")
create_data = {
'tlf': value,
'bemerkung': new_marker,
'kommKz': kommkz,
'online': espo_item.get('primary', False)
}
await self.advoware.create_kommunikation(betnr, create_data)
self.logger.info(f"[KOMM] ✅ Created new: value='{value[:30]}...', kommKz={kommkz}")
return True
except Exception as e:
import traceback
self.logger.error(f"[KOMM] Fehler beim Erstellen/Reuse: {e}")
self.logger.error(traceback.format_exc())
return False
# ========== CHANGE DETECTION ==========
def detect_kommunikation_changes(old_bet: Dict, new_bet: Dict) -> bool:
"""
Erkennt Änderungen in Kommunikationen via rowId
Args:
old_bet: Alte Beteiligte-Daten (mit kommunikation[])
new_bet: Neue Beteiligte-Daten (mit kommunikation[])
Returns:
True wenn Änderungen erkannt
"""
old_komm = old_bet.get('kommunikation', [])
new_komm = new_bet.get('kommunikation', [])
# Check Count
if len(old_komm) != len(new_komm):
return True
# Check rowIds
old_row_ids = {k.get('rowId') for k in old_komm}
new_row_ids = {k.get('rowId') for k in new_komm}
return old_row_ids != new_row_ids
def detect_espocrm_kommunikation_changes(old_data: Dict, new_data: Dict) -> bool:
"""
Erkennt Änderungen in EspoCRM emailAddressData/phoneNumberData
Returns:
True wenn Änderungen erkannt
"""
old_emails = old_data.get('emailAddressData', [])
new_emails = new_data.get('emailAddressData', [])
old_phones = old_data.get('phoneNumberData', [])
new_phones = new_data.get('phoneNumberData', [])
# Einfacher Vergleich: Count und Values
if len(old_emails) != len(new_emails) or len(old_phones) != len(new_phones):
return True
old_email_values = {e.get('emailAddress') for e in old_emails}
new_email_values = {e.get('emailAddress') for e in new_emails}
old_phone_values = {p.get('phoneNumber') for p in old_phones}
new_phone_values = {p.get('phoneNumber') for p in new_phones}
return old_email_values != new_email_values or old_phone_values != new_phone_values