""" Test: PhoneNumber und EmailAddress als System-Entities Hypothese: - PhoneNumber und EmailAddress sind separate Entities mit IDs - CBeteiligte hat Links/Relations zu diesen Entities - Wir können über related entries an die IDs kommen Ziele: 1. Hole CBeteiligte mit expanded relationships 2. Prüfe ob phoneNumbers/emailAddresses als Links verfügbar sind 3. Extrahiere IDs der verknüpften PhoneNumber/EmailAddress Entities """ import asyncio import json from services.espocrm import EspoCRMAPI TEST_BETEILIGTE_ID = '68e4af00172be7924' class SimpleContext: class Logger: def info(self, msg): print(f"[INFO] {msg}") def error(self, msg): print(f"[ERROR] {msg}") def warning(self, msg): print(f"[WARN] {msg}") def debug(self, msg): pass def __init__(self): self.logger = self.Logger() def print_section(title): print("\n" + "="*70) print(title) print("="*70) async def test_related_entities(): """Test 1: Hole CBeteiligte mit allen verfügbaren Feldern""" print_section("TEST 1: CBeteiligte - Alle Felder") context = SimpleContext() espo = EspoCRMAPI(context) # Hole Entity entity = await espo.get_entity('CBeteiligte', TEST_BETEILIGTE_ID) print(f"\n✅ Beteiligter: {entity.get('name')}") print(f"\n📋 Alle Top-Level Felder:") for key in sorted(entity.keys()): value = entity[key] value_type = type(value).__name__ # Zeige nur ersten Teil von langen Werten if isinstance(value, str) and len(value) > 60: display = f"{value[:60]}..." elif isinstance(value, list): display = f"[{len(value)} items]" elif isinstance(value, dict): display = f"{{dict with {len(value)} keys}}" else: display = value print(f" • {key:30s}: {value_type:10s} = {display}") # Suche nach ID-Feldern für Kommunikation print(f"\n🔍 Suche nach ID-Feldern für Email/Phone:") potential_id_fields = [k for k in entity.keys() if 'email' in k.lower() or 'phone' in k.lower()] for field in potential_id_fields: print(f" • {field}: {entity.get(field)}") return entity async def test_list_with_select(): """Test 2: Nutze select Parameter um spezifische Felder zu holen""" print_section("TEST 2: CBeteiligte mit select Parameter") context = SimpleContext() espo = EspoCRMAPI(context) # Versuche verschiedene Feld-Namen potential_fields = [ 'emailAddresses', 'phoneNumbers', 'emailAddressId', 'phoneNumberId', 'emailAddressIds', 'phoneNumberIds', 'emailAddressList', 'phoneNumberList' ] print(f"\n📋 Teste verschiedene Feld-Namen:") for field in potential_fields: try: result = await espo.api_call( f'CBeteiligte/{TEST_BETEILIGTE_ID}', params={'select': field} ) if result and field in result: print(f" ✅ {field:30s}: {result[field]}") else: print(f" ❌ {field:30s}: Nicht im Response") except Exception as e: print(f" ❌ {field:30s}: Error - {e}") async def test_entity_relationships(): """Test 3: Hole Links/Relationships über dedizierte Endpoints""" print_section("TEST 3: Entity Relationships") context = SimpleContext() espo = EspoCRMAPI(context) # Test verschiedene Relationship-Endpoints relationship_names = [ 'emailAddresses', 'phoneNumbers', 'emails', 'phones' ] for rel_name in relationship_names: print(f"\n🔗 Teste Relationship: {rel_name}") try: # EspoCRM API Format: /Entity/{id}/relationship-name result = await espo.api_call(f'CBeteiligte/{TEST_BETEILIGTE_ID}/{rel_name}') if result: print(f" ✅ Success! Type: {type(result)}") if isinstance(result, dict): print(f" 📋 Response Keys: {list(result.keys())}") # Häufige EspoCRM Response-Strukturen if 'list' in result: items = result['list'] print(f" 📊 {len(items)} Einträge in 'list'") if items: print(f"\n Erster Eintrag:") print(json.dumps(items[0], indent=6, ensure_ascii=False)) if 'total' in result: print(f" 📊 Total: {result['total']}") elif isinstance(result, list): print(f" 📊 {len(result)} Einträge direkt als Liste") if result: print(f"\n Erster Eintrag:") print(json.dumps(result[0], indent=6, ensure_ascii=False)) else: print(f" ⚠️ Empty response") except Exception as e: error_msg = str(e) if '404' in error_msg: print(f" ❌ 404 Not Found - Relationship existiert nicht") elif '403' in error_msg: print(f" ❌ 403 Forbidden - Kein Zugriff") else: print(f" ❌ Error: {error_msg}") async def test_direct_entity_access(): """Test 4: Direkter Zugriff auf PhoneNumber/EmailAddress Entities""" print_section("TEST 4: Direkte Entity-Abfrage") context = SimpleContext() espo = EspoCRMAPI(context) # Versuche die Entities direkt zu listen for entity_type in ['PhoneNumber', 'EmailAddress']: print(f"\n📋 Liste {entity_type} Entities:") try: # Mit Filter für unseren Beteiligten result = await espo.api_call( entity_type, params={ 'maxSize': 5, 'where': json.dumps([{ 'type': 'equals', 'attribute': 'parentId', 'value': TEST_BETEILIGTE_ID }]) } ) if result and 'list' in result: items = result['list'] print(f" ✅ {len(items)} Einträge gefunden") for item in items: print(f"\n 📧/📞 {entity_type}:") print(json.dumps(item, indent=6, ensure_ascii=False)) else: print(f" ⚠️ Keine Einträge oder unerwartetes Format") print(f" Response: {result}") except Exception as e: error_msg = str(e) if '403' in error_msg: print(f" ❌ 403 Forbidden") print(f" → Versuche ohne Filter...") try: # Ohne Filter result = await espo.api_call(entity_type, params={'maxSize': 3}) print(f" ✅ Ohne Filter: {result.get('total', 0)} total existieren") except Exception as e2: print(f" ❌ Auch ohne Filter: {e2}") else: print(f" ❌ Error: {error_msg}") async def test_espocrm_metadata(): """Test 5: Prüfe EspoCRM Metadata für CBeteiligte""" print_section("TEST 5: EspoCRM Metadata") context = SimpleContext() espo = EspoCRMAPI(context) print(f"\n📋 Hole Metadata für CBeteiligte:") try: # EspoCRM bietet manchmal Metadata-Endpoints result = await espo.api_call('Metadata') if result and 'entityDefs' in result: if 'CBeteiligte' in result['entityDefs']: bet_meta = result['entityDefs']['CBeteiligte'] print(f"\n ✅ CBeteiligte Metadata gefunden") if 'links' in bet_meta: print(f"\n 🔗 Links/Relationships:") for link_name, link_def in bet_meta['links'].items(): if 'email' in link_name.lower() or 'phone' in link_name.lower(): print(f" • {link_name}: {link_def}") if 'fields' in bet_meta: print(f"\n 📋 Relevante Felder:") for field_name, field_def in bet_meta['fields'].items(): if 'email' in field_name.lower() or 'phone' in field_name.lower(): print(f" • {field_name}: {field_def.get('type', 'unknown')}") else: print(f" ⚠️ Unerwartetes Format") except Exception as e: print(f" ❌ Error: {e}") async def main(): print("\n" + "="*70) print("ESPOCRM PHONENUMBER/EMAILADDRESS - ENTITIES & IDS") print("="*70) print("\nZiel: Finde IDs für PhoneNumber/EmailAddress über Relationships\n") try: # Test 1: Alle Felder inspizieren entity = await test_related_entities() # Test 2: Select Parameter await test_list_with_select() # Test 3: Relationships await test_entity_relationships() # Test 4: Direkte Entity-Abfrage await test_direct_entity_access() # Test 5: Metadata await test_espocrm_metadata() print_section("ZUSAMMENFASSUNG") print("\n🎯 Erkenntnisse:") print("\n Wenn PhoneNumber/EmailAddress System-Entities sind:") print(" 1. ✅ Sie haben eigene IDs") print(" 2. ✅ Stabiles Matching möglich") print(" 3. ✅ Bidirektionaler Sync machbar") print(" 4. ✅ Change Detection via ID") print("\n Wenn wir IDs haben:") print(" • Können Advoware-ID zu EspoCRM-ID mappen") print(" • Können Änderungen tracken") print(" • Kein Problem bei Wert-Änderungen") except Exception as e: print(f"\n❌ Fehler: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())