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

Fix timestamp-based update logic and remove global last_sync update to prevent redundant syncs and race conditions. Update README with anonymization and latest changes.

Browse Source
This commit is contained in:
root
2025-10-23 10:40:00 +00:00
parent 1de5bcd369
commit 9ab90fef5a
9 changed files with 910 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files

430
bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py
View File

@@ -8,12 +8,13 @@ from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.oauth2 import service_account
import asyncpg
import redis
from config import Config # Assuming Config has POSTGRES_HOST='localhost', USER, PASSWORD, DB_NAME, GOOGLE_CALENDAR_SERVICE_ACCOUNT_PATH, GOOGLE_CALENDAR_SCOPES, etc.
from services.advoware import AdvowareAPI # Assuming this is the existing wrapper for Advoware API calls
# Setup logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
logger.addHandler(handler)
@@ -24,6 +25,8 @@ BERLIN_TZ = pytz.timezone('Europe/Berlin')
FETCH_FROM = (datetime.datetime.now(BERLIN_TZ) - timedelta(days=365)).strftime('%Y-01-01T00:00:00')
FETCH_TO = (datetime.datetime.now(BERLIN_TZ) + timedelta(days=730)).strftime('%Y-12-31T23:59:59')
CALENDAR_SYNC_LOCK_KEY = 'calendar_sync_lock'
# Relevant fields for data comparison (simple diff)
COMPARISON_FIELDS = ['text', 'notiz', 'ort', 'datum', 'uhrzeitBis', 'datumBis', 'dauertermin', 'turnus', 'turnusArt']
@@ -95,6 +98,7 @@ async def fetch_advoware_appointments(advoware, employee_kuerzel):
'to': FETCH_TO
}
result = await advoware.api_call('api/v1/advonet/Termine', method='GET', params=params)
logger.debug(f"Raw Advoware API response: {result}")
appointments = result if isinstance(result, list) else []
logger.info(f"Fetched {len(appointments)} Advoware appointments for {employee_kuerzel}")
return appointments
@@ -108,13 +112,16 @@ async def fetch_google_events(service, calendar_id):
now = datetime.datetime.now(pytz.utc)
from_date = now - timedelta(days=365)
to_date = now + timedelta(days=730)
time_min = from_date.strftime('%Y-%m-%dT%H:%M:%SZ')
time_max = to_date.strftime('%Y-%m-%dT%H:%M:%SZ')
events_result = service.events().list(
calendarId=calendar_id,
timeMin=from_date.isoformat() + 'Z',
timeMax=to_date.isoformat() + 'Z',
timeMin=time_min,
timeMax=time_max,
singleEvents=True, # Expand recurring
orderBy='startTime'
).execute()
logger.debug(f"Raw Google API response: {events_result}")
events = [evt for evt in events_result.get('items', []) if evt.get('status') != 'cancelled']
logger.info(f"Fetched {len(events)} Google events for calendar {calendar_id}")
return events
@@ -129,17 +136,46 @@ def standardize_appointment_data(data, source):
"""Standardize data from Advoware or Google to comparable dict, with TZ handling."""
if source == 'advoware':
start_str = data.get('datum', '')
end_str = data.get('datumBis', data.get('datum', ''))
start_time = data.get('uhrzeitVon') or '09:00:00' if 'T' not in start_str else start_str.split('T')[1]
# Improved parsing: if datum contains 'T', it's datetime; else combine with uhrzeitVon
if 'T' in start_str:
try:
start_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(start_str.replace('Z', '')))
except ValueError:
logger.warning(f"Invalid start datetime in Advoware: {start_str}")
start_dt = BERLIN_TZ.localize(datetime.datetime.now())
else:
start_time = data.get('uhrzeitVon') or '09:00:00'
start_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(f"{start_str}T{start_time}"))
# For end: Use date from datumBis (or datum), time from uhrzeitBis
end_date_str = data.get('datumBis', data.get('datum', ''))
if 'T' in end_date_str:
base_end_date = end_date_str.split('T')[0]
else:
base_end_date = end_date_str
end_time = data.get('uhrzeitBis', '10:00:00')
start_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(start_str.replace('Z', '')))
end_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(end_str.replace('Z', '')))
try:
end_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(f"{base_end_date}T{end_time}"))
except ValueError:
logger.warning(f"Invalid end datetime in Advoware: {base_end_date}T{end_time}")
end_dt = start_dt + timedelta(hours=1)
# Anonymization for Google events
if Config.CALENDAR_SYNC_ANONYMIZE_GOOGLE_EVENTS:
text = 'Advoware blocked'
notiz = ''
ort = ''
else:
text = data.get('text', '')
notiz = data.get('notiz', '')
ort = data.get('ort', '')
return {
'start': start_dt,
'end': end_dt,
'text': data.get('text', ''),
'notiz': data.get('notiz', ''),
'ort': data.get('ort', ''),
'text': text,
'notiz': notiz,
'ort': ort,
'dauertermin': data.get('dauertermin', 0),
'turnus': data.get('turnus', 0),
'turnusArt': data.get('turnusArt', 0),
@@ -158,7 +194,17 @@ def standardize_appointment_data(data, source):
end_dt = datetime.datetime.fromisoformat(end_obj['dateTime'].rstrip('Z')).astimezone(BERLIN_TZ)
else:
end_dt = BERLIN_TZ.localize(datetime.datetime.fromisoformat(end_obj['date']))
dauertermin = 1 if all_day or (end_dt - start_dt).days > 1 else 0
# Improved dauertermin: set to 1 if all-day or duration >1 day
duration_days = (end_dt.date() - start_dt.date()).days
dauertermin = 1 if all_day or duration_days > 1 else 0
recurrence = data.get('recurrence')
if recurrence:
# Simple mapping: if recurrence exists, set turnus=1, turnusArt based on RRULE (simplified)
turnus = 1
turnus_art = 0 # Default, could parse RRULE for better mapping
else:
turnus = 0
turnus_art = 0
return {
'start': start_dt,
'end': end_dt,
@@ -166,9 +212,9 @@ def standardize_appointment_data(data, source):
'notiz': data.get('description', ''),
'ort': data.get('location', ''),
'dauertermin': dauertermin,
'turnus': 1 if data.get('recurrence') else 0, # Simplified
'turnusArt': 0, # Map RRULE to type if needed
'recurrence': data.get('recurrence')
'turnus': turnus,
'turnusArt': turnus_art,
'recurrence': recurrence
}
def data_diff(data1, data2):
@@ -200,7 +246,8 @@ async def create_advoware_appointment(advoware, data, employee_kuerzel):
}
try:
result = await advoware.api_call('api/v1/advonet/Termine', method='POST', json_data=appointment_data)
frnr = str(result.get('frNr'))
logger.debug(f"Raw Advoware POST response: {result}")
frnr = str(result.get('frNr') or result.get('frnr'))
logger.info(f"Created Advoware appointment frNr: {frnr}")
return frnr
except Exception as e:
@@ -246,7 +293,9 @@ async def create_google_event(service, calendar_id, data):
all_day = data['dauertermin'] == 1 and start_dt.time() == datetime.time(0,0) and end_dt.time() == datetime.time(0,0)
if all_day:
start_obj = {'date': start_dt.strftime('%Y-%m-%d')}
end_obj = {'date': end_dt.strftime('%Y-%m-%d')}
# For all-day events, end date is exclusive, so add 1 day
end_date = (start_dt + timedelta(days=1)).strftime('%Y-%m-%d')
end_obj = {'date': end_date}
else:
start_obj = {'dateTime': start_dt.isoformat(), 'timeZone': 'Europe/Berlin'}
end_obj = {'dateTime': end_dt.isoformat(), 'timeZone': 'Europe/Berlin'}
@@ -277,7 +326,9 @@ async def update_google_event(service, calendar_id, event_id, data):
all_day = data['dauertermin'] == 1 and start_dt.time() == datetime.time(0,0) and end_dt.time() == datetime.time(0,0)
if all_day:
start_obj = {'date': start_dt.strftime('%Y-%m-%d')}
end_obj = {'date': end_dt.strftime('%Y-%m-%d')}
# For all-day events, end date is exclusive, so add 1 day
end_date = (start_dt + timedelta(days=1)).strftime('%Y-%m-%d')
end_obj = {'date': end_date}
else:
start_obj = {'dateTime': start_dt.isoformat(), 'timeZone': 'Europe/Berlin'}
end_obj = {'dateTime': end_dt.isoformat(), 'timeZone': 'Europe/Berlin'}
@@ -311,167 +362,242 @@ async def delete_google_event(service, calendar_id, event_id):
logger.error(f"Failed to delete Google event {event_id}: {e}")
raise
async def get_advoware_timestamp(advoware, frnr):
"""Fetch last modified timestamp from Advoware (zuletztGeaendertAm)."""
try:
result = await advoware.api_call('api/v1/advonet/Termine', method='GET', params={'frnr': frnr})
if result and isinstance(result, list) and result:
return datetime.datetime.fromisoformat(result[0].get('zuletztGeaendertAm', '').rstrip('Z')).astimezone(BERLIN_TZ)
return None
except Exception as e:
logger.error(f"Failed to get Advoware timestamp for {frnr}: {e}")
async def safe_create_advoware_appointment(advoware, data, employee_kuerzel, write_allowed):
"""Safe wrapper for creating Advoware appointments with write permission check."""
if not write_allowed:
logger.warning("Cannot create in Advoware, write not allowed")
return None
return await create_advoware_appointment(advoware, data, employee_kuerzel)
async def safe_update_advoware_appointment(advoware, frnr, data, write_allowed):
"""Safe wrapper for updating Advoware appointments with write permission check."""
if not write_allowed:
logger.warning("Cannot update in Advoware, write not allowed")
return
await update_advoware_appointment(advoware, frnr, data)
async def safe_delete_advoware_appointment(advoware, frnr, write_allowed):
"""Safe wrapper for deleting Advoware appointments with write permission check."""
if not write_allowed:
logger.warning("Cannot delete in Advoware, write not allowed")
return
await delete_advoware_appointment(advoware, frnr)
async def handler(event, context):
"""Main event handler for calendar sync."""
employee_kuerzel = event.get('data', {}).get('body', {}).get('employee_kuerzel', 'AI') # Default to 'AI' for test
logger.info(f"Starting calendar sync for {employee_kuerzel}")
redis_client = redis.Redis(
host=Config.REDIS_HOST,
port=int(Config.REDIS_PORT),
db=int(Config.REDIS_DB_CALENDAR_SYNC),
socket_timeout=Config.REDIS_TIMEOUT_SECONDS
)
try:
logger.debug("Initializing Google service")
service = await get_google_service()
logger.debug(f"Ensuring Google calendar for {employee_kuerzel}")
calendar_id = await ensure_google_calendar(service, employee_kuerzel)
logger.debug("Initializing Advoware API")
advoware = AdvowareAPI(context)
async with await connect_db() as conn:
async with conn.transaction():
# Lock rows
rows = await conn.fetch(
"""
SELECT * FROM calendar_sync
WHERE employee_kuerzel = $1 AND deleted = FALSE
FOR UPDATE
""",
employee_kuerzel
)
conn = await connect_db()
try:
# Fetch fresh data
logger.info("Fetching fresh data from APIs")
adv_appointments = await fetch_advoware_appointments(advoware, employee_kuerzel)
adv_map = {str(app['frNr']): app for app in adv_appointments if app.get('frNr')}
google_events = await fetch_google_events(service, calendar_id)
google_map = {evt['id']: evt for evt in google_events}
# Build maps
adv_appointments = await fetch_advoware_appointments(advoware, employee_kuerzel)
adv_map = {str(app['frNr']): app for app in adv_appointments if app.get('frNr')}
google_events = await fetch_google_events(service, calendar_id)
google_map = {evt['id']: evt for evt in google_events}
# Fetch existing DB rows
rows = await conn.fetch(
"""
SELECT * FROM calendar_sync
WHERE employee_kuerzel = $1 AND deleted = FALSE
""",
employee_kuerzel
)
logger.info(f"Fetched {len(rows)} existing sync rows")
adv_map_std = {frnr: standardize_appointment_data(app, 'advoware') for frnr, app in adv_map.items()}
google_map_std = {eid: standardize_appointment_data(evt, 'google') for eid, evt in google_map.items()}
# Build indexes
db_adv_index = {str(row['advoware_frnr']): row for row in rows if row['advoware_frnr']}
db_google_index = {row['google_event_id']: row for row in rows if row['google_event_id']}
# Build index from DB rows
db_adv_index = {row['advoware_frnr']: row for row in rows if row['advoware_frnr']}
db_google_index = {row['google_event_id']: row for row in rows if row['google_event_id']}
# Process existing
for row in rows:
frnr = row['advoware_frnr']
event_id = row['google_event_id']
adv_data = adv_map.pop(frnr, None) if frnr else None
google_data = google_map.pop(event_id, None) if event_id else None
adv_std = adv_map_std.pop(frnr, None) if frnr else None
google_std = google_map_std.pop(event_id, None) if event_id else None
if not adv_data and not google_data:
# Both missing - soft delete
await conn.execute("UPDATE calendar_sync SET deleted = TRUE WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Marked as deleted: sync_id {row['sync_id']}")
continue
if adv_data and google_data:
# Both exist - check diff
if data_diff(adv_std, google_std):
strategy = row['sync_strategy']
if strategy == 'source_system_wins':
if row['source_system'] == 'advoware':
await update_google_event(service, calendar_id, event_id, adv_std)
elif row['source_system'] == 'google' and row['advoware_write_allowed']:
await update_advoware_appointment(advoware, frnr, google_std)
else:
logger.warning(f"Write to Advoware blocked for sync_id {row['sync_id']}")
elif strategy == 'last_change_wins':
adv_ts = await get_advoware_timestamp(advoware, frnr)
google_ts = datetime.datetime.fromisoformat(google_data.get('updated', '').rstrip('Z')).astimezone(BERLIN_TZ)
if adv_ts and google_ts:
if adv_ts > google_ts:
await update_google_event(service, calendar_id, event_id, adv_std)
elif row['advoware_write_allowed']:
await update_advoware_appointment(advoware, frnr, google_std)
else:
logger.warning(f"Missing timestamps for last_change_wins: sync_id {row['sync_id']}")
elif adv_data:
# Missing in Google - recreate or delete?
strategy = row['sync_strategy']
if strategy == 'source_system_wins' and row['source_system'] == 'advoware':
new_event_id = await create_google_event(service, calendar_id, adv_std)
# Phase 1: New from Advoware => Google
logger.info("Phase 1: Processing new appointments from Advoware")
for frnr, app in adv_map.items():
if frnr not in db_adv_index:
try:
event_id = await create_google_event(service, calendar_id, standardize_appointment_data(app, 'advoware'))
async with conn.transaction():
await conn.execute(
"UPDATE calendar_sync SET google_event_id = $1 WHERE sync_id = $2;",
new_event_id, row['sync_id']
)
elif strategy == 'last_change_wins':
# Assume Adv change newer - recreate
new_event_id = await create_google_event(service, calendar_id, adv_std)
await conn.execute(
"UPDATE calendar_sync SET google_event_id = $1 WHERE sync_id = $2;",
new_event_id, row['sync_id']
"""
INSERT INTO calendar_sync (employee_kuerzel, advoware_frnr, google_event_id, source_system, sync_strategy, sync_status, advoware_write_allowed)
VALUES ($1, $2, $3, 'advoware', 'source_system_wins', 'synced', FALSE);
""",
employee_kuerzel, int(frnr), event_id
)
logger.info(f"Phase 1: Created new from Advoware: frNr {frnr}, event_id {event_id}")
except Exception as e:
logger.warning(f"Phase 1: Failed to process new Advoware {frnr}: {e}")
# Phase 2: New from Google => Advoware
logger.info("Phase 2: Processing new events from Google")
for event_id, evt in google_map.items():
if event_id not in db_google_index:
try:
frnr = await safe_create_advoware_appointment(advoware, standardize_appointment_data(evt, 'google'), employee_kuerzel, True)
if frnr and str(frnr) != 'None':
async with conn.transaction():
await conn.execute(
"""
INSERT INTO calendar_sync (employee_kuerzel, advoware_frnr, google_event_id, source_system, sync_strategy, sync_status, advoware_write_allowed)
VALUES ($1, $2, $3, 'google', 'source_system_wins', 'synced', TRUE);
""",
employee_kuerzel, int(frnr), event_id
)
logger.info(f"Phase 2: Created new from Google: event_id {event_id}, frNr {frnr}")
else:
# Propagate delete to Advoware if allowed
if row['advoware_write_allowed']:
await delete_advoware_appointment(advoware, frnr)
await conn.execute("UPDATE calendar_sync SET deleted = TRUE WHERE sync_id = $1;", row['sync_id'])
logger.warning(f"Phase 2: Skipped DB insert for Google event {event_id}, frNr is None")
except Exception as e:
logger.warning(f"Phase 2: Failed to process new Google {event_id}: {e}")
elif google_data:
# Missing in Advoware - recreate or delete?
strategy = row['sync_strategy']
if strategy == 'source_system_wins' and row['source_system'] == 'google' and row['advoware_write_allowed']:
new_frnr = await create_advoware_appointment(advoware, google_std, employee_kuerzel)
await conn.execute(
"UPDATE calendar_sync SET advoware_frnr = $1 WHERE sync_id = $2;",
int(new_frnr), row['sync_id']
)
elif strategy == 'last_change_wins' and row['advoware_write_allowed']:
# Assume Google change newer - recreate
new_frnr = await create_advoware_appointment(advoware, google_std, employee_kuerzel)
await conn.execute(
"UPDATE calendar_sync SET advoware_frnr = $1 WHERE sync_id = $2;",
int(new_frnr), row['sync_id']
)
else:
# Propagate delete to Google
# Phase 3: Identify deleted entries
logger.info("Phase 3: Processing deleted entries")
for row in rows:
frnr = row['advoware_frnr']
event_id = row['google_event_id']
adv_exists = str(frnr) in adv_map if frnr else False
google_exists = event_id in google_map if event_id else False
if not adv_exists and not google_exists:
# 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)")
elif not adv_exists:
# Missing in Advoware
strategy = row['sync_strategy']
if strategy == 'source_system_wins' and row['source_system'] == 'advoware':
# Recreate in Google
try:
new_event_id = await create_google_event(service, calendar_id, standardize_appointment_data(google_map[event_id], 'google'))
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET google_event_id = $1, sync_status = 'synced' WHERE sync_id = $2;", new_event_id, row['sync_id'])
logger.info(f"Phase 3: Recreated Google event {new_event_id} for sync_id {row['sync_id']}")
except Exception as e:
logger.warning(f"Phase 3: Failed to recreate Google for sync_id {row['sync_id']}: {e}")
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
# Propagate delete to Google
try:
await delete_google_event(service, calendar_id, event_id)
await conn.execute("UPDATE calendar_sync SET deleted = TRUE WHERE sync_id = $1;", row['sync_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']}")
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}")
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:
# Missing in Google
strategy = row['sync_strategy']
if strategy == 'source_system_wins' and row['source_system'] == 'google' and row['advoware_write_allowed']:
# Recreate in Advoware
try:
new_frnr = await safe_create_advoware_appointment(advoware, standardize_appointment_data(adv_map[frnr], 'advoware'), employee_kuerzel, row['advoware_write_allowed'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET advoware_frnr = $1, sync_status = 'synced' WHERE sync_id = $2;", int(new_frnr), row['sync_id'])
logger.info(f"Phase 3: Recreated Advoware appointment {new_frnr} for sync_id {row['sync_id']}")
except Exception as e:
logger.warning(f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}: {e}")
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
# Propagate delete to Advoware
try:
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']}")
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}")
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
# New from Advoware
for frnr, app in adv_map.items():
adv_std = standardize_appointment_data(app, 'advoware')
event_id = await create_google_event(service, calendar_id, adv_std)
await conn.execute(
"""
INSERT INTO calendar_sync (employee_kuerzel, advoware_frnr, google_event_id, source_system, sync_strategy, advoware_write_allowed)
VALUES ($1, $2, $3, 'advoware', 'source_system_wins', FALSE);
""",
employee_kuerzel, int(frnr), event_id
)
logger.info(f"Created new from Advoware: frNr {frnr}, event_id {event_id}")
# Phase 4: Update existing entries if changed
logger.info("Phase 4: Processing updates for existing entries")
for row in rows:
frnr = row['advoware_frnr']
event_id = row['google_event_id']
adv_data = adv_map.get(str(frnr)) if frnr else None
google_data = google_map.get(event_id) if event_id else None
# New from Google
for event_id, evt in google_map.items():
google_std = standardize_appointment_data(evt, 'google')
frnr = await create_advoware_appointment(advoware, google_std, employee_kuerzel)
await conn.execute(
"""
INSERT INTO calendar_sync (employee_kuerzel, advoware_frnr, google_event_id, source_system, sync_strategy, advoware_write_allowed)
VALUES ($1, $2, $3, 'google', 'source_system_wins', TRUE);
""",
employee_kuerzel, int(frnr), event_id
)
logger.info(f"Created new from Google: event_id {event_id}, frNr {frnr}")
if adv_data and google_data:
adv_std = standardize_appointment_data(adv_data, 'advoware')
google_std = standardize_appointment_data(google_data, 'google')
strategy = row['sync_strategy']
try:
if strategy == 'source_system_wins':
if row['source_system'] == 'advoware':
# Check if Advoware was modified since last sync
adv_ts = datetime.datetime.fromisoformat(adv_data['zuletztGeaendertAm']).astimezone(BERLIN_TZ)
if adv_ts > row['last_sync']:
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'], adv_ts)
logger.info(f"Phase 4: Updated Google event {event_id} from Advoware frNr {frnr}")
elif row['source_system'] == 'google' and row['advoware_write_allowed']:
# Check if Google was modified since last sync
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
if google_ts and google_ts > row['last_sync']:
await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], google_ts)
logger.info(f"Phase 4: Updated Advoware frNr {frnr} from Google event {event_id}")
elif strategy == 'last_change_wins':
adv_ts = await get_advoware_timestamp(advoware, frnr)
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
if adv_ts and google_ts:
if adv_ts > google_ts:
await update_google_event(service, calendar_id, event_id, adv_std)
elif row['advoware_write_allowed']:
await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'])
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']}")
except Exception as e:
logger.warning(f"Phase 4: Failed to update sync_id {row['sync_id']}: {e}")
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
# Update last_sync
await conn.execute(
"UPDATE calendar_sync SET last_sync = NOW() WHERE employee_kuerzel = $1;",
employee_kuerzel
)
# Update last_sync timestamps
logger.debug("Updated last_sync timestamps")
finally:
await conn.close()
redis_client.delete(CALENDAR_SYNC_LOCK_KEY)
logger.info(f"Calendar sync completed for {employee_kuerzel}")
return {'status': 200, 'body': {'status': 'completed'}}
except Exception as e:
logger.error(f"Sync failed for {employee_kuerzel}: {e}")
return {'status': 500, 'body': {'error': str(e)}}
redis_client.delete(CALENDAR_SYNC_LOCK_KEY)
return {'status': 500, 'body': {'error': str(e)}}
# Motia Step Configuration
config = {
"type": "event",
"name": "Calendar Sync Event Step",
"description": "Handles bidirectional calendar sync between Advoware and Google Calendar using Postgres as hub",
"subscribes": ["calendar.sync.triggered"],
"emits": [],
"flows": ["advoware"]
}