Finalize SuperOffice production migration and multi-campaign architecture (v1.8)
This commit is contained in:
@@ -3,6 +3,7 @@ import logging
|
||||
import os
|
||||
import requests
|
||||
import json
|
||||
from datetime import datetime
|
||||
from queue_manager import JobQueue
|
||||
from superoffice_client import SuperOfficeClient
|
||||
from config import settings
|
||||
@@ -21,16 +22,15 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
"""
|
||||
Core logic for processing a single job.
|
||||
"""
|
||||
logger.info(f"Processing Job {job['id']} ({job['event_type']})")
|
||||
logger.info(f"--- [WORKER v1.8] Processing Job {job['id']} ({job['event_type']}) ---")
|
||||
payload = job['payload']
|
||||
event_low = job['event_type'].lower()
|
||||
|
||||
# 1. Extract IDs Early (Crucial for logging and logic)
|
||||
# 1. Extract IDs Early
|
||||
person_id = None
|
||||
contact_id = None
|
||||
job_title = payload.get("JobTitle")
|
||||
|
||||
# Try getting IDs from FieldValues (more reliable for Webhooks)
|
||||
field_values = payload.get("FieldValues", {})
|
||||
if "person_id" in field_values:
|
||||
person_id = int(field_values["person_id"])
|
||||
@@ -39,7 +39,6 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
if "title" in field_values and not job_title:
|
||||
job_title = field_values["title"]
|
||||
|
||||
# Fallback to older payload structure if not found
|
||||
if not person_id:
|
||||
if "PersonId" in payload:
|
||||
person_id = int(payload["PersonId"])
|
||||
@@ -53,7 +52,6 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
contact_id = int(payload["PrimaryKey"])
|
||||
|
||||
# Fallback/Deep Lookup & Fetch JobTitle if missing
|
||||
# Only fetch if we are missing critical info AND have a person_id
|
||||
if person_id and (not job_title or not contact_id):
|
||||
try:
|
||||
person_details = so_client.get_person(
|
||||
@@ -63,15 +61,12 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
if person_details:
|
||||
if not job_title:
|
||||
job_title = person_details.get("JobTitle") or person_details.get("Title")
|
||||
|
||||
# Robust extraction of ContactId
|
||||
if not contact_id:
|
||||
contact_obj = person_details.get("Contact")
|
||||
if contact_obj and isinstance(contact_obj, dict):
|
||||
contact_id = contact_obj.get("ContactId")
|
||||
elif "ContactId" in person_details: # Sometimes flat
|
||||
elif "ContactId" in person_details:
|
||||
contact_id = person_details.get("ContactId")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to fetch person details for {person_id}: {e}")
|
||||
|
||||
@@ -80,61 +75,17 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
logger.info(f"Skipping irrelevant event type: {job['event_type']}")
|
||||
return "SUCCESS"
|
||||
|
||||
changes = [c.lower() for c in payload.get("Changes", [])]
|
||||
if changes:
|
||||
relevant_contact = ["name", "department", "urladdress", "number1", "number2", "userdefinedfields"]
|
||||
if settings.UDF_VERTICAL:
|
||||
relevant_contact.append(settings.UDF_VERTICAL.lower())
|
||||
|
||||
relevant_person = ["jobtitle", "position", "title", "userdefinedfields", "person_id"]
|
||||
technical_fields = ["updated", "updated_associate_id", "contact_id", "person_id", "registered", "registered_associate_id"]
|
||||
actual_changes = [c for c in changes if c not in technical_fields]
|
||||
|
||||
is_relevant = False
|
||||
|
||||
if "contact" in event_low:
|
||||
logger.info(f"Checking relevance for Contact {contact_id or 'Unknown'}. Changes: {actual_changes}")
|
||||
if any(f in actual_changes for f in relevant_contact):
|
||||
is_relevant = True
|
||||
elif "urls" in actual_changes:
|
||||
is_relevant = True
|
||||
|
||||
if "person" in event_low:
|
||||
logger.info(f"Checking relevance for Person {person_id or 'Unknown'}. Changes: {actual_changes}")
|
||||
if any(f in actual_changes for f in relevant_person):
|
||||
is_relevant = True
|
||||
|
||||
if not is_relevant:
|
||||
logger.info(f"Skipping technical/irrelevant changes: {changes}")
|
||||
return "SUCCESS"
|
||||
else:
|
||||
logger.info("Change is deemed RELEVANT. Proceeding...")
|
||||
|
||||
if not contact_id:
|
||||
raise ValueError(f"Could not identify ContactId in payload: {payload}")
|
||||
|
||||
logger.info(f"Target Identified -> Person: {person_id}, Contact: {contact_id}, JobTitle: {job_title}")
|
||||
|
||||
# --- Cascading Logic ---
|
||||
if "contact" in event_low and not person_id:
|
||||
logger.info(f"Company event detected. Triggering cascade for all persons of Contact {contact_id}.")
|
||||
try:
|
||||
persons = so_client.search(f"Person?$select=PersonId&$filter=contact/contactId eq {contact_id}")
|
||||
if persons:
|
||||
q = JobQueue()
|
||||
for p in persons:
|
||||
p_id = p.get("PersonId")
|
||||
if p_id:
|
||||
logger.info(f"Cascading: Enqueueing job for Person {p_id}")
|
||||
q.add_job("person.changed", {"PersonId": p_id, "ContactId": contact_id, "Source": "Cascade"})
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to cascade to persons for contact {contact_id}: {e}")
|
||||
|
||||
# 1b. Fetch full contact details for 'Double Truth' check (Master Data Sync)
|
||||
# 1b. Fetch full contact details for 'Double Truth' check
|
||||
crm_name = None
|
||||
crm_website = None
|
||||
crm_industry_name = None
|
||||
contact_details = None
|
||||
campaign_tag = None
|
||||
|
||||
try:
|
||||
contact_details = so_client.get_contact(
|
||||
@@ -146,31 +97,46 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
|
||||
crm_name = contact_details.get("Name")
|
||||
crm_website = contact_details.get("UrlAddress")
|
||||
if not crm_website and "Urls" in contact_details and contact_details["Urls"]:
|
||||
crm_website = contact_details["Urls"][0].get("Value")
|
||||
|
||||
# --- Fetch Person UDFs for Campaign Tag ---
|
||||
if person_id:
|
||||
try:
|
||||
# We fetch the person again specifically for UDFs to ensure we get DisplayTexts
|
||||
person_details = so_client.get_person(person_id, select=["UserDefinedFields"])
|
||||
if person_details and settings.UDF_CAMPAIGN:
|
||||
udfs = person_details.get("UserDefinedFields", {})
|
||||
# SuperOffice REST returns DisplayText for lists as 'ProgID:DisplayText'
|
||||
display_key = f"{settings.UDF_CAMPAIGN}:DisplayText"
|
||||
campaign_tag = udfs.get(display_key)
|
||||
|
||||
if not campaign_tag:
|
||||
# Fallback to manual resolution if DisplayText is missing
|
||||
raw_tag = udfs.get(settings.UDF_CAMPAIGN, "")
|
||||
if raw_tag:
|
||||
campaign_tag = str(raw_tag).strip()
|
||||
|
||||
if campaign_tag:
|
||||
logger.info(f"🎯 CAMPAIGN DETECTED: '{campaign_tag}'")
|
||||
else:
|
||||
logger.info("ℹ️ No Campaign Tag found (Field is empty).")
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not fetch campaign tag: {e}")
|
||||
|
||||
if settings.UDF_VERTICAL:
|
||||
udfs = contact_details.get("UserDefinedFields", {})
|
||||
so_vertical_val = udfs.get(settings.UDF_VERTICAL)
|
||||
|
||||
if so_vertical_val:
|
||||
val_str = str(so_vertical_val)
|
||||
if val_str.startswith("[I:"):
|
||||
val_str = val_str.split(":")[1].strip("]")
|
||||
|
||||
val_str = str(so_vertical_val).replace("[I:","").replace("]","")
|
||||
try:
|
||||
vertical_map = json.loads(settings.VERTICAL_MAP_JSON)
|
||||
vertical_map_rev = {str(v): k for k, v in vertical_map.items()}
|
||||
if val_str in vertical_map_rev:
|
||||
crm_industry_name = vertical_map_rev[val_str]
|
||||
logger.info(f"Detected CRM Vertical Override: {so_vertical_val} -> {crm_industry_name}")
|
||||
except Exception as ex:
|
||||
logger.error(f"Error mapping vertical ID {val_str}: {ex}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch contact details for {contact_id}: {e}")
|
||||
# Critical failure: Without contact details, we cannot provision correctly.
|
||||
# Raising exception triggers a retry.
|
||||
raise Exception(f"SuperOffice API Failure: {e}")
|
||||
|
||||
# --- 3. Company Explorer Provisioning ---
|
||||
@@ -181,7 +147,8 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
"job_title": job_title,
|
||||
"crm_name": crm_name,
|
||||
"crm_website": crm_website,
|
||||
"crm_industry_name": crm_industry_name
|
||||
"crm_industry_name": crm_industry_name,
|
||||
"campaign_tag": campaign_tag
|
||||
}
|
||||
|
||||
ce_auth = (os.getenv("API_USER", "admin"), os.getenv("API_PASSWORD", "gemini"))
|
||||
@@ -189,34 +156,20 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
try:
|
||||
resp = requests.post(ce_url, json=ce_req, auth=ce_auth)
|
||||
if resp.status_code == 404:
|
||||
logger.warning(f"Company Explorer returned 404. Retrying later.")
|
||||
return "RETRY"
|
||||
|
||||
resp.raise_for_status()
|
||||
provisioning_data = resp.json()
|
||||
|
||||
if provisioning_data.get("status") == "processing":
|
||||
logger.info(f"Company Explorer is processing {provisioning_data.get('company_name', 'Unknown')}. Re-queueing job.")
|
||||
return "RETRY"
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"Company Explorer API failed: {e}")
|
||||
|
||||
logger.info(f"CE Response for Contact {contact_id}: {json.dumps(provisioning_data)}")
|
||||
|
||||
# Fetch fresh Contact Data for comparison
|
||||
try:
|
||||
contact_data = so_client.get_contact(contact_id)
|
||||
if not contact_data:
|
||||
logger.error(f"Contact {contact_id} not found in SuperOffice.")
|
||||
return "FAILED"
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch contact {contact_id}: {e}")
|
||||
return "RETRY"
|
||||
|
||||
contact_data = so_client.get_contact(contact_id)
|
||||
if not contact_data: return "FAILED"
|
||||
contact_patch = {}
|
||||
|
||||
if "UserDefinedFields" not in contact_data: contact_data["UserDefinedFields"] = {}
|
||||
|
||||
# --- A. Vertical Sync ---
|
||||
vertical_name = provisioning_data.get("vertical_name")
|
||||
@@ -226,13 +179,11 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
vertical_id = vertical_map.get(vertical_name)
|
||||
if vertical_id:
|
||||
udf_key = settings.UDF_VERTICAL
|
||||
current_val = contact_data["UserDefinedFields"].get(udf_key, "")
|
||||
if current_val and str(current_val).startswith("[I:"):
|
||||
current_val = str(current_val).split(":")[1].strip("]")
|
||||
|
||||
if str(current_val) != str(vertical_id):
|
||||
logger.info(f"Change detected: Vertical {current_val} -> {vertical_id}")
|
||||
contact_patch.setdefault("UserDefinedFields", {})[udf_key] = str(vertical_id)
|
||||
current_val = contact_data.get("UserDefinedFields", {}).get(udf_key, "")
|
||||
if str(current_val).replace("[I:","").replace("]","") != str(vertical_id):
|
||||
logger.info(f"Change detected: Vertical -> {vertical_id}")
|
||||
if "UserDefinedFields" not in contact_patch: contact_patch["UserDefinedFields"] = {}
|
||||
contact_patch["UserDefinedFields"][udf_key] = str(vertical_id)
|
||||
except Exception as e:
|
||||
logger.error(f"Vertical sync error: {e}")
|
||||
|
||||
@@ -243,154 +194,92 @@ def process_job(job, so_client: SuperOfficeClient):
|
||||
ce_vat = provisioning_data.get("vat_id")
|
||||
|
||||
if ce_city or ce_street or ce_zip:
|
||||
if "Address" not in contact_data or contact_data["Address"] is None:
|
||||
contact_data["Address"] = {"Street": {}, "Postal": {}}
|
||||
|
||||
addr_obj = contact_data["Address"]
|
||||
if "Postal" not in addr_obj or addr_obj["Postal"] is None: addr_obj["Postal"] = {}
|
||||
if "Street" not in addr_obj or addr_obj["Street"] is None: addr_obj["Street"] = {}
|
||||
for type_key in ["Postal", "Street"]:
|
||||
cur_addr = (contact_data.get("Address") or {}).get(type_key, {})
|
||||
if ce_city and cur_addr.get("City") != ce_city: contact_patch.setdefault("Address", {}).setdefault(type_key, {})["City"] = ce_city
|
||||
if ce_street and cur_addr.get("Address1") != ce_street: contact_patch.setdefault("Address", {}).setdefault(type_key, {})["Address1"] = ce_street
|
||||
if ce_zip and cur_addr.get("Zipcode") != ce_zip: contact_patch.setdefault("Address", {}).setdefault(type_key, {})["Zipcode"] = ce_zip
|
||||
|
||||
def update_addr_patch(field_name, new_val, log_name):
|
||||
if new_val:
|
||||
for type_key in ["Postal", "Street"]:
|
||||
cur = addr_obj[type_key].get(field_name, "")
|
||||
if cur != new_val:
|
||||
logger.info(f"Change detected: {type_key} {log_name} '{cur}' -> '{new_val}'")
|
||||
contact_patch.setdefault("Address", {}).setdefault(type_key, {})[field_name] = new_val
|
||||
if ce_vat and contact_data.get("OrgNr") != ce_vat:
|
||||
contact_patch["OrgNr"] = ce_vat
|
||||
|
||||
update_addr_patch("City", ce_city, "City")
|
||||
update_addr_patch("Address1", ce_street, "Street")
|
||||
update_addr_patch("Zipcode", ce_zip, "Zip")
|
||||
|
||||
if ce_vat:
|
||||
current_vat = contact_data.get("OrgNr", "")
|
||||
if current_vat != ce_vat:
|
||||
logger.info(f"Change detected: VAT '{current_vat}' -> '{ce_vat}'")
|
||||
contact_patch["OrgNr"] = ce_vat
|
||||
|
||||
# --- C. AI Openers Sync ---
|
||||
# --- C. AI Openers & Summary Sync ---
|
||||
ce_opener = provisioning_data.get("opener")
|
||||
ce_opener_secondary = provisioning_data.get("opener_secondary")
|
||||
ce_summary = provisioning_data.get("summary")
|
||||
|
||||
if ce_opener and ce_opener != "null":
|
||||
current_opener = contact_data["UserDefinedFields"].get(settings.UDF_OPENER, "")
|
||||
if current_opener != ce_opener:
|
||||
logger.info("Change detected: Primary Opener")
|
||||
contact_patch.setdefault("UserDefinedFields", {})[settings.UDF_OPENER] = ce_opener
|
||||
if ce_opener and ce_opener != "null" and contact_data.get("UserDefinedFields", {}).get(settings.UDF_OPENER) != ce_opener:
|
||||
if "UserDefinedFields" not in contact_patch: contact_patch["UserDefinedFields"] = {}
|
||||
contact_patch["UserDefinedFields"][settings.UDF_OPENER] = ce_opener
|
||||
if ce_opener_secondary and ce_opener_secondary != "null" and contact_data.get("UserDefinedFields", {}).get(settings.UDF_OPENER_SECONDARY) != ce_opener_secondary:
|
||||
if "UserDefinedFields" not in contact_patch: contact_patch["UserDefinedFields"] = {}
|
||||
contact_patch["UserDefinedFields"][settings.UDF_OPENER_SECONDARY] = ce_opener_secondary
|
||||
|
||||
if ce_opener_secondary and ce_opener_secondary != "null":
|
||||
current_opener_sec = contact_data["UserDefinedFields"].get(settings.UDF_OPENER_SECONDARY, "")
|
||||
if current_opener_sec != ce_opener_secondary:
|
||||
logger.info("Change detected: Secondary Opener")
|
||||
contact_patch.setdefault("UserDefinedFields", {})[settings.UDF_OPENER_SECONDARY] = ce_opener_secondary
|
||||
if ce_summary and ce_summary != "null":
|
||||
short_summary = (ce_summary[:132] + "...") if len(ce_summary) > 135 else ce_summary
|
||||
if contact_data.get("UserDefinedFields", {}).get(settings.UDF_SUMMARY) != short_summary:
|
||||
logger.info("Change detected: AI Summary")
|
||||
if "UserDefinedFields" not in contact_patch: contact_patch["UserDefinedFields"] = {}
|
||||
contact_patch["UserDefinedFields"][settings.UDF_SUMMARY] = short_summary
|
||||
|
||||
# --- D. Apply Updates (Single PATCH Transaction) ---
|
||||
# --- D. Timestamps & Website Sync ---
|
||||
if settings.UDF_LAST_UPDATE:
|
||||
now_so = f"[D:{datetime.now().strftime('%m/%d/%Y %H:%M:%S')}]"
|
||||
if "UserDefinedFields" not in contact_patch: contact_patch["UserDefinedFields"] = {}
|
||||
contact_patch["UserDefinedFields"][settings.UDF_LAST_UPDATE] = now_so
|
||||
|
||||
ce_website = provisioning_data.get("website")
|
||||
if ce_website and (not contact_data.get("Urls") or settings.ENABLE_WEBSITE_SYNC):
|
||||
current_urls = contact_data.get("Urls") or []
|
||||
if not any(u.get("Value") == ce_website for u in current_urls):
|
||||
logger.info(f"Syncing Website: {ce_website}")
|
||||
if "Urls" not in contact_patch: contact_patch["Urls"] = []
|
||||
contact_patch["Urls"] = [{"Value": ce_website, "Description": "AI Discovered"}] + current_urls
|
||||
|
||||
# --- E. Apply Updates (Single PATCH) ---
|
||||
if contact_patch:
|
||||
logger.info(f"Pushing combined PATCH updates for Contact {contact_id}...")
|
||||
try:
|
||||
so_client.patch_contact(contact_id, contact_patch)
|
||||
logger.info("✅ Contact Update Successful (PATCH).")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update Contact {contact_id}: {e}")
|
||||
raise
|
||||
else:
|
||||
logger.info(f"No changes detected for Contact {contact_id}. Skipping update.")
|
||||
logger.info(f"Pushing combined PATCH for Contact {contact_id}: {list(contact_patch.keys())}")
|
||||
so_client.patch_contact(contact_id, contact_patch)
|
||||
logger.info("✅ Contact Update Successful.")
|
||||
|
||||
# 2d. Sync Person Position (Role) - if Person exists
|
||||
# 2d. Sync Person Position
|
||||
role_name = provisioning_data.get("role_name")
|
||||
if person_id and role_name:
|
||||
try:
|
||||
persona_map = json.loads(settings.PERSONA_MAP_JSON)
|
||||
position_id = persona_map.get(role_name)
|
||||
if position_id:
|
||||
logger.info(f"Identified Role '{role_name}' -> Position ID {position_id}")
|
||||
so_client.update_person_position(person_id, int(position_id))
|
||||
except Exception as e:
|
||||
logger.error(f"Error syncing position for Person {person_id}: {e}")
|
||||
logger.error(f"Error syncing position: {e}")
|
||||
|
||||
# 3. Update SuperOffice Texts (Only if person_id is present)
|
||||
if not person_id:
|
||||
logger.info("Sync complete (Company only).")
|
||||
return "SUCCESS"
|
||||
|
||||
texts = provisioning_data.get("texts", {})
|
||||
if not any(texts.values()):
|
||||
logger.info("No texts returned from Matrix yet.")
|
||||
return "SUCCESS"
|
||||
|
||||
udf_update = {}
|
||||
if texts.get("subject"): udf_update[settings.UDF_SUBJECT] = texts["subject"]
|
||||
if texts.get("intro"): udf_update[settings.UDF_INTRO] = texts["intro"]
|
||||
if texts.get("social_proof"): udf_update[settings.UDF_SOCIAL_PROOF] = texts["social_proof"]
|
||||
|
||||
if udf_update:
|
||||
try:
|
||||
current_person = so_client.get_person(person_id)
|
||||
current_udfs = current_person.get("UserDefinedFields", {})
|
||||
needs_update = False
|
||||
for key, new_val in udf_update.items():
|
||||
if current_udfs.get(key, "") != new_val:
|
||||
needs_update = True
|
||||
break
|
||||
# 3. Update SuperOffice Texts (Person)
|
||||
if person_id:
|
||||
texts = provisioning_data.get("texts", {})
|
||||
unsubscribe_link = provisioning_data.get("unsubscribe_link")
|
||||
|
||||
udf_update = {}
|
||||
if texts.get("subject"): udf_update[settings.UDF_SUBJECT] = texts["subject"]
|
||||
if texts.get("intro"): udf_update[settings.UDF_INTRO] = texts["intro"]
|
||||
if texts.get("social_proof"): udf_update[settings.UDF_SOCIAL_PROOF] = texts["social_proof"]
|
||||
if unsubscribe_link and settings.UDF_UNSUBSCRIBE_LINK:
|
||||
udf_update[settings.UDF_UNSUBSCRIBE_LINK] = unsubscribe_link
|
||||
|
||||
if udf_update:
|
||||
logger.info(f"Applying text update to Person {person_id}.")
|
||||
so_client.update_entity_udfs(person_id, "Person", udf_update)
|
||||
|
||||
# Simulation Trigger: Either texts changed, OR it's a direct manual trigger
|
||||
if needs_update or (person_id and not "Source" in payload):
|
||||
if needs_update:
|
||||
logger.info(f"Applying text update to Person {person_id}.")
|
||||
so_client.update_entity_udfs(person_id, "Person", udf_update)
|
||||
else:
|
||||
logger.info(f"Texts already in sync for Person {person_id}, but triggering simulation.")
|
||||
|
||||
# --- 4. Create Email Simulation Appointment ---
|
||||
try:
|
||||
opener = provisioning_data.get("opener", "")
|
||||
intro = texts.get("intro", "")
|
||||
proof = texts.get("social_proof", "")
|
||||
subject = texts.get("subject", "No Subject")
|
||||
|
||||
salutation = "Hallo"
|
||||
p_data = so_client.get_person(person_id)
|
||||
if p_data:
|
||||
fname = p_data.get("Firstname", "")
|
||||
lname = p_data.get("Lastname", "")
|
||||
if fname or lname:
|
||||
salutation = f"Hallo {fname} {lname}".strip()
|
||||
# --- 4. Create Email Simulation Appointment ---
|
||||
try:
|
||||
opener = provisioning_data.get("opener") or ""
|
||||
intro = texts.get("intro") or ""
|
||||
proof = texts.get("social_proof") or ""
|
||||
subject = texts.get("subject", "No Subject")
|
||||
email_body = f"Betreff: {subject}\n\n{opener}\n\n{intro}\n\n{proof}\n\n(Generated via Gemini Marketing Engine)"
|
||||
so_client.create_appointment(f"KI: {subject}", email_body, contact_id, person_id)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed simulation: {e}")
|
||||
|
||||
cta = (
|
||||
"H\u00e4tten Sie am kommenden Mittwoch gegen 11 Uhr kurz Zeit, f\u00fcr einen kurzen Austausch hierzu?\n"
|
||||
"Gerne k\u00f6nnen Sie auch einen alternativen Termin in meinem Kalender buchen. (bookings Link)"
|
||||
)
|
||||
|
||||
email_body = (
|
||||
f"{salutation},\n\n"
|
||||
f"{opener.strip()}\n\n"
|
||||
f"{intro.strip()}\n\n"
|
||||
f"{cta.strip()}\n\n"
|
||||
f"{proof.strip()}\n\n"
|
||||
"(Generated via Gemini Marketing Engine)"
|
||||
)
|
||||
|
||||
from datetime import datetime
|
||||
now_str = datetime.now().strftime("%H:%M")
|
||||
appt_title = f"[{now_str}] KI: {subject}"
|
||||
|
||||
so_client.create_appointment(
|
||||
subject=appt_title,
|
||||
description=email_body,
|
||||
contact_id=contact_id,
|
||||
person_id=person_id
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create email simulation appointment: {e}")
|
||||
|
||||
else:
|
||||
logger.info(f"Skipping update for Person {person_id}: Values match.")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error during Person update: {e}")
|
||||
raise
|
||||
|
||||
logger.info("Job successfully processed.")
|
||||
return "SUCCESS"
|
||||
|
||||
def run_worker():
|
||||
@@ -401,7 +290,7 @@ def run_worker():
|
||||
so_client = SuperOfficeClient()
|
||||
if not so_client.access_token: raise Exception("Auth failed")
|
||||
except Exception as e:
|
||||
logger.critical(f"Failed to initialize SuperOffice Client: {e}. Retrying in 30s...")
|
||||
logger.critical(f"Failed to initialize SO Client. Retrying in 30s...")
|
||||
time.sleep(30)
|
||||
|
||||
logger.info("Worker started. Polling queue...")
|
||||
@@ -411,12 +300,9 @@ def run_worker():
|
||||
if job:
|
||||
try:
|
||||
result = process_job(job, so_client)
|
||||
if result == "RETRY":
|
||||
queue.retry_job_later(job['id'], delay_seconds=120, error_msg="CE is processing...")
|
||||
elif result == "FAILED":
|
||||
queue.fail_job(job['id'], "Job failed with FAILED status")
|
||||
else:
|
||||
queue.complete_job(job['id'])
|
||||
if result == "RETRY": queue.retry_job_later(job['id'], delay_seconds=120)
|
||||
elif result == "FAILED": queue.fail_job(job['id'], "Job failed status")
|
||||
else: queue.complete_job(job['id'])
|
||||
except Exception as e:
|
||||
logger.error(f"Job {job['id']} failed: {e}", exc_info=True)
|
||||
queue.fail_job(job['id'], str(e))
|
||||
@@ -427,4 +313,4 @@ def run_worker():
|
||||
time.sleep(POLL_INTERVAL)
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_worker()
|
||||
run_worker()
|
||||
Reference in New Issue
Block a user