Brancheneinstufung2/connector-superoffice/superoffice_client.py

import requests
import logging
from auth_handler import AuthHandler

logger = logging.getLogger(__name__)

class SuperOfficeClient:
    """
    A client for interacting with the SuperOffice REST API using OAuth 2.0 Bearer tokens.
    """
    def __init__(self, auth_handler: AuthHandler):
        self.auth_handler = auth_handler
        self.session = requests.Session()
        
        # Mapping for UDF fields (These are typical technical names, but might need adjustment)
        self.udf_mapping = {
            "robotics_potential": "x_robotics_potential", 
            "industry": "x_ai_industry",
            "summary": "x_ai_summary",
            "last_update": "x_ai_last_update",
            "status": "x_ai_status"
        }
        
        # Mapping for list values (Explorer -> SO ID)
        self.potential_id_map = {
            "High": 1,
            "Medium": 2,
            "Low": 3,
            "None": 4
        }

    def _get_headers(self):
        """Returns the authorization headers with Bearer token."""
        access_token, _ = self.auth_handler.get_ticket()
        return {
            'Authorization': f'Bearer {access_token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }

    def _get_url(self, path):
        """Constructs the full URL for a given API path."""
        _, webapi_url = self.auth_handler.get_ticket()
        base = webapi_url.rstrip('/')
        p = path.lstrip('/')
        return f"{base}/api/{p}"

    def test_connection(self):
        """Tests the connection by fetching the current user."""
        url = self._get_url("v1/User/currentPrincipal")
        try:
            resp = self.session.get(url, headers=self._get_headers())
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.error(f"Connection test failed: {e}")
            if hasattr(e, 'response') and e.response is not None:
                logger.error(f"Response: {e.response.text}")
            return None

    def find_contact_by_criteria(self, name=None, url=None, org_nr=None):
        """Searches for a contact by OrgNr, URL, or Name."""
        filters = []
        if org_nr:
            filters.append(f"orgNr eq '{org_nr}'")
        if url:
            filters.append(f"urlAddress eq '{url}'")
        
        if not filters and name:
            filters.append(f"name contains '{name}'")
            
        if not filters:
            return None

        query = " and ".join(filters)
        path = f"v1/Contact?$filter={query}"
        
        try:
            full_url = self._get_url(path)
            resp = self.session.get(full_url, headers=self._get_headers())
            resp.raise_for_status()
            data = resp.json()
            
            results = data.get("value", [])
            if results:
                logger.info(f"Found {len(results)} matching contacts.")
                return results[0]
            return None
        except Exception as e:
            logger.error(f"Error searching for contact: {e}")
            return None

    def create_contact(self, name, url=None, org_nr=None):
        """Creates a new contact (company) in SuperOffice with basic details."""
        url = self._get_url("v1/Contact")
        payload = {
            "Name": name,
            "OrgNr": org_nr,
            "UrlAddress": url,
            "ActivePublications": [],  # Required field, can be empty
            "Emails": [], # Required field, can be empty
            "Phones": [] # Required field, can be empty
        }
        
        # Remove None values
        payload = {k: v for k, v in payload.items() if v is not None}

        try:
            logger.info(f"Attempting to create contact: {name}")
            resp = self.session.post(url, headers=self._get_headers(), json=payload)
            resp.raise_for_status()
            created_contact = resp.json()
            logger.info(f"Successfully created contact: {created_contact.get('Name')} (ID: {created_contact.get('ContactId')})")
            return created_contact
        except Exception as e:
            if hasattr(e, 'response') and e.response is not None:
                logger.error(f"Response: {e.response.text}")
            return None

    def create_person(self, first_name, last_name, contact_id, email=None):
        """Creates a new person linked to a contact (company)."""
        url = self._get_url("v1/Person")
        
        payload = {
            "Firstname": first_name,
            "Lastname": last_name,
            "Contact": {
                "ContactId": contact_id
            },
            "Emails": []
        }
        
        if email:
            payload["Emails"].append({
                "Value": email,
                "Rank": 1,
                "Description": "Work" # Optional description
            })

        try:
            logger.info(f"Attempting to create person: {first_name} {last_name} for Contact ID {contact_id}")
            resp = self.session.post(url, headers=self._get_headers(), json=payload)
            resp.raise_for_status()
            created_person = resp.json()
            logger.info(f"Successfully created person: {created_person.get('Firstname')} {created_person.get('Lastname')} (ID: {created_person.get('PersonId')})")
            return created_person
        except Exception as e:
            logger.error(f"Error creating person: {e}")
            if hasattr(e, 'response') and e.response is not None:
                logger.error(f"Response: {e.response.text}")
            return None