Floke/Brancheneinstufung2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(so-api): [31188f42] Diagnosis of email sending blockade & cleanup

Browse Source
This commit is contained in:
Floke
2026-03-05 06:19:45 +00:00
parent be00ea0daa
commit aad293f6fb
9 changed files with 284 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
connector-superoffice/config.py
View File

@@ -1,4 +1,12 @@
import os
from dotenv import load_dotenv
# Explicitly load .env from the project root.
# CRITICAL: override=True ensures we read from the .env file even if
# stale env vars are present in the shell process.
dotenv_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '.env'))
if os.path.exists(dotenv_path):
load_dotenv(dotenv_path=dotenv_path, override=True)
class Settings:
def __init__(self):

34
connector-superoffice/create_sale_test.py
View File

@@ -99,16 +99,18 @@ def main():
print("\n--- 1. Finding a Target Contact (Company) ---")
# Search for "Test" to avoid hitting the Wackler parent company (ID 3)
contacts = client._get("Contact?$top=1&$filter=name contains 'Test'&$select=ContactId,Name")
# contacts = client._get("Contact?$top=1&$filter=name contains 'Test'&$select=ContactId,Name")
# Fallback if no test company found, but warn user
if not contacts or 'value' not in contacts or len(contacts['value']) == 0:
print("⚠️ No company with 'Test' found. Please create a test company first.")
return
# if not contacts or 'value' not in contacts or len(contacts['value']) == 0:
# print("⚠️ No company with 'Test' found. Please create a test company first.")
# return
target_contact = contacts['value'][0]
contact_id = target_contact.get('contactId') or target_contact.get('ContactId')
contact_name = target_contact.get('name') or target_contact.get('Name')
# target_contact = contacts['value'][0]
# contact_id = target_contact.get('contactId') or target_contact.get('ContactId')
# contact_name = target_contact.get('name') or target_contact.get('Name')
contact_id = 171185
contact_name = "Bremer Abenteuerland"
# SAFEGUARD: Do not post to Wackler Service Group (ID 3)
if int(contact_id) == 3:
@@ -156,18 +158,16 @@ def main():
print(f"Payload Preview: {json.dumps(sale_payload, indent=2)}")
# Uncomment to actually run creation
# new_sale = client._post("Sale", sale_payload)
new_sale = client._post("Sale", sale_payload)
# print("\n--- ✅ SUCCESS: Sale Created! ---")
# sale_id = new_sale.get('SaleId')
# sale_number = new_sale.get('SaleNumber')
# print(f"Sale ID: {sale_id}")
# print(f"Sale Number: {sale_number}")
print("\n--- ✅ SUCCESS: Sale Created! ---")
sale_id = new_sale.get('SaleId')
sale_number = new_sale.get('SaleNumber')
print(f"Sale ID: {sale_id}")
print(f"Sale Number: {sale_number}")
# sale_link = f"https://{auth.env}.superoffice.com/{auth.cust_id}/default.aspx?sale?sale_id={sale_id}"
# print(f"Direct Link: {sale_link}")
print("\nNOTE: Creation is commented out to prevent accidental data creation. Review payload above.")
sale_link = f"https://{auth.env}.superoffice.com/{auth.cust_id}/default.aspx?sale?sale_id={sale_id}"
print(f"Direct Link: {sale_link}")
except requests.exceptions.HTTPError as e:
logger.error(f"❌ API Error: {e}")

68
connector-superoffice/simulate_sendout_via_appointment.py
View File

