bsiggel/motia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement bidirectional synchronization utilities for Advoware and EspoCRM communications

Browse Source 
- Added KommunikationSyncManager class to handle synchronization logic.
- Implemented methods for loading data, computing diffs, and applying changes between Advoware and EspoCRM.
- Introduced 3-way diffing mechanism to intelligently resolve conflicts.
- Added helper methods for creating empty slots and detecting changes in communications.
- Enhanced logging for better traceability during synchronization processes.
This commit is contained in:
bitbylaw
2026-02-08 19:53:40 +00:00
parent da9a962858
commit ebbbf419ee
23 changed files with 7626 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
bitbylaw/steps/vmh/beteiligte_sync_event_step.py
View File

@@ -1,7 +1,13 @@
from typing import Dict, Any, Optional
from services.advoware import AdvowareAPI
from services.advoware_service import AdvowareService
from services.espocrm import EspoCRMAPI
from services.espocrm_mapper import BeteiligteMapper
from services.beteiligte_sync_utils import BeteiligteSync
from services.kommunikation_sync_utils import (
KommunikationSyncManager,
detect_kommunikation_changes
)
import json
import redis
from config import Config
@@ -54,6 +60,10 @@ async def handler(event_data, context):
sync_utils = BeteiligteSync(espocrm, redis_client, context)
mapper = BeteiligteMapper()
# Kommunikation Sync Manager
advo_service = AdvowareService(context)
komm_sync = KommunikationSyncManager(advo_service, espocrm, context)
try:
# 1. ACQUIRE LOCK (verhindert parallele Syncs)
lock_acquired = await sync_utils.acquire_sync_lock(entity_id)
@@ -85,7 +95,7 @@ async def handler(event_data, context):
# FALL B: Existiert (hat betnr) → UPDATE oder CHECK
elif betnr:
context.logger.info(f"♻️ Existierender Beteiligter (betNr: {betnr}) → UPDATE/CHECK")
await handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_utils, mapper, context)
await handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_utils, mapper, komm_sync, context)
# FALL C: DELETE (TODO: Implementierung später)
elif action == 'delete':
@@ -112,6 +122,28 @@ async def handler(event_data, context):
pass
async def run_kommunikation_sync(entity_id: str, betnr: int, komm_sync, context, direction: str = 'both') -> Dict[str, Any]:
"""
Helper: Führt Kommunikation-Sync aus mit Error-Handling
Args:
direction: 'both' (bidirektional), 'to_advoware' (nur EspoCRM→Advoware), 'to_espocrm' (nur Advoware→EspoCRM)
Returns:
Sync-Ergebnis oder None bei Fehler
"""
context.logger.info(f"📞 Starte Kommunikation-Sync (direction={direction})...")
try:
komm_result = await komm_sync.sync_bidirectional(entity_id, betnr, direction=direction)
context.logger.info(f"✅ Kommunikation synced: {komm_result}")
return komm_result
except Exception as e:
context.logger.error(f"⚠️ Kommunikation-Sync fehlgeschlagen: {e}")
import traceback
context.logger.error(traceback.format_exc())
return None
async def handle_create(entity_id, espo_entity, espocrm, advoware, sync_utils, mapper, context):
"""Erstellt neuen Beteiligten in Advoware"""
try:
@@ -167,7 +199,7 @@ async def handle_create(entity_id, espo_entity, espocrm, advoware, sync_utils, m
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
async def handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_utils, mapper, context):
async def handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_utils, mapper, komm_sync, context):
"""Synchronisiert existierenden Beteiligten"""
try:
context.logger.info(f"🔍 Fetch von Advoware betNr={betnr}...")
@@ -204,9 +236,18 @@ async def handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_u
context.logger.info(f"⏱️ Vergleich: {comparison}")
# KEIN SYNC NÖTIG
# KOMMUNIKATION-ÄNDERUNGSERKENNUNG (zusätzlich zu Stammdaten)
# Speichere alte Version für späteren Vergleich
old_advo_entity = advo_entity.copy()
komm_changes_detected = False
# KEIN STAMMDATEN-SYNC NÖTIG (aber Kommunikation könnte geändert sein)
if comparison == 'no_change':
context.logger.info(f"✅ Keine Änderungen, Sync übersprungen")
context.logger.info(f"✅ Keine Stammdaten-Änderungen erkannt")
# KOMMUNIKATION SYNC: Prüfe trotzdem Kommunikationen
await run_kommunikation_sync(entity_id, betnr, komm_sync, context)
await sync_utils.release_sync_lock(entity_id, 'clean')
return
@@ -230,27 +271,35 @@ async def handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_u
elif isinstance(put_result, dict):
new_rowid = put_result.get('rowId')
# Release Lock + Update rowId in einem Call (effizienter!)
context.logger.info(f"✅ Advoware STAMMDATEN aktualisiert, rowId: {new_rowid[:20] if new_rowid else 'N/A'}...")
# KOMMUNIKATION SYNC: Immer ausführen nach Stammdaten-Update
await run_kommunikation_sync(entity_id, betnr, komm_sync, context)
# Release Lock NACH Kommunikation-Sync + Update rowId
await sync_utils.release_sync_lock(
entity_id,
'clean',
extra_fields={'advowareRowId': new_rowid}
)
context.logger.info(f"✅ Advoware aktualisiert, rowId in EspoCRM geschrieben: {new_rowid[:20] if new_rowid else 'N/A'}...")
# ADVOWARE NEUER → Update EspoCRM
elif comparison == 'advoware_newer':
context.logger.info(f"📥 Advoware ist neuer → Update EspoCRM STAMMDATEN")
espo_data = mapper.map_advoware_to_cbeteiligte(advo_entity)
await espocrm.update_entity('CBeteiligte', entity_id, espo_data)
context.logger.info(f"✅ EspoCRM STAMMDATEN aktualisiert")
# KOMMUNIKATION SYNC: Immer ausführen nach Stammdaten-Update
await run_kommunikation_sync(entity_id, betnr, komm_sync, context)
# Release Lock NACH Kommunikation-Sync + Update rowId
await sync_utils.release_sync_lock(
entity_id,
'clean',
extra_fields={'advowareRowId': advo_entity.get('rowId')}
)
context.logger.info(f"✅ EspoCRM aktualisiert")
# KONFLIKT → EspoCRM WINS
elif comparison == 'conflict':
@@ -286,9 +335,19 @@ async def handle_update(entity_id, betnr, espo_entity, espocrm, advoware, sync_u
extra_fields={'advowareRowId': new_rowid}
)
context.logger.info(f"✅ Konflikt gelöst (EspoCRM won), neue rowId: {new_rowid[:20] if new_rowid else 'N/A'}...")
# KOMMUNIKATION SYNC: NUR EspoCRM→Advoware (EspoCRM wins!)
await run_kommunikation_sync(entity_id, betnr, komm_sync, context, direction='to_advoware')
# Release Lock NACH Kommunikation-Sync
await sync_utils.release_sync_lock(entity_id, 'clean')
except Exception as e:
context.logger.error(f"❌ UPDATE fehlgeschlagen: {e}")
import traceback
context.logger.error(traceback.format_exc())
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
# Alias für Tests/externe Aufrufe
handle = handler