diff --git a/bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py b/bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py index e1978b2e..78926737 100644 --- a/bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py +++ b/bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py @@ -30,18 +30,28 @@ FETCH_TO = f"{current_year + 9}-12-31T23:59:59" # End of 9 years ahead CALENDAR_SYNC_LOCK_KEY = 'calendar_sync_lock' -def log_operation(level, message, **context): - """Centralized logging with context.""" - context_str = ' '.join(f"{k}={v}" for k, v in context.items() if v is not None) +def log_operation(level, message, context=None, **context_vars): + """Centralized logging with context, supporting Motia workbench logging.""" + context_str = ' '.join(f"{k}={v}" for k, v in context_vars.items() if v is not None) full_message = f"{message} {context_str}".strip() - if level == 'info': - logger.info(full_message) - elif level == 'warning': - logger.warning(full_message) - elif level == 'error': - logger.error(full_message) - elif level == 'debug': - logger.debug(full_message) + if context: + if level == 'info': + context.logger.info(full_message) + elif level == 'warning': + context.logger.warning(full_message) + elif level == 'error': + context.logger.error(full_message) + elif level == 'debug': + context.logger.debug(full_message) + else: + if level == 'info': + logger.info(full_message) + elif level == 'warning': + logger.warning(full_message) + elif level == 'error': + logger.error(full_message) + elif level == 'debug': + logger.debug(full_message) async def connect_db(): """Connect to Postgres DB from Config.""" @@ -571,9 +581,9 @@ def get_timestamps(adv_data, google_data): return adv_ts, google_ts -async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware): +async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware, context=None): """Phase 1: Process new appointments from Advoware to Google.""" - logger.info("Phase 1: Processing new appointments from Advoware") + log_operation('info', "Phase 1: Processing new appointments from Advoware", context=context) for frnr, app in state['adv_map'].items(): if frnr not in state['db_adv_index']: try: @@ -586,14 +596,14 @@ async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel, """, kuerzel, int(frnr), event_id ) - logger.info(f"Phase 1: Created new from Advoware: frNr {frnr}, event_id {event_id}") + log_operation('info', f"Phase 1: Created new from Advoware: frNr {frnr}, event_id {event_id}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 1: Failed to process new Advoware {frnr}: {e}") + log_operation('warning', f"Phase 1: Failed to process new Advoware {frnr}: {e}", context=context) -async def process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware): +async def process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware, context=None): """Phase 2: Process new events from Google to Advoware.""" - logger.info("Phase 2: Processing new events from Google") + log_operation('info', "Phase 2: Processing new events from Google", context=context) for event_id, evt in state['google_map'].items(): # For recurring events, check if the master event (recurringEventId) is already synced # For regular events, check the event_id directly @@ -612,15 +622,15 @@ async def process_new_from_google(state, conn, service, calendar_id, kuerzel, ad """, kuerzel, int(frnr), event_id ) - logger.info(f"Phase 2: Created new from Google: event_id {event_id}, frNr {frnr}") + log_operation('info', f"Phase 2: Created new from Google: event_id {event_id}, frNr {frnr}", context=context) else: - logger.warning(f"Phase 2: Skipped DB insert for Google event {event_id}, frNr is None") + log_operation('warning', f"Phase 2: Skipped DB insert for Google event {event_id}, frNr is None", context=context) except Exception as e: - logger.warning(f"Phase 2: Failed to process new Google {event_id}: {e}") + log_operation('warning', f"Phase 2: Failed to process new Google {event_id}: {e}", context=context) -async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware): +async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware, context=None): """Phase 3: Process deleted entries.""" - logger.info("Phase 3: Processing deleted entries") + log_operation('info', "Phase 3: Processing deleted entries", context=context) for row in state['rows']: frnr = row['advoware_frnr'] event_id = row['google_event_id'] @@ -643,7 +653,7 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad # Both missing - soft delete async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Soft deleted sync_id {row['sync_id']} (both missing)") + log_operation('info', f"Phase 3: Soft deleted sync_id {row['sync_id']} (both missing)", context=context) elif not adv_exists: # Missing in Advoware strategy = row['sync_strategy'] @@ -654,10 +664,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad await delete_google_event(service, calendar_id, event_id) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) elif row['source_system'] == 'google' and row['advoware_write_allowed']: @@ -667,13 +677,13 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad if new_frnr and str(new_frnr) != 'None': async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET advoware_frnr = $1, sync_status = 'synced', last_sync = $3 WHERE sync_id = $2;", int(new_frnr), row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 3: Recreated Advoware appointment {new_frnr} for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Recreated Advoware appointment {new_frnr} for sync_id {row['sync_id']}", context=context) else: - logger.warning(f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}, frNr is None") + log_operation('warning', f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}, frNr is None", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) except Exception as e: - logger.warning(f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) else: @@ -682,10 +692,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad await delete_google_event(service, calendar_id, event_id) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) else: @@ -694,10 +704,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad await delete_google_event(service, calendar_id, event_id) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) elif not google_exists: @@ -711,13 +721,13 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad await safe_delete_advoware_appointment(advoware, frnr, row['advoware_write_allowed']) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}", context=context) except Exception as e: - logger.warning(f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) else: - logger.warning(f"Phase 3: Cannot delete in Advoware for sync_id {row['sync_id']}, write not allowed") + log_operation('warning', f"Phase 3: Cannot delete in Advoware for sync_id {row['sync_id']}, write not allowed", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) elif row['source_system'] == 'advoware': @@ -726,10 +736,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad new_event_id = await create_google_event(service, calendar_id, standardize_appointment_data(state['adv_map'][str(frnr)], 'advoware')) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET google_event_id = $1, sync_status = 'synced', last_sync = $3 WHERE sync_id = $2;", new_event_id, row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 3: Recreated Google event {new_event_id} for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Recreated Google event {new_event_id} for sync_id {row['sync_id']}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 3: Failed to recreate Google for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to recreate Google for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) else: @@ -738,15 +748,15 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad await safe_delete_advoware_appointment(advoware, frnr, row['advoware_write_allowed']) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id']) - logger.info(f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}") + log_operation('info', f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}", context=context) except Exception as e: - logger.warning(f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) -async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): +async def process_updates(state, conn, service, calendar_id, kuerzel, advoware, context=None): """Phase 4: Process updates for existing entries.""" - logger.info("Phase 4: Processing updates for existing entries") + log_operation('info', "Phase 4: Processing updates for existing entries", context=context) # Track which master events we've already processed to avoid duplicate updates processed_master_events = set() @@ -793,32 +803,32 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): await update_google_event(service, calendar_id, event_id, adv_std) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 4: Updated Google event {event_id} from Advoware frNr {frnr}") + log_operation('info', f"Phase 4: Updated Google event {event_id} from Advoware frNr {frnr}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits elif google_ts and google_ts > row['last_sync']: - logger.warning(f"Phase 4: Unauthorized change in Google event {event_id}, resetting to Advoware frNr {frnr}") + log_operation('warning', f"Phase 4: Unauthorized change in Google event {event_id}, resetting to Advoware frNr {frnr}", context=context) await update_google_event(service, calendar_id, event_id, adv_std) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 4: Reset Google event {event_id} to Advoware frNr {frnr}") + log_operation('info', f"Phase 4: Reset Google event {event_id} to Advoware frNr {frnr}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits elif row['source_system'] == 'google' and row['advoware_write_allowed']: # Check for changes in source (Google) or unauthorized changes in target (Advoware) google_ts_str = google_data.get('updated', '') google_ts = datetime.datetime.fromisoformat(google_ts_str.rstrip('Z')).astimezone(BERLIN_TZ) if google_ts_str else None adv_ts = BERLIN_TZ.localize(datetime.datetime.fromisoformat(adv_data['zuletztGeaendertAm'])) - logger.debug(f"Phase 4: Checking sync_id {row['sync_id']}: adv_ts={adv_ts}, google_ts={google_ts}, last_sync={row['last_sync']}") + log_operation('debug', f"Phase 4: Checking sync_id {row['sync_id']}: adv_ts={adv_ts}, google_ts={google_ts}, last_sync={row['last_sync']}", context=context) if google_ts and google_ts > row['last_sync']: await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel']) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 4: Updated Advoware frNr {frnr} from Google event {event_id}") + log_operation('info', f"Phase 4: Updated Advoware frNr {frnr} from Google event {event_id}", context=context) elif adv_ts > row['last_sync']: - logger.warning(f"Phase 4: Unauthorized change in Advoware frNr {frnr}, resetting to Google event {event_id}") + log_operation('warning', f"Phase 4: Unauthorized change in Advoware frNr {frnr}, resetting to Google event {event_id}", context=context) await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel']) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ)) - logger.info(f"Phase 4: Reset Advoware frNr {frnr} to Google event {event_id}") + log_operation('info', f"Phase 4: Reset Advoware frNr {frnr} to Google event {event_id}", context=context) elif strategy == 'last_change_wins': adv_ts = await get_advoware_timestamp(advoware, frnr) google_ts_str = google_data.get('updated', '') @@ -830,10 +840,10 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel']) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], max(adv_ts, google_ts)) - logger.info(f"Phase 4: Updated based on last_change_wins for sync_id {row['sync_id']}") + log_operation('info', f"Phase 4: Updated based on last_change_wins for sync_id {row['sync_id']}", context=context) await asyncio.sleep(0.1) # Small delay to avoid rate limits except Exception as e: - logger.warning(f"Phase 4: Failed to update sync_id {row['sync_id']}: {e}") + log_operation('warning', f"Phase 4: Failed to update sync_id {row['sync_id']}: {e}", context=context) async with conn.transaction(): await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id']) """Main event handler for calendar sync.""" @@ -904,7 +914,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): for row in state['rows']: if row['google_event_id']: state['db_google_index'][row['google_event_id']] = row - log_operation('debug', "Reloaded indexes", rows=len(state['rows']), adv=len(state['db_adv_index']), google=len(state['db_google_index'])) + log_operation('debug', "Reloaded indexes", context=context, rows=len(state['rows']), adv=len(state['db_adv_index']), google=len(state['db_google_index'])) async def reload_api_maps(): """Reload API maps after creating new events in phases.""" @@ -912,16 +922,16 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): state['adv_map'] = {str(app['frNr']): app for app in state['adv_appointments'] if app.get('frNr')} state['google_events'] = await fetch_google_events(service, calendar_id) state['google_map'] = {evt['id']: evt for evt in state['google_events']} - log_operation('debug', "Reloaded API maps", adv=len(state['adv_map']), google=len(state['google_map'])) + log_operation('debug', "Reloaded API maps", context=context, adv=len(state['adv_map']), google=len(state['google_map'])) # Initial fetch - log_operation('info', "Fetching fresh data from APIs") + log_operation('info', "Fetching fresh data from APIs", context=context) await reload_api_maps() await reload_db_indexes() - log_operation('info', "Fetched existing sync rows", count=len(state['rows'])) + log_operation('info', "Fetched existing sync rows", context=context, count=len(state['rows'])) # Phase 1: New from Advoware => Google - await process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware) + await process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware, context) # Reload indexes after Phase 1 changes await reload_db_indexes() @@ -929,7 +939,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): await reload_api_maps() # Phase 2: New from Google => Advoware - await process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware) + await process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware, context) # Reload indexes after Phase 2 changes await reload_db_indexes() @@ -937,7 +947,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): await reload_api_maps() # Phase 3: Identify deleted entries - await process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware) + await process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware, context) # Reload indexes after Phase 3 changes await reload_db_indexes() @@ -945,7 +955,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware): await reload_api_maps() # Phase 4: Update existing entries if changed - await process_updates(state, conn, service, calendar_id, kuerzel, advoware) + await process_updates(state, conn, service, calendar_id, kuerzel, advoware, context) # Update last_sync timestamps logger.debug("Updated last_sync timestamps")