Stabilize Lead Engine calendar logic (v1.4) and integrate GTM Architect, B2B Assistant, and Transcription Tool into Docker stack [30388f42]
This commit is contained in:
88
lead-engine/trading_twins/test_calendar_logic.py
Normal file
88
lead-engine/trading_twins/test_calendar_logic.py
Normal file
@@ -0,0 +1,88 @@
|
||||
# lead-engine/trading_twins/test_calendar_logic.py
|
||||
import sys
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from zoneinfo import ZoneInfo
|
||||
from dotenv import load_dotenv
|
||||
import msal
|
||||
import requests
|
||||
|
||||
# Load environment variables from the root .env
|
||||
load_dotenv(dotenv_path="/app/.env", override=True)
|
||||
|
||||
# Pfad anpassen, damit wir manager importieren können
|
||||
sys.path.append('/app')
|
||||
|
||||
from trading_twins.manager import get_availability, find_slots
|
||||
|
||||
# Re-import variables to ensure we see what's loaded
|
||||
CAL_APPID = os.getenv("CAL_APPID")
|
||||
CAL_SECRET = os.getenv("CAL_SECRET")
|
||||
CAL_TENNANT_ID = os.getenv("CAL_TENNANT_ID")
|
||||
|
||||
TZ_BERLIN = ZoneInfo("Europe/Berlin")
|
||||
|
||||
def test_internal():
|
||||
target = "e.melcer@robo-planet.de"
|
||||
print(f"🔍 Teste Kalender-Logik für {target}...")
|
||||
|
||||
# Debug Token Acquisition
|
||||
print("🔑 Authentifiziere mit MS Graph...")
|
||||
authority = f"https://login.microsoftonline.com/{CAL_TENNANT_ID}"
|
||||
app_msal = msal.ConfidentialClientApplication(client_id=CAL_APPID, authority=authority, client_credential=CAL_SECRET)
|
||||
result = app_msal.acquire_token_silent([".default"], account=None)
|
||||
if not result:
|
||||
print(" ... hole neues Token ...")
|
||||
result = app_msal.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
|
||||
|
||||
if "access_token" in result:
|
||||
print("✅ Token erhalten.")
|
||||
token = result['access_token']
|
||||
else:
|
||||
print(f"❌ Token-Fehler: {result.get('error')}")
|
||||
print(f"❌ Beschreibung: {result.get('error_description')}")
|
||||
return
|
||||
|
||||
# Debug API Call
|
||||
print("📡 Frage Kalender ab...")
|
||||
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json", "Prefer": 'outlook.timezone="Europe/Berlin"'}
|
||||
start_time = datetime.now(TZ_BERLIN).replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
end_time = start_time + timedelta(days=3)
|
||||
|
||||
payload = {
|
||||
"schedules": [target],
|
||||
"startTime": {"dateTime": start_time.isoformat(), "timeZone": "Europe/Berlin"},
|
||||
"endTime": {"dateTime": end_time.isoformat(), "timeZone": "Europe/Berlin"},
|
||||
"availabilityViewInterval": 15
|
||||
}
|
||||
|
||||
import requests
|
||||
try:
|
||||
url = f"https://graph.microsoft.com/v1.0/users/{target}/calendar/getSchedule"
|
||||
r = requests.post(url, headers=headers, json=payload)
|
||||
|
||||
print(f"📡 API Status: {r.status_code}")
|
||||
if r.status_code == 200:
|
||||
data = r.json()
|
||||
# print(f"DEBUG RAW: {data}")
|
||||
schedule = data['value'][0]
|
||||
view = schedule.get('availabilityView', '')
|
||||
print(f"✅ Verfügbarkeit (View Länge: {len(view)})")
|
||||
|
||||
# Test Slot Finding
|
||||
slots = find_slots(start_time, view, 15)
|
||||
if slots:
|
||||
print(f"✅ {len(slots)} Slots gefunden:")
|
||||
for s in slots:
|
||||
print(f" 📅 {s.strftime('%A, %d.%m.%Y um %H:%M')}")
|
||||
else:
|
||||
print("⚠️ Keine Slots gefunden (Logik korrekt, aber Kalender voll?)")
|
||||
else:
|
||||
print(f"❌ API Fehler: {r.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Exception beim API Call: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_internal()
|
||||
Reference in New Issue
Block a user