[2ff88f42] feat(connector-superoffice): Implement Company Explorer sync and Holiday logic
- **Company Explorer Sync**: Added `explorer_client.py` and integrated Step 9 in `main.py` for automated data transfer to the intelligence engine. - **Holiday Logic**: Implemented `BusinessCalendar` in `utils.py` using the `holidays` library to automatically detect weekends and Bavarian holidays, ensuring professional timing for automated outreaches. - **API Discovery**: Created `parse_ce_openapi.py` to facilitate technical field mapping through live OpenAPI analysis. - **Project Stability**: Refined error handling and logging for a smooth end-to-end workflow.
This commit is contained in:
78
connector-superoffice/utils.py
Normal file
78
connector-superoffice/utils.py
Normal file
@@ -0,0 +1,78 @@
|
||||
import holidays
|
||||
from datetime import date, timedelta, datetime
|
||||
|
||||
class BusinessCalendar:
|
||||
"""
|
||||
Handles business day calculations, considering weekends and holidays
|
||||
(specifically for Bavaria/Germany).
|
||||
"""
|
||||
def __init__(self, country='DE', state='BY'):
|
||||
# Initialize holidays for Germany, Bavaria
|
||||
self.holidays = holidays.country_holidays(country, subdiv=state)
|
||||
|
||||
def is_business_day(self, check_date: date) -> bool:
|
||||
"""
|
||||
Checks if a given date is a business day (Mon-Fri) and not a holiday.
|
||||
"""
|
||||
# Check for weekend (Saturday=5, Sunday=6)
|
||||
if check_date.weekday() >= 5:
|
||||
return False
|
||||
|
||||
# Check for holiday
|
||||
if check_date in self.holidays:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def get_next_business_day(self, start_date: date) -> date:
|
||||
"""
|
||||
Returns the next valid business day starting from (and including) start_date.
|
||||
If start_date is a business day, it is returned.
|
||||
Otherwise, it searches forward.
|
||||
"""
|
||||
current_date = start_date
|
||||
# Safety limit to prevent infinite loops in case of misconfiguration
|
||||
# (though 365 days of holidays is unlikely)
|
||||
for _ in range(365):
|
||||
if self.is_business_day(current_date):
|
||||
return current_date
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
return current_date
|
||||
|
||||
def get_next_send_time(self, scheduled_time: datetime) -> datetime:
|
||||
"""
|
||||
Calculates the next valid timestamp for sending emails.
|
||||
If scheduled_time falls on a holiday or weekend, it moves to the
|
||||
next business day at the same time.
|
||||
"""
|
||||
original_date = scheduled_time.date()
|
||||
next_date = self.get_next_business_day(original_date)
|
||||
|
||||
if next_date == original_date:
|
||||
return scheduled_time
|
||||
|
||||
# Combine the new date with the original time
|
||||
return datetime.combine(next_date, scheduled_time.time())
|
||||
|
||||
# Example usage for testing
|
||||
if __name__ == "__main__":
|
||||
calendar = BusinessCalendar()
|
||||
|
||||
# Test dates
|
||||
dates_to_test = [
|
||||
date(2026, 5, 1), # Holiday (Labor Day)
|
||||
date(2026, 12, 25), # Holiday (Christmas)
|
||||
date(2026, 4, 6), # Holiday (Easter Monday 2026)
|
||||
date(2026, 2, 10), # Likely a Tuesday (Business Day)
|
||||
date(2026, 2, 14) # Saturday
|
||||
]
|
||||
|
||||
print("--- Business Day Check (Bayern 2026) ---")
|
||||
for d in dates_to_test:
|
||||
is_biz = calendar.is_business_day(d)
|
||||
next_biz = calendar.get_next_business_day(d)
|
||||
holiday_name = calendar.holidays.get(d) if d in calendar.holidays else ""
|
||||
|
||||
status = "✅ Business Day" if is_biz else f"❌ Blocked ({holiday_name if holiday_name else 'Weekend'})"
|
||||
print(f"Date: {d} | {status} -> Next: {next_biz}")
|
||||
Reference in New Issue
Block a user