feat(webhooks): Add webhook handlers for Beteiligte and Document entities
- Implemented create, update, and delete webhook handlers for Beteiligte. - Implemented create, update, and delete webhook handlers for Document entities. - Added logging and error handling for each webhook handler. - Created a universal step for generating document previews. - Ensured payload validation and entity ID extraction for batch processing.
This commit is contained in:
bsiggel
0
src/steps/crm/akte/__init__.py
Normal file
0
src/steps/crm/akte/__init__.py
Normal file
165
src/steps/crm/akte/akte_sync_cron_step.py
Normal file
165
src/steps/crm/akte/akte_sync_cron_step.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
Akte Sync - Cron Poller
|
||||
|
||||
Polls two Redis Sorted Sets every 10 seconds (10 s debounce each):
|
||||
|
||||
advoware:pending_aktennummern – written by Windows Advoware Watcher
|
||||
{ aktennummer → timestamp }
|
||||
akte:pending_entity_ids – written by EspoCRM webhook
|
||||
{ akte_id → timestamp }
|
||||
|
||||
Eligibility (either flag triggers sync):
|
||||
syncSchalter AND aktivierungsstatus in valid list → Advoware sync
|
||||
aiAktivierungsstatus in valid list → xAI sync
|
||||
"""
|
||||
|
||||
from motia import FlowContext, cron
|
||||
|
||||
|
||||
config = {
|
||||
"name": "Akte Sync - Cron Poller",
|
||||
"description": "Poll Redis for pending Aktennummern and emit akte.sync events (10 s debounce)",
|
||||
"flows": ["akte-sync"],
|
||||
"triggers": [cron("*/10 * * * * *")],
|
||||
"enqueues": ["akte.sync"],
|
||||
}
|
||||
|
||||
# Queue 1: written by Windows Advoware Watcher (keyed by Aktennummer)
|
||||
PENDING_ADVO_KEY = "advoware:pending_aktennummern"
|
||||
PROCESSING_ADVO_KEY = "advoware:processing_aktennummern"
|
||||
|
||||
# Queue 2: written by EspoCRM webhook (keyed by entity ID)
|
||||
PENDING_ID_KEY = "akte:pending_entity_ids"
|
||||
PROCESSING_ID_KEY = "akte:processing_entity_ids"
|
||||
|
||||
DEBOUNCE_SECS = 10
|
||||
|
||||
VALID_ADVOWARE_STATUSES = {'import', 'neu', 'new', 'aktiv', 'active'}
|
||||
VALID_AI_STATUSES = {'new', 'neu', 'aktiv', 'active'}
|
||||
|
||||
|
||||
async def handler(input_data: None, ctx: FlowContext) -> None:
|
||||
import time
|
||||
from services.redis_client import get_redis_client
|
||||
from services.espocrm import EspoCRMAPI
|
||||
|
||||
ctx.logger.info("=" * 60)
|
||||
ctx.logger.info("⏰ AKTE CRON POLLER")
|
||||
|
||||
redis_client = get_redis_client(strict=False)
|
||||
if not redis_client:
|
||||
ctx.logger.error("❌ Redis unavailable")
|
||||
ctx.logger.info("=" * 60)
|
||||
return
|
||||
|
||||
espocrm = EspoCRMAPI(ctx)
|
||||
cutoff = time.time() - DEBOUNCE_SECS
|
||||
|
||||
advo_pending = redis_client.zcard(PENDING_ADVO_KEY)
|
||||
id_pending = redis_client.zcard(PENDING_ID_KEY)
|
||||
ctx.logger.info(f" Pending (aktennr) : {advo_pending}")
|
||||
ctx.logger.info(f" Pending (akte_id) : {id_pending}")
|
||||
|
||||
processed = False
|
||||
|
||||
# ── Queue 1: Advoware Watcher (by Aktennummer) ─────────────────────
|
||||
advo_entries = redis_client.zrangebyscore(PENDING_ADVO_KEY, min=0, max=cutoff, start=0, num=1)
|
||||
if advo_entries:
|
||||
aktennr = advo_entries[0]
|
||||
if isinstance(aktennr, bytes):
|
||||
aktennr = aktennr.decode()
|
||||
|
||||
score = redis_client.zscore(PENDING_ADVO_KEY, aktennr) or 0
|
||||
age = time.time() - score
|
||||
redis_client.zrem(PENDING_ADVO_KEY, aktennr)
|
||||
redis_client.sadd(PROCESSING_ADVO_KEY, aktennr)
|
||||
|
||||
ctx.logger.info(f"📋 Aktennummer: {aktennr} (age={age:.1f}s)")
|
||||
processed = True
|
||||
|
||||
try:
|
||||
result = await espocrm.list_entities(
|
||||
'CAkten',
|
||||
where=[{'type': 'equals', 'attribute': 'aktennummer', 'value': int(aktennr)}],
|
||||
max_size=1,
|
||||
)
|
||||
if not result or not result.get('list'):
|
||||
ctx.logger.warn(f"⚠️ No CAkten found for aktennummer={aktennr} – removing")
|
||||
redis_client.srem(PROCESSING_ADVO_KEY, aktennr)
|
||||
else:
|
||||
akte = result['list'][0]
|
||||
await _emit_if_eligible(akte, aktennr, ctx)
|
||||
redis_client.srem(PROCESSING_ADVO_KEY, aktennr)
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Error (aktennr queue) {aktennr}: {e}")
|
||||
redis_client.zadd(PENDING_ADVO_KEY, {aktennr: time.time()})
|
||||
redis_client.srem(PROCESSING_ADVO_KEY, aktennr)
|
||||
raise
|
||||
|
||||
# ── Queue 2: EspoCRM Webhook (by Entity ID) ────────────────────────
|
||||
id_entries = redis_client.zrangebyscore(PENDING_ID_KEY, min=0, max=cutoff, start=0, num=1)
|
||||
if id_entries:
|
||||
akte_id = id_entries[0]
|
||||
if isinstance(akte_id, bytes):
|
||||
akte_id = akte_id.decode()
|
||||
|
||||
score = redis_client.zscore(PENDING_ID_KEY, akte_id) or 0
|
||||
age = time.time() - score
|
||||
redis_client.zrem(PENDING_ID_KEY, akte_id)
|
||||
redis_client.sadd(PROCESSING_ID_KEY, akte_id)
|
||||
|
||||
ctx.logger.info(f"📋 Entity ID: {akte_id} (age={age:.1f}s)")
|
||||
processed = True
|
||||
|
||||
try:
|
||||
akte = await espocrm.get_entity('CAkten', akte_id)
|
||||
if not akte:
|
||||
ctx.logger.warn(f"⚠️ No CAkten found for id={akte_id} – removing")
|
||||
redis_client.srem(PROCESSING_ID_KEY, akte_id)
|
||||
else:
|
||||
await _emit_if_eligible(akte, None, ctx)
|
||||
redis_client.srem(PROCESSING_ID_KEY, akte_id)
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Error (entity-id queue) {akte_id}: {e}")
|
||||
redis_client.zadd(PENDING_ID_KEY, {akte_id: time.time()})
|
||||
redis_client.srem(PROCESSING_ID_KEY, akte_id)
|
||||
raise
|
||||
|
||||
if not processed:
|
||||
if advo_pending > 0 or id_pending > 0:
|
||||
ctx.logger.info(f"⏸️ Entries pending but all too recent (< {DEBOUNCE_SECS}s)")
|
||||
else:
|
||||
ctx.logger.info("✓ Both queues empty")
|
||||
|
||||
ctx.logger.info("=" * 60)
|
||||
|
||||
|
||||
async def _emit_if_eligible(akte: dict, aktennr, ctx: FlowContext) -> None:
|
||||
"""Check eligibility and emit akte.sync if applicable."""
|
||||
akte_id = akte['id']
|
||||
# Prefer aktennr from argument; fall back to entity field
|
||||
aktennummer = aktennr or akte.get('aktennummer')
|
||||
sync_schalter = akte.get('syncSchalter', False)
|
||||
aktivierungsstatus = str(akte.get('aktivierungsstatus') or '').lower()
|
||||
ai_status = str(akte.get('aiAktivierungsstatus') or '').lower()
|
||||
|
||||
advoware_eligible = bool(aktennummer) and sync_schalter and aktivierungsstatus in VALID_ADVOWARE_STATUSES
|
||||
xai_eligible = ai_status in VALID_AI_STATUSES
|
||||
|
||||
ctx.logger.info(f" akte_id : {akte_id}")
|
||||
ctx.logger.info(f" aktennummer : {aktennummer or '—'}")
|
||||
ctx.logger.info(f" aktivierungsstatus : {aktivierungsstatus} ({'✅' if advoware_eligible else '⏭️'})")
|
||||
ctx.logger.info(f" aiAktivierungsstatus : {ai_status} ({'✅' if xai_eligible else '⏭️'})")
|
||||
|
||||
if not advoware_eligible and not xai_eligible:
|
||||
ctx.logger.warn(f"⚠️ Akte {akte_id} not eligible for any sync")
|
||||
return
|
||||
|
||||
await ctx.enqueue({
|
||||
'topic': 'akte.sync',
|
||||
'data': {
|
||||
'akte_id': akte_id,
|
||||
'aktennummer': aktennummer, # may be None for xAI-only Akten
|
||||
},
|
||||
})
|
||||
ctx.logger.info(f"📤 akte.sync emitted (akte_id={akte_id}, aktennummer={aktennummer or '—'})")
|
||||
435
src/steps/crm/akte/akte_sync_event_step.py
Normal file
435
src/steps/crm/akte/akte_sync_event_step.py
Normal file
@@ -0,0 +1,435 @@
|
||||
"""
|
||||
Akte Sync - Event Handler
|
||||
|
||||
Unified sync for one CAkten entity across all configured backends:
|
||||
- Advoware (3-way merge: Windows ↔ EspoCRM ↔ History)
|
||||
- xAI (Blake3 hash-based upload to Collection)
|
||||
|
||||
Both run in the same event to keep CDokumente perfectly in sync.
|
||||
|
||||
Trigger: akte.sync { akte_id, aktennummer }
|
||||
Lock: Redis per-Akte (30 min TTL, prevents double-sync of same Akte)
|
||||
Parallel: Different Akten sync simultaneously.
|
||||
|
||||
Enqueues:
|
||||
- document.generate_preview (after CREATE / UPDATE_ESPO)
|
||||
"""
|
||||
|
||||
from typing import Dict, Any
|
||||
from datetime import datetime
|
||||
from motia import FlowContext, queue
|
||||
|
||||
|
||||
config = {
|
||||
"name": "Akte Sync - Event Handler",
|
||||
"description": "Unified sync for one Akte: Advoware 3-way merge + xAI upload",
|
||||
"flows": ["akte-sync"],
|
||||
"triggers": [queue("akte.sync")],
|
||||
"enqueues": ["document.generate_preview"],
|
||||
}
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Entry point
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async def handler(event_data: Dict[str, Any], ctx: FlowContext) -> None:
|
||||
akte_id = event_data.get('akte_id')
|
||||
aktennummer = event_data.get('aktennummer')
|
||||
|
||||
ctx.logger.info("=" * 80)
|
||||
ctx.logger.info("🔄 AKTE SYNC STARTED")
|
||||
ctx.logger.info(f" Aktennummer : {aktennummer}")
|
||||
ctx.logger.info(f" EspoCRM ID : {akte_id}")
|
||||
ctx.logger.info("=" * 80)
|
||||
|
||||
from services.redis_client import get_redis_client
|
||||
from services.espocrm import EspoCRMAPI
|
||||
|
||||
redis_client = get_redis_client(strict=False)
|
||||
if not redis_client:
|
||||
ctx.logger.error("❌ Redis unavailable")
|
||||
return
|
||||
|
||||
lock_key = f"akte_sync:{akte_id}"
|
||||
lock_acquired = redis_client.set(lock_key, datetime.now().isoformat(), nx=True, ex=1800)
|
||||
if not lock_acquired:
|
||||
ctx.logger.warn(f"⏸️ Lock busy for Akte {akte_id} – requeueing")
|
||||
raise RuntimeError(f"Lock busy for akte_id={akte_id}")
|
||||
|
||||
espocrm = EspoCRMAPI(ctx)
|
||||
|
||||
try:
|
||||
# ── Load Akte ──────────────────────────────────────────────────────
|
||||
akte = await espocrm.get_entity('CAkten', akte_id)
|
||||
if not akte:
|
||||
ctx.logger.error(f"❌ Akte {akte_id} not found in EspoCRM")
|
||||
return
|
||||
|
||||
# aktennummer can come from the event payload OR from the entity
|
||||
# (Akten without Advoware have no aktennummer)
|
||||
if not aktennummer:
|
||||
aktennummer = akte.get('aktennummer')
|
||||
|
||||
sync_schalter = akte.get('syncSchalter', False)
|
||||
aktivierungsstatus = str(akte.get('aktivierungsstatus') or '').lower()
|
||||
ai_aktivierungsstatus = str(akte.get('aiAktivierungsstatus') or '').lower()
|
||||
|
||||
ctx.logger.info(f"📋 Akte '{akte.get('name')}'")
|
||||
ctx.logger.info(f" syncSchalter : {sync_schalter}")
|
||||
ctx.logger.info(f" aktivierungsstatus : {aktivierungsstatus}")
|
||||
ctx.logger.info(f" aiAktivierungsstatus : {ai_aktivierungsstatus}")
|
||||
|
||||
# Advoware sync requires an aktennummer (Akten without Advoware won't have one)
|
||||
advoware_enabled = bool(aktennummer) and sync_schalter and aktivierungsstatus in ('import', 'neu', 'new', 'aktiv', 'active')
|
||||
xai_enabled = ai_aktivierungsstatus in ('new', 'neu', 'aktiv', 'active')
|
||||
|
||||
ctx.logger.info(f" Advoware sync : {'✅ ON' if advoware_enabled else '⏭️ OFF'}")
|
||||
ctx.logger.info(f" xAI sync : {'✅ ON' if xai_enabled else '⏭️ OFF'}")
|
||||
|
||||
if not advoware_enabled and not xai_enabled:
|
||||
ctx.logger.info("⏭️ Both syncs disabled – nothing to do")
|
||||
return
|
||||
|
||||
# ── ADVOWARE SYNC ──────────────────────────────────────────────────
|
||||
advoware_results = None
|
||||
if advoware_enabled:
|
||||
advoware_results = await _run_advoware_sync(akte, aktennummer, akte_id, espocrm, ctx)
|
||||
|
||||
# ── xAI SYNC ──────────────────────────────────────────────────────
|
||||
if xai_enabled:
|
||||
await _run_xai_sync(akte, akte_id, espocrm, ctx)
|
||||
|
||||
# ── Final Status ───────────────────────────────────────────────────
|
||||
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
final_update: Dict[str, Any] = {'globalLastSync': now, 'globalSyncStatus': 'synced'}
|
||||
if advoware_enabled:
|
||||
final_update['syncStatus'] = 'synced'
|
||||
final_update['lastSync'] = now
|
||||
# 'import' = erster Sync → danach auf 'aktiv' setzen
|
||||
if aktivierungsstatus == 'import':
|
||||
final_update['aktivierungsstatus'] = 'aktiv'
|
||||
ctx.logger.info("🔄 aktivierungsstatus: import → aktiv")
|
||||
if xai_enabled:
|
||||
final_update['aiSyncStatus'] = 'synced'
|
||||
final_update['aiLastSync'] = now
|
||||
# 'new' = Collection wurde gerade erstmalig angelegt → auf 'aktiv' setzen
|
||||
if ai_aktivierungsstatus == 'new':
|
||||
final_update['aiAktivierungsstatus'] = 'aktiv'
|
||||
ctx.logger.info("🔄 aiAktivierungsstatus: new → aktiv")
|
||||
|
||||
await espocrm.update_entity('CAkten', akte_id, final_update)
|
||||
# Clean up processing sets (both queues may have triggered this sync)
|
||||
if aktennummer:
|
||||
redis_client.srem("advoware:processing_aktennummern", aktennummer)
|
||||
redis_client.srem("akte:processing_entity_ids", akte_id)
|
||||
|
||||
ctx.logger.info("=" * 80)
|
||||
ctx.logger.info("✅ AKTE SYNC COMPLETE")
|
||||
if advoware_results:
|
||||
ctx.logger.info(f" Advoware: created={advoware_results['created']} updated={advoware_results['updated']} deleted={advoware_results['deleted']} errors={advoware_results['errors']}")
|
||||
ctx.logger.info("=" * 80)
|
||||
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Sync failed: {e}")
|
||||
import traceback
|
||||
ctx.logger.error(traceback.format_exc())
|
||||
|
||||
# Requeue for retry (into the appropriate queue(s))
|
||||
import time
|
||||
now_ts = time.time()
|
||||
if aktennummer:
|
||||
redis_client.zadd("advoware:pending_aktennummern", {aktennummer: now_ts})
|
||||
redis_client.zadd("akte:pending_entity_ids", {akte_id: now_ts})
|
||||
|
||||
try:
|
||||
await espocrm.update_entity('CAkten', akte_id, {
|
||||
'syncStatus': 'failed',
|
||||
'globalSyncStatus': 'failed',
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
|
||||
finally:
|
||||
if lock_acquired and redis_client:
|
||||
redis_client.delete(lock_key)
|
||||
ctx.logger.info(f"🔓 Lock released for Akte {aktennummer}")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Advoware 3-way merge
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async def _run_advoware_sync(
|
||||
akte: Dict[str, Any],
|
||||
aktennummer: str,
|
||||
akte_id: str,
|
||||
espocrm,
|
||||
ctx: FlowContext,
|
||||
) -> Dict[str, int]:
|
||||
from services.advoware_watcher_service import AdvowareWatcherService
|
||||
from services.advoware_history_service import AdvowareHistoryService
|
||||
from services.advoware_service import AdvowareService
|
||||
from services.advoware_document_sync_utils import AdvowareDocumentSyncUtils
|
||||
from services.blake3_utils import compute_blake3
|
||||
import mimetypes
|
||||
|
||||
watcher = AdvowareWatcherService(ctx)
|
||||
history_service = AdvowareHistoryService(ctx)
|
||||
advoware_service = AdvowareService(ctx)
|
||||
sync_utils = AdvowareDocumentSyncUtils(ctx)
|
||||
|
||||
results = {'created': 0, 'updated': 0, 'deleted': 0, 'skipped': 0, 'errors': 0}
|
||||
|
||||
ctx.logger.info("")
|
||||
ctx.logger.info("─" * 60)
|
||||
ctx.logger.info("📂 ADVOWARE SYNC")
|
||||
ctx.logger.info("─" * 60)
|
||||
|
||||
# ── Fetch from all 3 sources ───────────────────────────────────────
|
||||
espo_docs_result = await espocrm.list_related('CAkten', akte_id, 'dokumentes')
|
||||
espo_docs = espo_docs_result.get('list', [])
|
||||
|
||||
try:
|
||||
windows_files = await watcher.get_akte_files(aktennummer)
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Windows watcher failed: {e}")
|
||||
windows_files = []
|
||||
|
||||
try:
|
||||
advo_history = await history_service.get_akte_history(aktennummer)
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Advoware history failed: {e}")
|
||||
advo_history = []
|
||||
|
||||
ctx.logger.info(f" EspoCRM docs : {len(espo_docs)}")
|
||||
ctx.logger.info(f" Windows files : {len(windows_files)}")
|
||||
ctx.logger.info(f" History entries: {len(advo_history)}")
|
||||
|
||||
# ── Cleanup Windows list (only files in History) ───────────────────
|
||||
windows_files = sync_utils.cleanup_file_list(windows_files, advo_history)
|
||||
|
||||
# ── Build indexes by HNR (stable identifier from Advoware) ────────
|
||||
espo_by_hnr = {}
|
||||
for doc in espo_docs:
|
||||
if doc.get('hnr'):
|
||||
espo_by_hnr[doc['hnr']] = doc
|
||||
|
||||
history_by_hnr = {}
|
||||
for entry in advo_history:
|
||||
if entry.get('hNr'):
|
||||
history_by_hnr[entry['hNr']] = entry
|
||||
|
||||
windows_by_path = {f.get('path', '').lower(): f for f in windows_files}
|
||||
|
||||
all_hnrs = set(espo_by_hnr.keys()) | set(history_by_hnr.keys())
|
||||
ctx.logger.info(f" Unique HNRs : {len(all_hnrs)}")
|
||||
|
||||
# ── 3-way merge per HNR ───────────────────────────────────────────
|
||||
for hnr in all_hnrs:
|
||||
espo_doc = espo_by_hnr.get(hnr)
|
||||
history_entry = history_by_hnr.get(hnr)
|
||||
|
||||
windows_file = None
|
||||
if history_entry and history_entry.get('datei'):
|
||||
windows_file = windows_by_path.get(history_entry['datei'].lower())
|
||||
|
||||
if history_entry and history_entry.get('datei'):
|
||||
filename = history_entry['datei'].split('\\')[-1]
|
||||
elif espo_doc:
|
||||
filename = espo_doc.get('name', f'hnr_{hnr}')
|
||||
else:
|
||||
filename = f'hnr_{hnr}'
|
||||
|
||||
try:
|
||||
action = sync_utils.merge_three_way(espo_doc, windows_file, history_entry)
|
||||
ctx.logger.info(f" [{action.action:12s}] {filename} (hnr={hnr}) – {action.reason}")
|
||||
|
||||
if action.action == 'SKIP':
|
||||
results['skipped'] += 1
|
||||
|
||||
elif action.action == 'CREATE':
|
||||
if not windows_file:
|
||||
ctx.logger.error(f" ❌ CREATE: no Windows file for hnr {hnr}")
|
||||
results['errors'] += 1
|
||||
continue
|
||||
|
||||
content = await watcher.download_file(aktennummer, windows_file.get('relative_path', filename))
|
||||
blake3_hash = compute_blake3(content)
|
||||
mime_type, _ = mimetypes.guess_type(filename)
|
||||
mime_type = mime_type or 'application/octet-stream'
|
||||
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
attachment = await espocrm.upload_attachment_for_file_field(
|
||||
file_content=content,
|
||||
filename=filename,
|
||||
related_type='CDokumente',
|
||||
field='dokument',
|
||||
mime_type=mime_type,
|
||||
)
|
||||
new_doc = await espocrm.create_entity('CDokumente', {
|
||||
'name': filename,
|
||||
'dokumentId': attachment.get('id'),
|
||||
'hnr': history_entry.get('hNr') if history_entry else None,
|
||||
'advowareArt': (history_entry.get('art', 'Schreiben') or 'Schreiben')[:100] if history_entry else 'Schreiben',
|
||||
'advowareBemerkung': (history_entry.get('text', '') or '')[:255] if history_entry else '',
|
||||
'dateipfad': windows_file.get('path', ''),
|
||||
'blake3hash': blake3_hash,
|
||||
'syncedHash': blake3_hash,
|
||||
'usn': windows_file.get('usn', 0),
|
||||
'syncStatus': 'synced',
|
||||
'lastSyncTimestamp': now,
|
||||
'cAktenId': akte_id, # Direct FK to CAkten
|
||||
})
|
||||
doc_id = new_doc.get('id')
|
||||
|
||||
# Link to Akte
|
||||
await espocrm.link_entities('CAkten', akte_id, 'dokumentes', doc_id)
|
||||
results['created'] += 1
|
||||
|
||||
# Trigger preview
|
||||
try:
|
||||
await ctx.emit('document.generate_preview', {
|
||||
'entity_id': doc_id,
|
||||
'entity_type': 'CDokumente',
|
||||
})
|
||||
except Exception as e:
|
||||
ctx.logger.warn(f" ⚠️ Preview trigger failed: {e}")
|
||||
|
||||
elif action.action == 'UPDATE_ESPO':
|
||||
if not windows_file:
|
||||
ctx.logger.error(f" ❌ UPDATE_ESPO: no Windows file for hnr {hnr}")
|
||||
results['errors'] += 1
|
||||
continue
|
||||
|
||||
content = await watcher.download_file(aktennummer, windows_file.get('relative_path', filename))
|
||||
blake3_hash = compute_blake3(content)
|
||||
mime_type, _ = mimetypes.guess_type(filename)
|
||||
mime_type = mime_type or 'application/octet-stream'
|
||||
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
update_data: Dict[str, Any] = {
|
||||
'name': filename,
|
||||
'blake3hash': blake3_hash,
|
||||
'syncedHash': blake3_hash,
|
||||
'usn': windows_file.get('usn', 0),
|
||||
'dateipfad': windows_file.get('path', ''),
|
||||
'syncStatus': 'synced',
|
||||
'lastSyncTimestamp': now,
|
||||
}
|
||||
if history_entry:
|
||||
update_data['hnr'] = history_entry.get('hNr')
|
||||
update_data['advowareArt'] = (history_entry.get('art', 'Schreiben') or 'Schreiben')[:100]
|
||||
update_data['advowareBemerkung'] = (history_entry.get('text', '') or '')[:255]
|
||||
|
||||
await espocrm.update_entity('CDokumente', espo_doc['id'], update_data)
|
||||
results['updated'] += 1
|
||||
|
||||
# Mark for re-sync to xAI (hash changed)
|
||||
if espo_doc.get('aiSyncStatus') == 'synced':
|
||||
await espocrm.update_entity('CDokumente', espo_doc['id'], {
|
||||
'aiSyncStatus': 'unclean',
|
||||
})
|
||||
|
||||
try:
|
||||
await ctx.emit('document.generate_preview', {
|
||||
'entity_id': espo_doc['id'],
|
||||
'entity_type': 'CDokumente',
|
||||
})
|
||||
except Exception as e:
|
||||
ctx.logger.warn(f" ⚠️ Preview trigger failed: {e}")
|
||||
|
||||
elif action.action == 'DELETE':
|
||||
if espo_doc:
|
||||
# Only delete if the HNR is genuinely absent from Advoware History
|
||||
# (not just absent from Windows – avoids deleting docs whose file
|
||||
# is temporarily unavailable on the Windows share)
|
||||
if hnr in history_by_hnr:
|
||||
ctx.logger.warn(f" ⚠️ SKIP DELETE hnr={hnr}: still in Advoware History, only missing from Windows")
|
||||
results['skipped'] += 1
|
||||
else:
|
||||
await espocrm.delete_entity('CDokumente', espo_doc['id'])
|
||||
results['deleted'] += 1
|
||||
|
||||
except Exception as e:
|
||||
ctx.logger.error(f" ❌ Error for hnr {hnr} ({filename}): {e}")
|
||||
results['errors'] += 1
|
||||
|
||||
# ── Ablage check + Rubrum sync ─────────────────────────────────────
|
||||
try:
|
||||
akte_details = await advoware_service.get_akte(aktennummer)
|
||||
if akte_details:
|
||||
espo_update: Dict[str, Any] = {}
|
||||
|
||||
if akte_details.get('ablage') == 1:
|
||||
ctx.logger.info("📁 Akte marked as ablage → deactivating")
|
||||
espo_update['aktivierungsstatus'] = 'deaktiviert'
|
||||
|
||||
rubrum = akte_details.get('rubrum')
|
||||
if rubrum and rubrum != akte.get('rubrum'):
|
||||
espo_update['rubrum'] = rubrum
|
||||
ctx.logger.info(f"📝 Rubrum synced: {rubrum[:80]}")
|
||||
|
||||
if espo_update:
|
||||
await espocrm.update_entity('CAkten', akte_id, espo_update)
|
||||
except Exception as e:
|
||||
ctx.logger.warn(f"⚠️ Ablage/Rubrum check failed: {e}")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# xAI sync
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async def _run_xai_sync(
|
||||
akte: Dict[str, Any],
|
||||
akte_id: str,
|
||||
espocrm,
|
||||
ctx: FlowContext,
|
||||
) -> None:
|
||||
from services.xai_service import XAIService
|
||||
from services.xai_upload_utils import XAIUploadUtils
|
||||
|
||||
xai = XAIService(ctx)
|
||||
upload_utils = XAIUploadUtils(ctx)
|
||||
|
||||
ctx.logger.info("")
|
||||
ctx.logger.info("─" * 60)
|
||||
ctx.logger.info("🤖 xAI SYNC")
|
||||
ctx.logger.info("─" * 60)
|
||||
|
||||
try:
|
||||
# ── Ensure collection exists ───────────────────────────────────
|
||||
collection_id = await upload_utils.ensure_collection(akte, xai, espocrm)
|
||||
if not collection_id:
|
||||
ctx.logger.error("❌ Could not obtain xAI collection – aborting xAI sync")
|
||||
await espocrm.update_entity('CAkten', akte_id, {'aiSyncStatus': 'failed'})
|
||||
return
|
||||
|
||||
# ── Load all linked documents ──────────────────────────────────
|
||||
docs_result = await espocrm.list_related('CAkten', akte_id, 'dokumentes')
|
||||
docs = docs_result.get('list', [])
|
||||
ctx.logger.info(f" Documents to check: {len(docs)}")
|
||||
|
||||
synced = 0
|
||||
skipped = 0
|
||||
failed = 0
|
||||
|
||||
for doc in docs:
|
||||
ok = await upload_utils.sync_document_to_xai(doc, collection_id, xai, espocrm)
|
||||
if ok:
|
||||
if doc.get('aiSyncStatus') == 'synced' and doc.get('aiSyncHash') == doc.get('blake3hash'):
|
||||
skipped += 1
|
||||
else:
|
||||
synced += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
ctx.logger.info(f" ✅ Synced : {synced}")
|
||||
ctx.logger.info(f" ⏭️ Skipped : {skipped}")
|
||||
ctx.logger.info(f" ❌ Failed : {failed}")
|
||||
|
||||
finally:
|
||||
await xai.close()
|
||||
0
src/steps/crm/akte/webhooks/__init__.py
Normal file
0
src/steps/crm/akte/webhooks/__init__.py
Normal file
68
src/steps/crm/akte/webhooks/akte_webhook_step.py
Normal file
68
src/steps/crm/akte/webhooks/akte_webhook_step.py
Normal file
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Akte Sync - EspoCRM Webhook
|
||||
|
||||
Empfängt EspoCRM-Webhooks für CAkten (create / update / delete).
|
||||
Schreibt die Entity-ID in die Redis-Queue `akte:pending_entity_ids`
|
||||
mit 10-Sekunden-Debounce — der Cron-Poller übernimmt den Rest.
|
||||
|
||||
Route: POST /akte/webhook/update
|
||||
Payload: { "id": "..." } oder [{ "id": "..." }, ...]
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
from typing import Any
|
||||
from motia import FlowContext, http, ApiRequest, ApiResponse
|
||||
|
||||
|
||||
config = {
|
||||
"name": "Akte Webhook - EspoCRM",
|
||||
"description": "Empfängt EspoCRM-Webhooks für CAkten und queued Entity-IDs für den Sync",
|
||||
"flows": ["akte-sync"],
|
||||
"triggers": [http("POST", "/crm/akte/webhook/update")],
|
||||
"enqueues": [],
|
||||
}
|
||||
|
||||
PENDING_KEY = "akte:pending_entity_ids"
|
||||
|
||||
|
||||
async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
|
||||
try:
|
||||
payload = request.body or {}
|
||||
|
||||
ctx.logger.info("=" * 60)
|
||||
ctx.logger.info("📥 AKTE WEBHOOK")
|
||||
ctx.logger.info(f" Payload: {json.dumps(payload, ensure_ascii=False)[:200]}")
|
||||
|
||||
# ── Collect entity IDs ─────────────────────────────────────
|
||||
entity_ids: set[str] = set()
|
||||
if isinstance(payload, list):
|
||||
for item in payload:
|
||||
if isinstance(item, dict) and 'id' in item:
|
||||
entity_ids.add(item['id'])
|
||||
elif isinstance(payload, dict) and 'id' in payload:
|
||||
entity_ids.add(payload['id'])
|
||||
|
||||
if not entity_ids:
|
||||
ctx.logger.warn("⚠️ No entity IDs in payload")
|
||||
return ApiResponse(status_code=400, body={"error": "No entity ID found in payload"})
|
||||
|
||||
# ── Push to Redis with current timestamp (debounce in cron) ─
|
||||
from services.redis_client import get_redis_client
|
||||
redis_client = get_redis_client(strict=False)
|
||||
if not redis_client:
|
||||
ctx.logger.error("❌ Redis unavailable")
|
||||
return ApiResponse(status_code=503, body={"error": "Service unavailable"})
|
||||
|
||||
ts = time.time()
|
||||
mapping = {eid: ts for eid in entity_ids}
|
||||
redis_client.zadd(PENDING_KEY, mapping)
|
||||
|
||||
ctx.logger.info(f"✅ Queued {len(entity_ids)} entity ID(s): {entity_ids}")
|
||||
ctx.logger.info("=" * 60)
|
||||
|
||||
return ApiResponse(status_code=200, body={"queued": len(entity_ids)})
|
||||
|
||||
except Exception as e:
|
||||
ctx.logger.error(f"❌ Webhook error: {e}")
|
||||
return ApiResponse(status_code=500, body={"error": str(e)})
|
||||
Reference in New Issue
Block a user