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

Vollständige Advoware-EspoCRM Integration implementiert

Browse Source 
- Advoware API Proxy für alle HTTP-Methoden (GET/POST/PUT/DELETE)
- EspoCRM Webhook-Receiver für Beteiligte CRUD-Operationen
- Redis-basierte Deduplikation für Webhook-Events
- Event-driven Synchronisations-Handler (Placeholder)
- Detaillierte README.md mit Setup und Verwendungsanleitung
- Fehlerbehebungen für Context-Attribute und Redis-Verbindungen
This commit is contained in:
root
2025-10-20 11:42:52 +00:00
parent 1f893812b2
commit 805d8c7362
11 changed files with 611 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
bitbylaw/steps/vmh/beteiligte_sync_event_step.py Normal file
View File

@@ -0,0 +1,52 @@
from services.advoware import AdvowareAPI
import json
import redis
from config import Config
config = {
'type': 'event',
'name': 'VMH Beteiligte Sync',
'description': 'Synchronisiert Beteiligte Entities von Advoware nach Änderungen (Create/Update/Delete)',
'subscribes': ['vmh.beteiligte.create', 'vmh.beteiligte.update', 'vmh.beteiligte.delete'],
'flows': ['vmh'],
'emits': []
}
async def handler(event_data, context):
try:
entity_id = event_data.get('entity_id')
action = event_data.get('action', 'unknown')
if not entity_id:
context.logger.error("Keine entity_id im Event gefunden")
return
context.logger.info(f"Starte {action.upper()} Sync für Beteiligte Entity: {entity_id}")
# Advoware API initialisieren (für später)
# advoware = AdvowareAPI(context)
# PLATZHALTER: Für jetzt nur loggen, keine API-Anfrage
context.logger.info(f"PLATZHALTER: {action.upper()} Sync für Entity {entity_id} würde hier Advoware API aufrufen")
context.logger.info(f"PLATZHALTER: Entity-Daten würden hier verarbeitet werden")
# TODO: Hier die Entity in das Zielsystem syncen (EspoCRM?)
# Für Create: Neu anlegen
# Für Update: Aktualisieren
# Für Delete: Löschen
# Entferne die ID aus der entsprechenden Pending-Queue
redis_client = redis.Redis(
host=Config.REDIS_HOST,
port=int(Config.REDIS_PORT),
db=int(Config.REDIS_DB_ADVOWARE_CACHE),
decode_responses=True
)
pending_key = f'vmh:beteiligte:{action}_pending'
redis_client.srem(pending_key, entity_id)
context.logger.info(f"Entity {entity_id} aus {action.upper()}-Pending-Queue entfernt")
except Exception as e:
context.logger.error(f"Fehler beim {event_data.get('action', 'unknown').upper()} Sync von Beteiligte Entity: {e}")
context.logger.error(f"Event Data: {event_data}")

96
bitbylaw/steps/vmh/webhook/beteiligte_create_api_step.py Normal file
View File

