126 lines
4.8 KiB
Python
126 lines
4.8 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 ---
|
|
PROG_ID_CONTACT_CHALLENGE = "SuperOffice:6"
|
|
PROG_ID_PERSON_SUBJECT = "SuperOffice:5"
|
|
PROG_ID_PERSON_INTRO = "SuperOffice:6"
|
|
PROG_ID_PERSON_PROOF = "SuperOffice:7"
|
|
# Annahme: Das sind die ProgIds der Felder, die die IDs speichern
|
|
PROG_ID_CONTACT_VERTICAL = "SuperOffice:5"
|
|
PROG_ID_PERSON_ROLE = "SuperOffice:3" # KORRIGIERT
|
|
|
|
# --- TARGET DATA ---
|
|
TARGET_PERSON_ID = 1
|
|
TARGET_CONTACT_ID = 2
|
|
|
|
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"}
|
|
resp = requests.post(url, data=data)
|
|
return resp.json().get("access_token") if resp.status_code == 200 else 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 get_entity_data(entity, entity_id, token):
|
|
url = f"{BASE_URL}/{entity}/{entity_id}"
|
|
headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
|
|
resp = requests.get(url, headers=headers)
|
|
return resp.json() if resp.status_code == 200 else 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"}
|
|
|
|
existing_data = get_entity_data(entity, entity_id, token)
|
|
if not existing_data:
|
|
print(f"❌ ERROR fetching {entity} {entity_id}")
|
|
return False
|
|
|
|
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 (LOGIC CORRECTED) ---")
|
|
|
|
# 1. Get API Token
|
|
access_token = get_token()
|
|
if not access_token:
|
|
print("❌ ERROR: Could not get SuperOffice token. Aborting.")
|
|
exit()
|
|
|
|
# 2. Get real data from SuperOffice
|
|
print(f"Fetching data for Person {TARGET_PERSON_ID} and Contact {TARGET_CONTACT_ID}...")
|
|
person_data = get_entity_data("Person", TARGET_PERSON_ID, access_token)
|
|
contact_data = get_entity_data("Contact", TARGET_CONTACT_ID, access_token)
|
|
|
|
if not person_data or not contact_data:
|
|
print("❌ ERROR: Could not fetch test entities. Aborting.")
|
|
exit()
|
|
|
|
# Extract and CLEAN the IDs from the UDFs
|
|
try:
|
|
vertical_id_raw = contact_data["UserDefinedFields"][PROG_ID_CONTACT_VERTICAL]
|
|
role_id_raw = person_data["UserDefinedFields"][PROG_ID_PERSON_ROLE]
|
|
|
|
# Clean the "[I:xx]" format to a pure integer
|
|
vertical_id = int(vertical_id_raw.replace("[I:", "").replace("]", ""))
|
|
role_id = int(role_id_raw.replace("[I:", "").replace("]", ""))
|
|
|
|
print(f"Detected Vertical ID: {vertical_id} (Raw: {vertical_id_raw}), Role ID: {role_id} (Raw: {role_id_raw})")
|
|
except KeyError as e:
|
|
print(f"❌ ERROR: A ProgId is wrong or the field is empty: {e}. Aborting.")
|
|
exit()
|
|
|
|
# 3. Get Text from local DB
|
|
subject, intro, proof = get_text_from_matrix(vertical_id, role_id)
|
|
if not subject:
|
|
print(f"❌ ERROR: Could not find matching text for V_ID:{vertical_id}, R_ID:{role_id} in local DB. Aborting.")
|
|
exit()
|
|
|
|
print(f"Found texts for V_ID:{vertical_id}, R_ID:{role_id}")
|
|
|
|
# 4. Prepare Payloads
|
|
contact_payload = {
|
|
PROG_ID_CONTACT_CHALLENGE: intro
|
|
}
|
|
person_payload = {
|
|
PROG_ID_PERSON_SUBJECT: subject,
|
|
PROG_ID_PERSON_INTRO: intro,
|
|
PROG_ID_PERSON_PROOF: proof
|
|
}
|
|
|
|
# 5. Inject data
|
|
update_udfs("Contact", TARGET_CONTACT_ID, contact_payload, access_token)
|
|
update_udfs("Person", TARGET_PERSON_ID, person_payload, access_token)
|
|
|
|
print("\n--- Test complete ---")
|