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

feat: Enhance audit_calendar_sync tool with comprehensive management features

Browse Source 
- Add detailed documentation in README.md for all 10 audit commands
- Fix Advoware API parameter from 'frNr' to 'frnr' for proper filtering
- Fix subject field mapping from 'betreff' to 'text' in API responses
- Add verify-sync command for bidirectional sync verification
- Add query-frnr and query-event commands for individual record lookup
- Add management commands: find-duplicates, delete-duplicates, find-orphaned, cleanup-orphaned
- Improve error handling and output formatting
- Remove temporary test script
This commit is contained in:
root
2025-10-25 08:18:48 +00:00
parent c5ddd02307
commit c5600b42ec
5 changed files with 310 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
bitbylaw/steps/advoware_cal_sync/audit_calendar_sync.py
View File

@@ -513,12 +513,12 @@ async def query_frnr(frnr):
rows = await conn.fetch(
"""
SELECT sync_id, employee_kuerzel, advoware_frnr, google_event_id,
source_system, sync_strategy, sync_status, last_sync, created_at, updated_at
source_system, sync_strategy, sync_status, last_sync, created_at
FROM calendar_sync
WHERE advoware_frnr = $1
ORDER BY sync_id
""",
str(frnr)
int(frnr)
)
if not rows:
@@ -538,7 +538,7 @@ async def query_frnr(frnr):
print(f" Sync Status: {row['sync_status']}")
print(f" Last Sync: {row['last_sync']}")
print(f" Created: {row['created_at']}")
print(f" Updated: {row['updated_at']}")
print(f" Updated: {row['created_at']}") # Using created_at as updated_at doesn't exist
finally:
await conn.close()
@@ -551,7 +551,7 @@ async def query_event(event_id):
row = await conn.fetchrow(
"""
SELECT sync_id, employee_kuerzel, advoware_frnr, google_event_id,
source_system, sync_strategy, sync_status, last_sync, created_at, updated_at
source_system, sync_strategy, sync_status, last_sync, created_at
FROM calendar_sync
WHERE google_event_id = $1
""",
@@ -572,7 +572,7 @@ async def query_event(event_id):
print(f" Sync Status: {row['sync_status']}")
print(f" Last Sync: {row['last_sync']}")
print(f" Created: {row['created_at']}")
print(f" Updated: {row['updated_at']}")
print(f" Updated: {row['created_at']}") # Using created_at as updated_at doesn't exist
finally:
await conn.close()
@@ -589,7 +589,7 @@ async def verify_sync(frnr, service, advoware_api):
FROM calendar_sync
WHERE advoware_frnr = $1 AND deleted = FALSE
""",
str(frnr)
int(frnr)
)
if not row:
@@ -607,23 +607,45 @@ async def verify_sync(frnr, service, advoware_api):
# Check Advoware
print(f"\n--- Checking Advoware ---")
try:
# Use frNr with a broad date range to query the appointment
advoware_result = await advoware_api.api_call(
'api/v1/advonet/Termine',
method='GET',
params={
'kuerzel': employee_kuerzel,
'frNr': str(frnr)
'frnr': int(frnr), # Use lowercase 'frnr' as per API docs
'from': '2000-01-01T00:00:00',
'to': '2030-12-31T23:59:59'
}
)
if advoware_result and len(advoware_result) > 0:
appointment = advoware_result[0]
# API returns a list, find the specific appointment
target_appointment = None
if isinstance(advoware_result, list):
for appointment in advoware_result:
if str(appointment.get('frNr', '')) == str(frnr):
target_appointment = appointment
break
if target_appointment:
print("✅ Found in Advoware:")
print(f" Subject: {appointment.get('betreff', 'N/A')}")
print(f" Date: {appointment.get('datum', 'N/A')}")
print(f" Time: {appointment.get('uhrzeit', 'N/A')}")
print(f" Subject: {target_appointment.get('text', 'N/A')}")
print(f" Date: {target_appointment.get('datum', 'N/A')}")
print(f" Time: {target_appointment.get('uhrzeitVon', 'N/A')}")
print(f" End Time: {target_appointment.get('uhrzeitBis', 'N/A')}")
print(f" End Date: {target_appointment.get('datumBis', 'N/A')}")
print(f" Last Modified: {target_appointment.get('zuletztGeaendertAm', 'N/A')}")
print(f" frNr: {target_appointment.get('frNr', 'N/A')}")
advoware_exists = True
else:
print("❌ Not found in Advoware")
print(f"❌ Not found in Advoware (checked {len(advoware_result) if isinstance(advoware_result, list) else 0} appointments)")
# Show first few appointments for debugging (limited to 5)
if isinstance(advoware_result, list) and len(advoware_result) > 0:
print(" First few appointments returned:")
for i, app in enumerate(advoware_result[:5]):
print(f" [{i}] Subject: {app.get('text', 'N/A')}")
print(f" Date: {app.get('datum', 'N/A')}")
print(f" frNr: {app.get('frNr', 'N/A')}")
advoware_exists = False
except Exception as e:
print(f"❌ Error checking Advoware: {e}")