@@ -0,0 +1,96 @@
from typing import Any, Dict, Set
import json
import redis
from config import Config
import datetime
config = {
'type': 'api',
'name': 'VMH Webhook Beteiligte Create',
'description': 'Empfängt Create-Webhooks von EspoCRM für Beteiligte',
'path': '/vmh/webhook/beteiligte/create',
'method': 'POST',
'flows': ['vmh'],
'emits': ['vmh.beteiligte.create']
}
async def handler(req, context):
try:
# Payload aus dem Request-Body holen
payload = req.get('body', [])
# Detailliertes Logging
context.logger.info("VMH Webhook Beteiligte Create empfangen")
context.logger.info(f"Headers: {json.dumps(dict(req.get('headers', {})), indent=2)}")
context.logger.info(f"Payload: {json.dumps(payload, indent=2, ensure_ascii=False)}")
# Sammle alle IDs aus dem Batch
ids_to_sync: Set[str] = set()
if isinstance(payload, list):
for entity in payload:
if isinstance(entity, dict) and 'id' in entity:
entity_id = entity['id']
ids_to_sync.add(entity_id)
context.logger.info(f"Create Entity ID gefunden: {entity_id}")
elif isinstance(payload, dict) and 'id' in payload:
ids_to_sync.add(payload['id'])
context.logger.info(f"Create Single Entity ID gefunden: {payload['id']}")
context.logger.info(f"Insgesamt {len(ids_to_sync)} eindeutige IDs zum Create-Sync gefunden")
# Redis Verbindung für Deduplizierung
redis_client = redis.Redis(
host=Config.REDIS_HOST,
port=int(Config.REDIS_PORT),
db=int(Config.REDIS_DB_ADVOWARE_CACHE),
decode_responses=True
)
# Deduplizierung: Prüfe welche IDs schon in der Queue sind
pending_key = 'vmh:beteiligte:create_pending'
existing_ids = redis_client.smembers(pending_key)
new_ids = ids_to_sync - set(existing_ids)
if new_ids:
# Füge neue IDs zur Pending-Queue hinzu
redis_client.sadd(pending_key, *new_ids)
context.logger.info(f"{len(new_ids)} neue IDs zur Create-Sync-Queue hinzugefügt: {list(new_ids)}")
# Emittiere Events für neue IDs
for entity_id in new_ids:
await context.emit({
'topic': 'vmh.beteiligte.create',
'data': {
'entity_id': entity_id,
'action': 'create',
'source': 'webhook',
'timestamp': req.get('timestamp') or datetime.datetime.now().isoformat()
}
})
context.logger.info(f"Create-Event emittiert für ID: {entity_id}")
else:
context.logger.info("Keine neuen IDs zum Create-Sync gefunden")
context.logger.info("VMH Create Webhook erfolgreich verarbeitet")
return {
'status': 200,
'body': {
'status': 'received',
'action': 'create',
'new_ids_count': len(new_ids) if 'new_ids' in locals() else 0,
'total_ids_in_batch': len(ids_to_sync)
}
}
except Exception as e:
context.logger.error(f"Fehler beim Verarbeiten des VMH Create Webhooks: {e}")
context.logger.error(f"Request: {req}")
return {
'status': 500,
'body': {
'error': 'Internal server error',
'details': str(e)
}
}

77
bitbylaw/steps/vmh/webhook/beteiligte_delete_api_step.py Normal file
View File

@@ -0,0 +1,77 @@
from typing import Any, Dict, Set
import json
import redis
from config import Config
import datetime
config = {
'type': 'api',
'name': 'VMH Webhook Beteiligte Delete',
'description': 'Empfängt Delete-Webhooks von EspoCRM für Beteiligte',
'path': '/vmh/webhook/beteiligte/delete',
'method': 'POST',
'flows': ['vmh'],
'emits': ['vmh.beteiligte.delete']
}
async def handler(req, context):
try:
payload = req.get('body', [])
context.logger.info("VMH Webhook Beteiligte Delete empfangen")
context.logger.info(f"Payload: {json.dumps(payload, indent=2, ensure_ascii=False)}")
ids_to_sync: Set[str] = set()
if isinstance(payload, list):
for entity in payload:
if isinstance(entity, dict) and 'id' in entity:
entity_id = entity['id']
ids_to_sync.add(entity_id)
elif isinstance(payload, dict) and 'id' in payload:
ids_to_sync.add(payload['id'])
context.logger.info(f"{len(ids_to_sync)} IDs zum Delete-Sync gefunden")
# Redis Verbindung für Deduplizierung
redis_client = redis.Redis(
host=Config.REDIS_HOST,
port=int(Config.REDIS_PORT),
db=int(Config.REDIS_DB_ADVOWARE_CACHE),
decode_responses=True
)
pending_key = 'vmh:beteiligte:pending_delete'
existing_ids = redis_client.smembers(pending_key)
new_ids = ids_to_sync - set(existing_ids)
if new_ids:
redis_client.sadd(pending_key, *new_ids)
context.logger.info(f"{len(new_ids)} neue IDs zur Delete-Queue hinzugefügt")
for entity_id in new_ids:
await context.emit({
'topic': 'vmh.beteiligte.delete',
'data': {
'entity_id': entity_id,
'action': 'delete',
'source': 'webhook',
'timestamp': req.get('timestamp') or datetime.datetime.now().isoformat()
}
})
return {
'status': 200,
'body': {
'status': 'received',
'action': 'delete',
'new_ids_count': len(new_ids) if 'new_ids' in locals() else 0
}
}
except Exception as e:
context.logger.error(f"Fehler beim Delete-Webhook: {e}")
return {
'status': 500,
'body': {'error': 'Internal server error', 'details': str(e)}
}

