import os import sys import json import logging import requests from datetime import datetime, timedelta from dotenv import load_dotenv # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger("create_sale_test") # --- Inline AuthHandler & SuperOfficeClient (Proven Working Logic) --- class AuthHandler: def __init__(self): # CRITICAL: override=True ensures we read from .env even if env vars are already set load_dotenv(override=True) self.client_id = os.getenv("SO_CLIENT_ID") or os.getenv("SO_SOD") self.client_secret = os.getenv("SO_CLIENT_SECRET") self.refresh_token = os.getenv("SO_REFRESH_TOKEN") self.redirect_uri = os.getenv("SO_REDIRECT_URI", "http://localhost") self.env = os.getenv("SO_ENVIRONMENT", "sod") self.cust_id = os.getenv("SO_CONTEXT_IDENTIFIER", "Cust55774") if not all([self.client_id, self.client_secret, self.refresh_token]): raise ValueError("SuperOffice credentials missing in .env file.") def get_access_token(self): return self._refresh_access_token() def _refresh_access_token(self): token_domain = "online.superoffice.com" if "online" in self.env.lower() else "sod.superoffice.com" url = f"https://{token_domain}/login/common/oauth/tokens" data = { "grant_type": "refresh_token", "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "redirect_uri": self.redirect_uri } try: resp = requests.post(url, data=data) if resp.status_code != 200: logger.error(f"❌ Token Refresh Failed (Status {resp.status_code}): {resp.text}") return None return resp.json().get("access_token") except Exception as e: logger.error(f"❌ Connection Error during token refresh: {e}") return None class SuperOfficeClient: def __init__(self, auth_handler): self.auth_handler = auth_handler self.env = auth_handler.env self.cust_id = auth_handler.cust_id self.base_url = f"https://{self.env}.superoffice.com/{self.cust_id}/api/v1" self.access_token = self.auth_handler.get_access_token() if not self.access_token: raise Exception("Failed to obtain access token.") self.headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", "Accept": "application/json" } def _get(self, endpoint): resp = requests.get(f"{self.base_url}/{endpoint}", headers=self.headers) resp.raise_for_status() return resp.json() def _post(self, endpoint, payload): resp = requests.post(f"{self.base_url}/{endpoint}", headers=self.headers, json=payload) resp.raise_for_status() return resp.json() def main(): try: # Initialize Auth and Client auth = AuthHandler() client = SuperOfficeClient(auth) print("\n--- 0. Pre-Flight: Finding Currency ID for EUR ---") currencies = client._get("List/Currency/Items") eur_id = None for curr in currencies: if curr.get('Name') == 'EUR' or curr.get('value') == 'EUR': # Check keys carefully eur_id = curr.get('Id') print(f"✅ Found EUR Currency ID: {eur_id}") break if not eur_id: print("⚠️ EUR Currency not found. Defaulting to ID 33 (from discovery).") eur_id = 33 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") # 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 # 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: logger.error("⛔ ABORTING: Target company is Wackler Service Group (ID 3). This is the parent company.") return print(f"✅ Found Target Contact: {contact_name} (ID: {contact_id})") print("\n--- 2. Finding a Person (Optional but recommended) ---") persons_endpoint = f"Contact/{contact_id}/Persons?$top=1&$select=PersonId,FirstName,LastName" persons = client._get(persons_endpoint) person_id = None if persons and 'value' in persons and len(persons['value']) > 0: target_person = persons['value'][0] person_id = target_person.get('personId') or target_person.get('PersonId') first_name = target_person.get('firstName') or target_person.get('FirstName') last_name = target_person.get('lastName') or target_person.get('LastName') print(f"✅ Found Target Person: {first_name} {last_name} (ID: {person_id})") else: print("⚠️ No person found for this contact. Creating sale without person link.") print("\n--- 3. Creating Sale (Opportunity) ---") # Calculate Estimated Sale Date (e.g., 30 days from now) sale_date = (datetime.utcnow() + timedelta(days=30)).isoformat() + "Z" # Construct the payload for a new Sale sale_payload = { "Heading": "AI Prospect: Automation Potential Detected", "Description": "Automated opportunity created by Gemini AI based on high automation potential score.\n\nKey Insights:\n- High robot density potential\n- Manual processes identified", "Amount": 5000.0, "Saledate": sale_date, # MANDATORY: Estimated closing date "Currency": { "Id": eur_id }, "SaleType": { "Id": 14 }, # FIXED: 14 = Roboplanet Verkauf "Stage": { "Id": 10 }, # 5% Angebot prospektiv "Contact": { "ContactId": contact_id }, "Source": { "Id": 2 } # Proposal Center } if person_id: sale_payload["Person"] = { "PersonId": person_id } print(f"Payload Preview: {json.dumps(sale_payload, indent=2)}") # Uncomment to actually run creation 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}") 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}") if e.response is not None: logger.error(f"Response Body: {e.response.text}") except Exception as e: logger.error(f"Fatal Error: {e}") if __name__ == "__main__": main()