Files
Brancheneinstufung2/connector-superoffice/main.py
Floke 938a81fd97 feat(connector-superoffice): implement OAuth 2.0 flow and S2S architecture
Completed POC for SuperOffice integration with the following key achievements:
- Switched from RSA/SOAP to OAuth 2.0 (Refresh Token Flow) for better compatibility with SOD environment.
- Implemented robust token refreshing and caching mechanism in .
- Solved 'Wrong Subdomain' issue by enforcing  for tenant .
- Created  for REST API interaction (Search, Create, Update UDFs).
- Added helper scripts: , , .
- Documented usage and configuration in .
- Updated  configuration requirements.

[2ff88f42]
2026-02-09 16:04:16 +00:00

60 lines
1.9 KiB
Python

import os
import logging
from dotenv import load_dotenv
from auth_handler import AuthHandler
from superoffice_client import SuperOfficeClient
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def main():
# Load .env from the root directory
load_dotenv(dotenv_path="../.env")
logger.info("Starting SuperOffice Connector S2S POC...")
try:
# Initialize Auth
auth = AuthHandler()
# Initialize Client
client = SuperOfficeClient(auth)
# 1. Test Connection
logger.info("Step 1: Testing connection...")
user_info = client.test_connection()
if user_info:
logger.info(f"Connected successfully as: {user_info.get('Name')}")
else:
logger.error("Connection test failed.")
return
# 2. Search for a test contact
test_company = "Wackler Holding"
logger.info(f"Step 2: Searching for company '{test_company}'...")
contact = client.find_contact_by_criteria(name=test_company)
if contact:
logger.info(f"Found contact: {contact.get('Name')} (ID: {contact.get('ContactId')})")
# 3. Try to update UDFs (Warning: technical names might be wrong)
# logger.info("Step 3: Attempting UDF update (experimental)...")
# ai_test_data = {
# "potential": "High",
# "industry": "Facility Management",
# "summary": "KI-Analyse erfolgreich durchgeführt."
# }
# client.update_udfs(contact.get('ContactId'), ai_test_data)
else:
logger.info(f"Contact '{test_company}' not found.")
except Exception as e:
logger.error(f"An error occurred: {e}", exc_info=True)
if __name__ == "__main__":
main()