96
bitbylaw/steps/vmh/webhook/beteiligte_update_api_step.py Normal file
View File

@@ -0,0 +1,96 @@
from typing import Any, Dict, Set
import json
import redis
from config import Config
import datetime
config = {
'type': 'api',
'name': 'VMH Webhook Beteiligte Update',
'description': 'Empfängt Update-Webhooks von EspoCRM für Beteiligte',
'path': '/vmh/webhook/beteiligte/update',
'method': 'POST',
'flows': ['vmh'],
'emits': ['vmh.beteiligte.update']
}
async def handler(req, context):
try:
# Payload aus dem Request-Body holen
payload = req.get('body', [])
# Detailliertes Logging
context.logger.info("VMH Webhook Beteiligte Update empfangen")
context.logger.info(f"Headers: {json.dumps(dict(req.get('headers', {})), indent=2)}")
context.logger.info(f"Payload: {json.dumps(payload, indent=2, ensure_ascii=False)}")
# Sammle alle IDs aus dem Batch
ids_to_sync: Set[str] = set()
if isinstance(payload, list):
for entity in payload:
if isinstance(entity, dict) and 'id' in entity:
entity_id = entity['id']
ids_to_sync.add(entity_id)
context.logger.info(f"Update Entity ID gefunden: {entity_id}")
elif isinstance(payload, dict) and 'id' in payload:
ids_to_sync.add(payload['id'])
context.logger.info(f"Update Single Entity ID gefunden: {payload['id']}")
context.logger.info(f"Insgesamt {len(ids_to_sync)} eindeutige IDs zum Update-Sync gefunden")
# Redis Verbindung für Deduplizierung
redis_client = redis.Redis(
host=Config.REDIS_HOST,
port=int(Config.REDIS_PORT),
db=int(Config.REDIS_DB_ADVOWARE_CACHE),
decode_responses=True
)
# Deduplizierung: Prüfe welche IDs schon in der Queue sind
pending_key = 'vmh:beteiligte:update_pending'
existing_ids = redis_client.smembers(pending_key)
new_ids = ids_to_sync - set(existing_ids)
if new_ids:
# Füge neue IDs zur Pending-Queue hinzu
redis_client.sadd(pending_key, *new_ids)
context.logger.info(f"{len(new_ids)} neue IDs zur Update-Sync-Queue hinzugefügt: {list(new_ids)}")
# Emittiere Events für neue IDs
for entity_id in new_ids:
await context.emit({
'topic': 'vmh.beteiligte.update',
'data': {
'entity_id': entity_id,
'action': 'update',
'source': 'webhook',
'timestamp': req.get('timestamp') or datetime.datetime.now().isoformat()
}
})
context.logger.info(f"Update-Event emittiert für ID: {entity_id}")
else:
context.logger.info("Keine neuen IDs zum Update-Sync gefunden")
context.logger.info("VMH Update Webhook erfolgreich verarbeitet")
return {
'status': 200,
'body': {
'status': 'received',
'action': 'update',
'new_ids_count': len(new_ids) if 'new_ids' in locals() else 0,
'total_ids_in_batch': len(ids_to_sync)
}
}
except Exception as e:
context.logger.error(f"Fehler beim Verarbeiten des VMH Update Webhooks: {e}")
context.logger.error(f"Request: {req}")
return {
'status': 500,
'body': {
'error': 'Internal server error',
'details': str(e)
}
}