109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
import os
|
|
import sqlite3
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(override=True)
|
|
|
|
# --- CONFIGURATION ---
|
|
DB_FILE = "marketing_matrix.db"
|
|
SO_CLIENT_ID = os.getenv("SO_CLIENT_ID") or os.getenv("SO_SOD")
|
|
SO_CLIENT_SECRET = os.getenv("SO_CLIENT_SECRET")
|
|
SO_REFRESH_TOKEN = os.getenv("SO_REFRESH_TOKEN")
|
|
BASE_URL = "https://app-sod.superoffice.com/Cust55774/api/v1"
|
|
|
|
# --- SUPEROFFICE UDF ProgIds (from your discovery) ---
|
|
PROG_ID_CONTACT_CHALLENGE = "SuperOffice:6"
|
|
PROG_ID_PERSON_SUBJECT = "SuperOffice:5"
|
|
PROG_ID_PERSON_INTRO = "SuperOffice:6"
|
|
PROG_ID_PERSON_PROOF = "SuperOffice:7"
|
|
|
|
# --- TEST DATA ---
|
|
TEST_PERSON_ID = 1
|
|
TEST_CONTACT_ID = 2
|
|
TEST_VERTICAL_ID = 24 # Healthcare - Hospital
|
|
TEST_ROLE_ID = 19 # Operativer Entscheider
|
|
|
|
def get_token():
|
|
url = "https://sod.superoffice.com/login/common/oauth/tokens"
|
|
data = {"grant_type": "refresh_token", "client_id": SO_CLIENT_ID, "client_secret": SO_CLIENT_SECRET, "refresh_token": SO_REFRESH_TOKEN, "redirect_uri": "http://localhost"}
|
|
try:
|
|
resp = requests.post(url, data=data)
|
|
if resp.status_code == 200:
|
|
return resp.json().get("access_token")
|
|
else:
|
|
print(f"Token Error: {resp.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Connection Error: {e}")
|
|
return None
|
|
|
|
def get_text_from_matrix(vertical_id, role_id):
|
|
conn = sqlite3.connect(DB_FILE)
|
|
c = conn.cursor()
|
|
c.execute("SELECT subject, intro, social_proof FROM text_blocks WHERE vertical_id = ? AND role_id = ?", (vertical_id, role_id))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
return row if row else (None, None, None)
|
|
|
|
def update_udfs(entity, entity_id, payload, token):
|
|
url = f"{BASE_URL}/{entity}/{entity_id}"
|
|
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json"}
|
|
|
|
# SuperOffice expects the full JSON body, not just the UDF part for PUT
|
|
# First, GET the existing entity
|
|
get_resp = requests.get(url, headers=headers)
|
|
if get_resp.status_code != 200:
|
|
print(f"❌ ERROR fetching {entity} {entity_id}: {get_resp.text}")
|
|
return False
|
|
|
|
existing_data = get_resp.json()
|
|
|
|
# Merge UDFs
|
|
if "UserDefinedFields" not in existing_data:
|
|
existing_data["UserDefinedFields"] = {}
|
|
existing_data["UserDefinedFields"].update(payload)
|
|
|
|
print(f"Updating {entity} {entity_id} with new UDFs...")
|
|
put_resp = requests.put(url, headers=headers, json=existing_data)
|
|
|
|
if put_resp.status_code == 200:
|
|
print(f"✅ SUCCESS: Updated {entity} {entity_id}")
|
|
return True
|
|
else:
|
|
print(f"❌ ERROR updating {entity} {entity_id}: {put_resp.status_code} - {put_resp.text}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
print("--- Starting SuperOffice Injection Test ---")
|
|
|
|
# 1. Get Text from local DB
|
|
subject, intro, proof = get_text_from_matrix(TEST_VERTICAL_ID, TEST_ROLE_ID)
|
|
if not subject:
|
|
print("❌ ERROR: Could not find matching text in local DB. Aborting.")
|
|
exit()
|
|
|
|
print(f"Found texts for V_ID:{TEST_VERTICAL_ID}, R_ID:{TEST_ROLE_ID}")
|
|
|
|
# 2. Get API Token
|
|
access_token = get_token()
|
|
if not access_token:
|
|
print("❌ ERROR: Could not get SuperOffice token. Aborting.")
|
|
exit()
|
|
|
|
# 3. Prepare Payloads
|
|
contact_payload = {
|
|
PROG_ID_CONTACT_CHALLENGE: intro # Using intro for challenge in this demo
|
|
}
|
|
person_payload = {
|
|
PROG_ID_PERSON_SUBJECT: subject,
|
|
PROG_ID_PERSON_INTRO: intro,
|
|
PROG_ID_PERSON_PROOF: proof
|
|
}
|
|
|
|
# 4. Inject data
|
|
update_udfs("Contact", TEST_CONTACT_ID, contact_payload, access_token)
|
|
update_udfs("Person", TEST_PERSON_ID, person_payload, access_token)
|
|
|
|
print("\n--- Test complete ---")
|