180 lines
7.5 KiB
Python
180 lines
7.5 KiB
Python
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() |