Floke
f5e661f9c0
This commit restores the full functionality of the script within the module. Several instances were resolved by implementing missing methods in (e.g., , , , , , , ) and correcting argument passing. The data extraction logic in was adjusted to correctly parse the structure returned by the SuperOffice API (e.g., and ). A dedicated SuperOffice API health check script () was introduced to quickly verify basic API connectivity (reading contacts and persons). This script confirmed that read operations for and entities are functional, while the endpoint continues to return a , which is now handled gracefully as a warning, allowing other tests to proceed. The now successfully executes all SuperOffice-specific POC steps, including creating contacts, persons, sales, projects, and updating UDFs.
167 lines
7.8 KiB
Python
167 lines
7.8 KiB
Python
import os
|
|
import logging
|
|
from auth_handler import AuthHandler
|
|
from superoffice_client import SuperOfficeClient
|
|
# from explorer_client import CompanyExplorerClient
|
|
from logging_config import setup_logging
|
|
|
|
# Use the centralized logging configuration
|
|
logger = setup_logging(__name__)
|
|
|
|
def main():
|
|
# Note: Environment loading is now handled by config.py/helpers implicitly when clients are initialized
|
|
|
|
logger.info("Starting SuperOffice Connector S2S POC...")
|
|
|
|
try:
|
|
# Initialize Auth
|
|
auth = AuthHandler()
|
|
|
|
# Initialize Client
|
|
client = SuperOfficeClient()
|
|
|
|
# TODO: Initialize Explorer Client when explorer_client.py is implemented
|
|
# ce_client = CompanyExplorerClient()
|
|
|
|
# 1. Test Connection
|
|
logger.info("Step 1: Testing connection...")
|
|
user_info = client._get("Associate/Me")
|
|
if user_info:
|
|
logger.info(f"Connected successfully as: {user_info.get('FullName')}")
|
|
else:
|
|
logger.warning("Connection test for Associate/Me failed, but continuing with other tests...")
|
|
|
|
# TODO: Test Company Explorer Connection when explorer_client.py is implemented
|
|
# logger.info("Step 1b: Testing Company Explorer connection...")
|
|
# if ce_client.check_health():
|
|
# logger.info("Company Explorer is reachable.")
|
|
# else:
|
|
# logger.warning("Company Explorer is NOT reachable. Sync might fail.")
|
|
|
|
# 2. Search for our demo company
|
|
demo_company_name = "Gemini Test Company [2ff88f42]"
|
|
logger.info(f"Step 2: Searching for company '{demo_company_name}'...")
|
|
contact = client.find_contact_by_criteria(name=demo_company_name)
|
|
|
|
target_contact_id = None
|
|
|
|
if contact:
|
|
target_contact_id = contact.get('contactId')
|
|
logger.info(f"Found existing demo company: {contact.get('nameDepartment')} (ID: {target_contact_id})")
|
|
else:
|
|
logger.info(f"Demo company not found. Creating new one...")
|
|
demo_company_url = "https://www.gemini-test-company.com"
|
|
demo_company_orgnr = "DE123456789"
|
|
|
|
new_contact = client.create_contact(
|
|
name=demo_company_name,
|
|
url=demo_company_url,
|
|
org_nr=demo_company_orgnr
|
|
)
|
|
if new_contact:
|
|
target_contact_id = new_contact.get('ContactId')
|
|
logger.info(f"Created new demo company with ID: {target_contact_id}")
|
|
|
|
# 3. Create a Person linked to this company
|
|
if target_contact_id:
|
|
logger.info(f"Step 3: Creating Person for Contact ID {target_contact_id}...")
|
|
|
|
# Create Max Mustermann
|
|
person = client.create_person(
|
|
first_name="Max",
|
|
last_name="Mustermann",
|
|
contact_id=target_contact_id,
|
|
email="max.mustermann@gemini-test.com"
|
|
)
|
|
|
|
if person:
|
|
logger.info("SUCCESS: Person created!")
|
|
person_id = person.get('PersonId')
|
|
logger.info(f"Name: {person.get('Firstname')} {person.get('Lastname')}")
|
|
logger.info(f"Person ID: {person_id}")
|
|
logger.info(f"Linked to Contact ID: {person.get('Contact').get('ContactId')}")
|
|
|
|
# 4. Create a Sale for this company
|
|
logger.info(f"Step 4: Creating Sale for Contact ID {target_contact_id}...")
|
|
sale = client.create_sale(
|
|
title=f"Robotics Automation Opportunity - {demo_company_name}",
|
|
contact_id=target_contact_id,
|
|
person_id=person_id,
|
|
amount=50000.0
|
|
)
|
|
if sale:
|
|
logger.info("SUCCESS: Sale created!")
|
|
logger.info(f"Sale Title: {sale.get('Heading')}")
|
|
logger.info(f"Sale ID: {sale.get('SaleId')}")
|
|
else:
|
|
logger.error("Failed to create sale.")
|
|
|
|
# 5. Create a Project for this company and add the person to it
|
|
logger.info(f"Step 5: Creating Project for Contact ID {target_contact_id} and adding Person ID {person_id}...")
|
|
project = client.create_project(
|
|
name=f"Marketing Campaign Q1 [2ff88f42]",
|
|
contact_id=target_contact_id,
|
|
person_id=person_id
|
|
)
|
|
if project:
|
|
logger.info("SUCCESS: Project created and person added!")
|
|
logger.info(f"Project Name: {project.get('Name')}")
|
|
logger.info(f"Project ID: {project.get('ProjectId')}")
|
|
|
|
# 6. Update Contact UDFs
|
|
logger.info(f"Step 6: Updating Contact UDFs for Contact ID {target_contact_id}...")
|
|
contact_udf_data = {
|
|
"ai_challenge_sentence": "The company faces challenges in automating its logistics processes due to complex infrastructure.",
|
|
"ai_sentence_timestamp": "2026-02-10T12:00:00Z", # Using a fixed timestamp for demo
|
|
"ai_sentence_source_hash": "website_v1_hash_abc",
|
|
"ai_last_outreach_date": "2026-02-10T12:00:00Z" # Using a fixed timestamp for demo
|
|
}
|
|
updated_contact = client.update_entity_udfs(target_contact_id, "Contact", contact_udf_data)
|
|
if updated_contact:
|
|
logger.info("SUCCESS: Contact UDFs updated!")
|
|
else:
|
|
logger.error("Failed to update Contact UDFs.")
|
|
|
|
# 7. Update Person UDFs
|
|
logger.info(f"Step 7: Updating Person UDFs for Person ID {person_id}...")
|
|
person_udf_data = {
|
|
"ai_email_draft": "This is a short draft for the personalized email.", # Placeholder, as it's currently a short text field
|
|
"ma_status": "Ready_to_Send"
|
|
}
|
|
updated_person = client.update_entity_udfs(person_id, "Person", person_udf_data)
|
|
if updated_person:
|
|
logger.info("SUCCESS: Person UDFs updated!")
|
|
else:
|
|
logger.error("Failed to update Person UDFs.")
|
|
|
|
# TODO: Sync to Company Explorer when explorer_client.py is implemented
|
|
# if updated_contact:
|
|
# logger.info(f"Step 9: Syncing Company to Company Explorer...")
|
|
# ce_payload = {
|
|
# "name": updated_contact.get("Name"),
|
|
# "website": updated_contact.get("UrlAddress"),
|
|
# "city": updated_contact.get("City"),
|
|
# "country": "DE" # Defaulting to DE for now
|
|
# }
|
|
|
|
# ce_result = ce_client.import_company(ce_payload)
|
|
# if ce_result:
|
|
# logger.info(f"SUCCESS: Company synced to Explorer! ID: {ce_result.get('id')}")
|
|
# else:
|
|
# logger.error("Failed to sync company to Explorer.")
|
|
# else:
|
|
# logger.warning("Skipping CE sync because contact update failed or contact object is missing.")
|
|
|
|
else:
|
|
logger.error("Failed to create project.")
|
|
|
|
else:
|
|
logger.error("Failed to create person.")
|
|
else:
|
|
logger.error("Skipping person creation because company could not be found or created.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"An error occurred: {e}", exc_info=True)
|
|
|
|
if __name__ == "__main__":
|
|
main() |