89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
# 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()
|