Floke
c25ed4d941
- worker.py: Circuit Breaker implementiert (Ignoriert Associate ID 528), um Ping-Pong-Loops zu verhindern. - worker.py: Resiliente UDF-Behandlung hinzugefügt (behebt 'unhashable type: dict' API-Antwort-Problem). - tools/: Umfangreiche Test- und Diagnose-Suite hinzugefügt. Die Anreicherung für 'Bremer Abenteuerland' wurde erfolgreich verifiziert.
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
import sys
|
|
import os
|
|
import json
|
|
from dotenv import load_dotenv
|
|
|
|
# Explicitly load .env from the project root
|
|
dotenv_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '.env'))
|
|
load_dotenv(dotenv_path=dotenv_path, override=True)
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
from superoffice_client import SuperOfficeClient
|
|
|
|
def get_current_user():
|
|
print(f"🚀 Fetching current user info via Associate/Me...")
|
|
try:
|
|
client = SuperOfficeClient()
|
|
if not client.access_token:
|
|
print("❌ Authentication failed.")
|
|
return
|
|
|
|
# Try the most reliable endpoint for current user context
|
|
user = client._get("Associate/Me")
|
|
|
|
if user:
|
|
print("\n--- 👤 Current User Info ---")
|
|
print(f"Associate ID: {user.get('AssociateId')}")
|
|
print(f"Name: {user.get('FullName')}")
|
|
print(f"UserName: {user.get('UserName')}")
|
|
print("----------------------------")
|
|
return user.get('AssociateId')
|
|
else:
|
|
# Fallback: List all associates and try to match by name or username
|
|
print("⚠️ Associate/Me failed. Trying alternative...")
|
|
# This might be too much data, but let's see
|
|
return None
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
return None
|
|
|
|
if __name__ == "__main__":
|
|
get_current_user() |