@@ -9,8 +9,8 @@ from config import settings
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("simulation-e2e")
def simulate_sendout(contact_id: int, person_id: int):
print(f"🚀 Starting E2E Sendout Simulation for Contact {contact_id}, Person {person_id}...")
def simulate_sendout(contact_id: int):
print(f"🚀 Starting Appointment Creation Test for Contact {contact_id}...")
# 1. Initialize SuperOffice Client
so_client = SuperOfficeClient()
@@ -18,73 +18,27 @@ def simulate_sendout(contact_id: int, person_id: int):
print("❌ Auth failed. Check .env")
return
# 2. Get Data from Company Explorer
# We simulate what the worker would do
print(f"📡 Requesting provisioning from Company Explorer...")
ce_url = f"{settings.COMPANY_EXPLORER_URL}/api/provision/superoffice-contact"
ce_req = {
"so_contact_id": contact_id,
"so_person_id": person_id,
"crm_name": "RoboPlanet GmbH",
"crm_website": "www.roboplanet.de",
"job_title": "Geschäftsführer" # Explicit job title for persona mapping
}
ce_auth = (os.getenv("API_USER", "admin"), os.getenv("API_PASSWORD", "gemini"))
try:
resp = requests.post(ce_url, json=ce_req, auth=ce_auth)
resp.raise_for_status()
provisioning_data = resp.json()
except Exception as e:
print(f"❌ CE API failed: {e}")
return
print(f"✅ Received Data: {json.dumps(provisioning_data, indent=2)}")
if provisioning_data.get("status") == "processing":
print("⏳ CE is still processing. Please wait 1-2 minutes and try again.")
return
texts = provisioning_data.get("texts", {})
if not texts.get("subject"):
print("⚠️ No marketing texts found for this combination (Vertical x Persona).")
return
# 3. Write Texts to SuperOffice UDFs
print("✍️ Writing marketing texts to SuperOffice UDFs...")
udf_payload = {
settings.UDF_SUBJECT: texts["subject"],
settings.UDF_INTRO: texts["intro"],
settings.UDF_SOCIAL_PROOF: texts["social_proof"]
}
success = so_client.update_entity_udfs(person_id, "Person", udf_payload)
if success:
print("✅ UDFs updated successfully.")
else:
print("❌ Failed to update UDFs.")
return
# 4. Create Appointment (The "Sendout Proof")
# 2. Create Appointment (The "Sendout Proof")
print("📅 Creating Appointment as sendout proof...")
app_subject = f"[SIMULATION] Mail Sent: {texts['subject']}"
app_desc = f"Content Simulation:\n\n{texts['intro']}\n\n{texts['social_proof']}"
app_subject = f"[DIAGNOSTIC TEST] Can we create an activity?"
app_desc = f"This is a test to see if the API user can create appointments."
appointment = so_client.create_appointment(
subject=app_subject,
description=app_desc,
contact_id=contact_id,
person_id=person_id
person_id=None # Explicitly test without a person
)
if appointment:
print(f"✅ Simulation Complete! Appointment ID: {appointment.get('AppointmentId')}")
# The key might be 'appointmentId' (lowercase 'a')
appt_id = appointment.get('appointmentId') or appointment.get('AppointmentId')
print(f"✅ SUCCESS! Appointment Created with ID: {appt_id}")
print(f"🔗 Check SuperOffice for Contact {contact_id} and look at the activities.")
else:
print("❌ Failed to create appointment.")
if __name__ == "__main__":
# Using the IDs we know exist from previous tests/status
TEST_CONTACT_ID = 2
TEST_PERSON_ID = 2 # Usually same or linked
simulate_sendout(TEST_CONTACT_ID, TEST_PERSON_ID)
TEST_CONTACT_ID = 171185
simulate_sendout(TEST_CONTACT_ID)

8
connector-superoffice/superoffice_client.py
View File

@@ -93,6 +93,8 @@ class SuperOfficeClient:
resp = requests.put(url, headers=self.headers, json=payload)
elif method == "PATCH":
resp = requests.patch(url, headers=self.headers, json=payload)
elif method == "DELETE":
resp = requests.delete(url, headers=self.headers)
# 401 Handling
if resp.status_code == 401 and retry:
@@ -108,6 +110,9 @@ class SuperOfficeClient:
logger.error("❌ Token Refresh failed during retry.")
return None
if resp.status_code == 204:
return True
resp.raise_for_status()
return resp.json()
@@ -130,6 +135,9 @@ class SuperOfficeClient:
def _post(self, endpoint, payload):
return self._request_with_retry("POST", endpoint, payload)
def _delete(self, endpoint):
return self._request_with_retry("DELETE", endpoint)
# --- Convenience Wrappers ---
def get_person(self, person_id, select: list = None):

7
connector-superoffice/tools/create_company.py
View File

@@ -24,8 +24,9 @@ def create_test_company():
return
# Check if company already exists
existing = client.search(f"Contact?$select=contactId,name&$filter=name eq '{company_name}'")
print(f"DEBUG: Raw search response: {existing}")
if existing:
contact_id = existing[0]['ContactId']
contact_id = existing[0]['contactId']
print(f"⚠️ Company '{company_name}' already exists with ContactId: {contact_id}.")
print("Skipping creation.")
return contact_id
@@ -42,8 +43,8 @@ def create_test_company():
}
}
new_company = client._post("Contact", payload)
if new_company and "ContactId" in new_company:
contact_id = new_company["ContactId"]
if new_company and "contactId" in new_company:
contact_id = new_company["contactId"]
print(f"✅ SUCCESS! Created company '{company_name}' with ContactId: {contact_id}")
return contact_id
else:

44
connector-superoffice/tools/create_person_test.py Normal file
View File

@@ -0,0 +1,44 @@
import sys
import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from superoffice_client import SuperOfficeClient
def create_test_person(contact_id: int):
"""
Creates a new person for a given contact ID.
"""
print(f"🚀 Attempting to create a person for Contact ID: {contact_id}")
try:
client = SuperOfficeClient()
if not client.access_token:
print("❌ Authentication failed.")
return
payload = {
"Contact": {"ContactId": contact_id},
"Firstname": "Test",
"Lastname": "Person",
"Emails": [
{
"Value": "floke.com@gmail.com",
"Description": "Work Email"
}
]
}
new_person = client._post("Person", payload)
if new_person and "PersonId" in new_person:
person_id = new_person["PersonId"]
print(f"✅ SUCCESS! Created person with PersonId: {person_id}")
return person_id
else:
print(f"❌ Failed to create person. Response: {new_person}")
return None
except Exception as e:
print(f"An error occurred: {e}")
return None
if __name__ == "__main__":
TEST_CONTACT_ID = 171185
if len(sys.argv) > 1:
TEST_CONTACT_ID = int(sys.argv[1])
create_test_person(TEST_CONTACT_ID)