696 lines
28 KiB
Python
696 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Callable
|
|
|
|
from sqlalchemy import select
|
|
|
|
from services.sales_service import sales_events as sales_event_types
|
|
from services.sales_service.deal_state_machine import DealStateMachineError, transition_deal_stage
|
|
from services.sales_service.event_publisher import SalesEventPublisher
|
|
from services.shared.core import new_id, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.sales_sql_models import (
|
|
SalesAutomationTaskRow,
|
|
SalesChannelSwitchRow,
|
|
SalesCommunicationSessionRow,
|
|
SalesDealRow,
|
|
SalesEscalationRow,
|
|
SalesInvoiceRow,
|
|
SalesPipelineStageRow,
|
|
)
|
|
|
|
|
|
RETRY_DELAYS_SECONDS = {
|
|
1: 60,
|
|
2: 300,
|
|
3: 900,
|
|
}
|
|
ACTIVE_TASK_STATUSES = {"pending", "running"}
|
|
TERMINAL_TASK_STATUSES = {"completed", "failed", "canceled"}
|
|
|
|
|
|
def _now_dt() -> datetime:
|
|
return datetime.now(timezone.utc).replace(microsecond=0)
|
|
|
|
|
|
def _iso(dt: datetime) -> str:
|
|
return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()
|
|
|
|
|
|
def _payload(row: SalesAutomationTaskRow) -> dict[str, Any]:
|
|
try:
|
|
parsed = json.loads(row.payload_json or "{}")
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
|
|
|
|
def _set_payload(row: SalesAutomationTaskRow, payload: dict[str, Any]) -> None:
|
|
row.payload_json = json.dumps(payload, ensure_ascii=False)
|
|
|
|
|
|
def _task_error_message(exc: Exception) -> str:
|
|
message = str(exc).strip()
|
|
return message[:1000] if message else exc.__class__.__name__
|
|
|
|
|
|
class SalesAutomationWorker:
|
|
def __init__(self, *, worker_id: str | None = None, batch_size: int = 10) -> None:
|
|
self.worker_id = worker_id or f"sales-worker-{socket.gethostname()}-{new_id('wrk')}"
|
|
self.batch_size = batch_size
|
|
self.handlers: dict[str, Callable[[Any, SalesAutomationTaskRow], None]] = {
|
|
"follow_up_customer": self._handle_follow_up_customer,
|
|
"mark_invoice_overdue": self._handle_mark_invoice_overdue,
|
|
"send_invoice_reminder": self._handle_send_invoice_reminder,
|
|
"recommend_channel_switch": self._handle_recommend_channel_switch,
|
|
"close_stale_communication": self._handle_close_stale_communication,
|
|
"escalate_to_human": self._handle_escalate_to_human,
|
|
"post_sale_transfer": self._handle_post_sale_transfer,
|
|
}
|
|
|
|
def claim_pending_tasks(self, session, *, limit: int | None = None, now: str | None = None) -> list[SalesAutomationTaskRow]:
|
|
claim_limit = limit or self.batch_size
|
|
now_text = now or utc_now_iso()
|
|
stmt = (
|
|
select(SalesAutomationTaskRow)
|
|
.where(
|
|
SalesAutomationTaskRow.status == "pending",
|
|
SalesAutomationTaskRow.run_at <= now_text,
|
|
)
|
|
.order_by(SalesAutomationTaskRow.run_at.asc(), SalesAutomationTaskRow.id.asc())
|
|
.limit(claim_limit)
|
|
)
|
|
if session.bind is not None and session.bind.dialect.name == "postgresql":
|
|
stmt = stmt.with_for_update(skip_locked=True)
|
|
rows = session.execute(stmt).scalars().all()
|
|
for row in rows:
|
|
row.status = "running"
|
|
row.locked_at = now_text
|
|
row.locked_by = self.worker_id
|
|
row.last_error = None
|
|
row.updated_at = now_text
|
|
session.flush()
|
|
return list(rows)
|
|
|
|
def run_once(self, *, limit: int | None = None) -> int:
|
|
session = get_session()
|
|
try:
|
|
tasks = self.claim_pending_tasks(session, limit=limit)
|
|
task_ids = [task.task_id for task in tasks]
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
for task_id in task_ids:
|
|
self._execute_claimed_task(task_id)
|
|
return len(task_ids)
|
|
|
|
def _execute_claimed_task(self, task_id: str) -> None:
|
|
session = get_session()
|
|
try:
|
|
task = session.execute(
|
|
select(SalesAutomationTaskRow).where(
|
|
SalesAutomationTaskRow.task_id == task_id,
|
|
SalesAutomationTaskRow.status == "running",
|
|
SalesAutomationTaskRow.locked_by == self.worker_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if task is None:
|
|
session.rollback()
|
|
return
|
|
try:
|
|
self._execute_task(session, task)
|
|
except Exception as exc: # noqa: BLE001
|
|
self._mark_failed_or_retry(task, exc)
|
|
else:
|
|
self._mark_completed(task)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
def _execute_task(self, session, task: SalesAutomationTaskRow) -> None:
|
|
handler = self.handlers.get(task.task_type)
|
|
if handler is None:
|
|
raise RuntimeError(f"Unsupported automation task type: {task.task_type}")
|
|
handler(session, task)
|
|
|
|
def _mark_completed(self, task: SalesAutomationTaskRow) -> None:
|
|
now = utc_now_iso()
|
|
task.status = "completed"
|
|
task.completed_at = now
|
|
task.failed_at = None
|
|
task.locked_at = None
|
|
task.locked_by = None
|
|
task.last_error = None
|
|
task.updated_at = now
|
|
|
|
def _mark_failed_or_retry(self, task: SalesAutomationTaskRow, exc: Exception) -> None:
|
|
now_dt = _now_dt()
|
|
task.retry_count += 1
|
|
task.last_error = _task_error_message(exc)
|
|
task.locked_at = None
|
|
task.locked_by = None
|
|
task.updated_at = _iso(now_dt)
|
|
max_retries = int(task.max_retries or 3)
|
|
if task.retry_count < max_retries:
|
|
task.status = "pending"
|
|
delay_seconds = RETRY_DELAYS_SECONDS.get(task.retry_count, 900)
|
|
task.run_at = _iso(now_dt + timedelta(seconds=delay_seconds))
|
|
return
|
|
task.status = "failed"
|
|
task.failed_at = _iso(now_dt)
|
|
|
|
def _get_deal(self, session, task: SalesAutomationTaskRow) -> SalesDealRow:
|
|
row = session.execute(
|
|
select(SalesDealRow).where(
|
|
SalesDealRow.tenant_id == task.tenant_id,
|
|
SalesDealRow.deal_id == task.deal_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
raise RuntimeError(f"Deal not found: {task.deal_id}")
|
|
return row
|
|
|
|
def _get_invoice(self, session, task: SalesAutomationTaskRow, invoice_id: str) -> SalesInvoiceRow:
|
|
row = session.execute(
|
|
select(SalesInvoiceRow).where(
|
|
SalesInvoiceRow.tenant_id == task.tenant_id,
|
|
SalesInvoiceRow.invoice_id == invoice_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
raise RuntimeError(f"Invoice not found: {invoice_id}")
|
|
if row.deal_id != task.deal_id:
|
|
raise RuntimeError("Invoice does not belong to task deal")
|
|
return row
|
|
|
|
def _stage_code(self, session, deal: SalesDealRow) -> str | None:
|
|
stage = session.execute(
|
|
select(SalesPipelineStageRow).where(
|
|
SalesPipelineStageRow.tenant_id == deal.tenant_id,
|
|
SalesPipelineStageRow.stage_id == deal.stage_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
return stage.code if stage is not None else None
|
|
|
|
def _transition_stage(
|
|
self,
|
|
session,
|
|
*,
|
|
task: SalesAutomationTaskRow,
|
|
target_stage_code: str,
|
|
reason: str,
|
|
force_on_invalid: bool = True,
|
|
) -> None:
|
|
metadata = {"automation_task_id": task.task_id, "task_type": task.task_type}
|
|
try:
|
|
transition_deal_stage(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
target_stage_code=target_stage_code,
|
|
actor_type="system",
|
|
reason=reason,
|
|
metadata=metadata,
|
|
)
|
|
except DealStateMachineError as exc:
|
|
if not force_on_invalid or exc.error != "invalid_stage_transition":
|
|
raise
|
|
transition_deal_stage(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
target_stage_code=target_stage_code,
|
|
actor_type="system",
|
|
reason=reason,
|
|
metadata={**metadata, "force_reason": reason},
|
|
force=True,
|
|
)
|
|
|
|
def _publish(
|
|
self,
|
|
session,
|
|
*,
|
|
task: SalesAutomationTaskRow,
|
|
event_type: str,
|
|
aggregate_type: str,
|
|
aggregate_id: str,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
SalesEventPublisher.publish_sales_event(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
event_type=event_type,
|
|
aggregate_type=aggregate_type,
|
|
aggregate_id=aggregate_id,
|
|
actor_type="system",
|
|
actor_id=self.worker_id,
|
|
payload={
|
|
"deal_id": task.deal_id,
|
|
"automation_task_id": task.task_id,
|
|
"task_type": task.task_type,
|
|
**payload,
|
|
},
|
|
)
|
|
|
|
def _find_related_task(
|
|
self,
|
|
session,
|
|
*,
|
|
tenant_id: str,
|
|
deal_id: str,
|
|
task_type: str,
|
|
match_payload: dict[str, Any],
|
|
) -> SalesAutomationTaskRow | None:
|
|
rows = session.execute(
|
|
select(SalesAutomationTaskRow)
|
|
.where(
|
|
SalesAutomationTaskRow.tenant_id == tenant_id,
|
|
SalesAutomationTaskRow.deal_id == deal_id,
|
|
SalesAutomationTaskRow.task_type == task_type,
|
|
SalesAutomationTaskRow.status.in_(["pending", "running", "completed"]),
|
|
)
|
|
.order_by(SalesAutomationTaskRow.id.desc())
|
|
).scalars().all()
|
|
for row in rows:
|
|
payload = _payload(row)
|
|
if all(payload.get(key) == value for key, value in match_payload.items()):
|
|
return row
|
|
return None
|
|
|
|
def _create_task_once(
|
|
self,
|
|
session,
|
|
*,
|
|
tenant_id: str,
|
|
deal_id: str,
|
|
task_type: str,
|
|
run_at: str | None = None,
|
|
payload: dict[str, Any] | None = None,
|
|
match_payload: dict[str, Any] | None = None,
|
|
) -> SalesAutomationTaskRow:
|
|
payload_data = payload or {}
|
|
existing = self._find_related_task(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
deal_id=deal_id,
|
|
task_type=task_type,
|
|
match_payload=match_payload or payload_data,
|
|
)
|
|
if existing is not None:
|
|
return existing
|
|
now = utc_now_iso()
|
|
row = SalesAutomationTaskRow(
|
|
task_id=new_id("tsk"),
|
|
tenant_id=tenant_id,
|
|
deal_id=deal_id,
|
|
task_type=task_type,
|
|
payload_json=json.dumps(payload_data, ensure_ascii=False),
|
|
run_at=run_at or now,
|
|
status="pending",
|
|
retry_count=0,
|
|
max_retries=3,
|
|
locked_at=None,
|
|
locked_by=None,
|
|
completed_at=None,
|
|
failed_at=None,
|
|
last_error=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
return row
|
|
|
|
def _handle_follow_up_customer(self, session, task: SalesAutomationTaskRow) -> None:
|
|
deal = self._get_deal(session, task)
|
|
data = _payload(task)
|
|
if deal.status != "active":
|
|
data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
|
|
_set_payload(task, data)
|
|
return
|
|
|
|
now = utc_now_iso()
|
|
deal.next_action_type = str(data.get("next_action_type") or "follow_up_customer")
|
|
deal.next_action_at = now
|
|
deal.updated_at = now
|
|
current_stage = self._stage_code(session, deal)
|
|
preferred_channel = str(data.get("preferred_channel") or deal.current_channel or deal.preferred_channel or "text")
|
|
if current_stage == "follow_up_scheduled":
|
|
target = "active_voice_communication" if preferred_channel == "voice" else "active_text_communication"
|
|
self._transition_stage(
|
|
session,
|
|
task=task,
|
|
target_stage_code=target,
|
|
reason="automation.follow_up_customer",
|
|
)
|
|
data["automation_result"] = {
|
|
"status": "follow_up_ready",
|
|
"next_action_type": deal.next_action_type,
|
|
"next_action_at": deal.next_action_at,
|
|
}
|
|
_set_payload(task, data)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.DEAL_FOLLOW_UP_READY,
|
|
aggregate_type="deal",
|
|
aggregate_id=deal.deal_id,
|
|
payload={"next_action_type": deal.next_action_type, "next_action_at": deal.next_action_at},
|
|
)
|
|
|
|
def _handle_mark_invoice_overdue(self, session, task: SalesAutomationTaskRow) -> None:
|
|
data = _payload(task)
|
|
invoice_id = str(data.get("invoice_id") or "").strip()
|
|
if not invoice_id:
|
|
raise RuntimeError("invoice_id is required")
|
|
invoice = self._get_invoice(session, task, invoice_id)
|
|
deal = self._get_deal(session, task)
|
|
if invoice.status in {"paid", "canceled"}:
|
|
data["automation_result"] = {"status": "skipped", "invoice_status": invoice.status}
|
|
_set_payload(task, data)
|
|
return
|
|
|
|
previous_status = invoice.status
|
|
now = utc_now_iso()
|
|
invoice.status = "overdue"
|
|
invoice.updated_at = now
|
|
if deal.status == "active":
|
|
self._transition_stage(
|
|
session,
|
|
task=task,
|
|
target_stage_code="payment_overdue",
|
|
reason="automation.mark_invoice_overdue",
|
|
)
|
|
if previous_status != "overdue":
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.INVOICE_OVERDUE,
|
|
aggregate_type="invoice",
|
|
aggregate_id=invoice.invoice_id,
|
|
payload={
|
|
"invoice_id": invoice.invoice_id,
|
|
"invoice_number": invoice.invoice_number,
|
|
"amount": invoice.amount,
|
|
"currency": invoice.currency,
|
|
"due_date": invoice.due_date,
|
|
},
|
|
)
|
|
reminder = self._create_task_once(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
task_type="send_invoice_reminder",
|
|
payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.overdue"},
|
|
match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.overdue"},
|
|
)
|
|
data["automation_result"] = {"status": "invoice_overdue", "reminder_task_id": reminder.task_id}
|
|
_set_payload(task, data)
|
|
|
|
def _handle_send_invoice_reminder(self, session, task: SalesAutomationTaskRow) -> None:
|
|
data = _payload(task)
|
|
invoice_id = str(data.get("invoice_id") or "").strip()
|
|
if not invoice_id:
|
|
raise RuntimeError("invoice_id is required")
|
|
invoice = self._get_invoice(session, task, invoice_id)
|
|
deal = self._get_deal(session, task)
|
|
if invoice.status in {"paid", "canceled"} or deal.status != "active":
|
|
data["automation_result"] = {
|
|
"status": "skipped",
|
|
"invoice_status": invoice.status,
|
|
"deal_status": deal.status,
|
|
}
|
|
_set_payload(task, data)
|
|
return
|
|
|
|
now = utc_now_iso()
|
|
deal.next_action_type = "invoice_reminder"
|
|
deal.next_action_at = now
|
|
deal.updated_at = now
|
|
follow_up = self._create_task_once(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
task_type="follow_up_customer",
|
|
payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.reminder"},
|
|
match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.reminder"},
|
|
)
|
|
data["automation_result"] = {
|
|
"status": "invoice_reminder_ready",
|
|
"follow_up_task_id": follow_up.task_id,
|
|
}
|
|
_set_payload(task, data)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.INVOICE_REMINDER_SENT,
|
|
aggregate_type="invoice",
|
|
aggregate_id=invoice.invoice_id,
|
|
payload={
|
|
"invoice_id": invoice.invoice_id,
|
|
"invoice_number": invoice.invoice_number,
|
|
"invoice_status": invoice.status,
|
|
"follow_up_task_id": follow_up.task_id,
|
|
},
|
|
)
|
|
|
|
def _handle_recommend_channel_switch(self, session, task: SalesAutomationTaskRow) -> None:
|
|
deal = self._get_deal(session, task)
|
|
data = _payload(task)
|
|
recommendation = {
|
|
"recommended_from_channel": data.get("recommended_from_channel") or deal.current_channel,
|
|
"recommended_to_channel": data.get("recommended_to_channel") or data.get("to_channel") or "voice",
|
|
"reason": data.get("reason") or "automation_recommendation",
|
|
}
|
|
deal.next_action_type = "channel_switch_recommended"
|
|
deal.next_action_at = utc_now_iso()
|
|
deal.updated_at = deal.next_action_at
|
|
data["automation_result"] = recommendation
|
|
if data.get("auto_apply"):
|
|
existing_switch = session.execute(
|
|
select(SalesChannelSwitchRow)
|
|
.where(
|
|
SalesChannelSwitchRow.tenant_id == task.tenant_id,
|
|
SalesChannelSwitchRow.deal_id == task.deal_id,
|
|
SalesChannelSwitchRow.recommended_by_task_id == task.task_id,
|
|
)
|
|
.order_by(SalesChannelSwitchRow.id.desc())
|
|
).scalars().first()
|
|
if existing_switch is None:
|
|
now = utc_now_iso()
|
|
to_channel = "voice" if recommendation["recommended_to_channel"] == "voice" else "text"
|
|
from_channel = "voice" if recommendation["recommended_from_channel"] == "voice" else "text"
|
|
existing_switch = SalesChannelSwitchRow(
|
|
switch_id=new_id("swc"),
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
communication_id=None,
|
|
communication_session_id=None,
|
|
previous_communication_session_id=None,
|
|
new_communication_session_id=None,
|
|
from_channel=from_channel,
|
|
to_channel=to_channel,
|
|
reason_code=str(data.get("reason_code") or "system_recommendation"),
|
|
reason_text=recommendation["reason"],
|
|
reason_for_channel_switch=recommendation["reason"],
|
|
initiated_by_type="system",
|
|
initiated_by_id=None,
|
|
source_type="automation_task",
|
|
source_id=task.task_id,
|
|
recommended_by_task_id=task.task_id,
|
|
switched_at=now,
|
|
created_at=now,
|
|
)
|
|
session.add(existing_switch)
|
|
deal.current_channel = to_channel
|
|
deal.updated_at = now
|
|
current_stage = self._stage_code(session, deal)
|
|
target_stage = "active_voice_communication" if to_channel == "voice" else "active_text_communication"
|
|
if current_stage in {"active_text_communication", "active_voice_communication"} and current_stage != target_stage:
|
|
transition_deal_stage(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
target_stage_code=target_stage,
|
|
actor_type="system",
|
|
reason="automation.recommend_channel_switch",
|
|
metadata={"automation_task_id": task.task_id, "switch_id": existing_switch.switch_id},
|
|
)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.COMMUNICATION_CHANNEL_SWITCHED,
|
|
aggregate_type="deal",
|
|
aggregate_id=deal.deal_id,
|
|
payload={
|
|
"switch_id": existing_switch.switch_id,
|
|
"from_channel": existing_switch.from_channel,
|
|
"to_channel": existing_switch.to_channel,
|
|
"reason_code": existing_switch.reason_code,
|
|
"recommended_by_task_id": task.task_id,
|
|
},
|
|
)
|
|
data["actual_switch_id"] = existing_switch.switch_id
|
|
_set_payload(task, data)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.DEAL_CHANNEL_SWITCH_RECOMMENDED,
|
|
aggregate_type="deal",
|
|
aggregate_id=deal.deal_id,
|
|
payload=recommendation,
|
|
)
|
|
|
|
def _handle_close_stale_communication(self, session, task: SalesAutomationTaskRow) -> None:
|
|
data = _payload(task)
|
|
communication_id = str(data.get("communication_id") or "").strip()
|
|
if not communication_id:
|
|
raise RuntimeError("communication_id is required")
|
|
communication = session.execute(
|
|
select(SalesCommunicationSessionRow).where(
|
|
SalesCommunicationSessionRow.tenant_id == task.tenant_id,
|
|
SalesCommunicationSessionRow.deal_id == task.deal_id,
|
|
SalesCommunicationSessionRow.communication_id == communication_id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if communication is None:
|
|
raise RuntimeError(f"Communication not found: {communication_id}")
|
|
if communication.status != "completed":
|
|
now = utc_now_iso()
|
|
communication.status = "completed"
|
|
communication.ended_at = communication.ended_at or now
|
|
communication.result_code = communication.result_code or str(data.get("result_code") or "stale_closed")
|
|
communication.updated_at = now
|
|
follow_up = self._create_task_once(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
task_type="follow_up_customer",
|
|
payload={"communication_id": communication.communication_id, "source_task_id": task.task_id, "reason": "stale_communication"},
|
|
match_payload={"communication_id": communication.communication_id, "reason": "stale_communication"},
|
|
)
|
|
data["automation_result"] = {"status": "communication_closed", "follow_up_task_id": follow_up.task_id}
|
|
_set_payload(task, data)
|
|
|
|
def _handle_escalate_to_human(self, session, task: SalesAutomationTaskRow) -> None:
|
|
deal = self._get_deal(session, task)
|
|
data = _payload(task)
|
|
escalation_type = str(data.get("escalation_type") or "automation_escalation").strip()
|
|
reason = str(data.get("reason") or "automation.escalate_to_human").strip()
|
|
assigned_to = str(data.get("assigned_to_user_id") or deal.assigned_human_user_id or "").strip() or None
|
|
existing = session.execute(
|
|
select(SalesEscalationRow)
|
|
.where(
|
|
SalesEscalationRow.tenant_id == task.tenant_id,
|
|
SalesEscalationRow.deal_id == task.deal_id,
|
|
SalesEscalationRow.escalation_type == escalation_type,
|
|
SalesEscalationRow.status.in_(["open", "assigned", "in_progress"]),
|
|
)
|
|
.order_by(SalesEscalationRow.id.desc())
|
|
).scalars().first()
|
|
now = utc_now_iso()
|
|
if existing is None:
|
|
existing = SalesEscalationRow(
|
|
escalation_id=new_id("esc"),
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
escalation_type=escalation_type,
|
|
reason=reason,
|
|
severity=str(data.get("severity") or "medium"),
|
|
status="assigned" if assigned_to else "open",
|
|
assigned_to_user_id=assigned_to,
|
|
assigned_at=now if assigned_to else None,
|
|
started_at=None,
|
|
created_at=now,
|
|
resolved_at=None,
|
|
canceled_at=None,
|
|
resolution_code=None,
|
|
resolution_summary=None,
|
|
sla_due_at=data.get("sla_due_at"),
|
|
source_channel=deal.current_channel,
|
|
source_communication_session_id=data.get("communication_id"),
|
|
source_automation_task_id=task.task_id,
|
|
updated_at=now,
|
|
)
|
|
session.add(existing)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.ESCALATION_CREATED,
|
|
aggregate_type="escalation",
|
|
aggregate_id=existing.escalation_id,
|
|
payload={
|
|
"escalation_id": existing.escalation_id,
|
|
"escalation_type": existing.escalation_type,
|
|
"status": existing.status,
|
|
"severity": existing.severity,
|
|
},
|
|
)
|
|
else:
|
|
existing.updated_at = now
|
|
deal.assigned_human_user_id = assigned_to
|
|
deal.scenario_type = "custom_human_escalation"
|
|
deal.updated_at = now
|
|
if self._stage_code(session, deal) != "transferred_to_support":
|
|
transition_deal_stage(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
target_stage_code="transferred_to_support",
|
|
actor_type="system",
|
|
reason="automation.escalate_to_human",
|
|
metadata={"automation_task_id": task.task_id, "escalation_id": existing.escalation_id},
|
|
force=True,
|
|
)
|
|
data["automation_result"] = {"status": "escalated", "escalation_id": existing.escalation_id}
|
|
_set_payload(task, data)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.DEAL_ESCALATION_REQUESTED,
|
|
aggregate_type="deal",
|
|
aggregate_id=deal.deal_id,
|
|
payload={
|
|
"escalation_id": existing.escalation_id,
|
|
"escalation_type": existing.escalation_type,
|
|
"severity": existing.severity,
|
|
"assigned_to_user_id": existing.assigned_to_user_id,
|
|
},
|
|
)
|
|
|
|
def _handle_post_sale_transfer(self, session, task: SalesAutomationTaskRow) -> None:
|
|
deal = self._get_deal(session, task)
|
|
current_stage = self._stage_code(session, deal)
|
|
data = _payload(task)
|
|
if current_stage == "transferred_to_execution":
|
|
data["automation_result"] = {"status": "skipped", "reason": "already_transferred"}
|
|
_set_payload(task, data)
|
|
return
|
|
if deal.status != "won":
|
|
data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
|
|
_set_payload(task, data)
|
|
return
|
|
transition_deal_stage(
|
|
session,
|
|
tenant_id=task.tenant_id,
|
|
deal_id=task.deal_id,
|
|
target_stage_code="transferred_to_execution",
|
|
actor_type="system",
|
|
reason="automation.post_sale_transfer",
|
|
metadata={"automation_task_id": task.task_id},
|
|
force=True,
|
|
)
|
|
data["automation_result"] = {"status": "transferred_post_sale"}
|
|
_set_payload(task, data)
|
|
self._publish(
|
|
session,
|
|
task=task,
|
|
event_type=sales_event_types.DEAL_TRANSFERRED_POST_SALE,
|
|
aggregate_type="deal",
|
|
aggregate_id=deal.deal_id,
|
|
payload={"transfer_status": "stub_created"},
|
|
)
|