408 lines
18 KiB
Python
408 lines
18 KiB
Python
"""Materialize central opportunity next actions into operator tasks.
|
|
|
|
The workflow engine can decide that the next step is a human action even when no
|
|
message-triggered task exists. Operator pages should not leave those actions as
|
|
abstract labels only: a pending task must exist so the workbench, related-task
|
|
list and task completion flow remain consistent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from typing import Any, Mapping
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.task_service import get_action_config
|
|
from app.work_center_action_policy import (
|
|
RECONSTRUCTED_SENSITIVE_ACTIONS,
|
|
canonical_action_code,
|
|
reconstructed_review_required,
|
|
)
|
|
|
|
|
|
# v4928.1.5.96 marker: MATERIALIZED_ACTIONS = {"SEND_INVOICE"}
|
|
# v4928.1.5.105 marker: MATERIALIZED_ACTIONS = {"SEND_INVOICE", "FOLLOW_UP_PAYMENT"}
|
|
MATERIALIZED_ACTIONS = {"SEND_INVOICE", "FOLLOW_UP_PAYMENT", "PREPARE_ORDER"}
|
|
# v4928.1.5.129: central workflow emits SHIP_ORDER; operator tasks persist CREATE_SHIPMENT.
|
|
MATERIALIZED_ACTIONS.add("CREATE_SHIPMENT")
|
|
MATERIALIZED_ACTIONS.add("VALIDATE_PHYSICAL_ORDER")
|
|
MATERIALIZED_ACTIONS.add("REVIEW_RECONSTRUCTED_PROCESS")
|
|
ACTION_ALIASES = {"SHIP_ORDER": "CREATE_SHIPMENT"}
|
|
|
|
|
|
def _s(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value or {}, ensure_ascii=False, default=str)
|
|
|
|
|
|
def ensure_pending_task_for_next_action(
|
|
opportunity_id: str,
|
|
next_action: Mapping[str, Any] | None,
|
|
*,
|
|
source: str = "opportunity_next_action",
|
|
actor: str = "system",
|
|
reactivate_skipped: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Ensure a pending task exists for a materialized next action.
|
|
|
|
Returns a small status dictionary. The function is idempotent by both a
|
|
pending-task lookup and a stable idempotency key.
|
|
"""
|
|
|
|
if not opportunity_id or not next_action:
|
|
return {"created": False, "reason": "missing_input"}
|
|
|
|
raw_action_code = canonical_action_code(next_action.get("action_code") or next_action.get("code"))
|
|
document_id = _s(next_action.get("document_id"))
|
|
document_number = _s(next_action.get("document_number"))
|
|
document_key = document_id or document_number or "no_document"
|
|
|
|
with engine.begin() as conn:
|
|
opp = conn.execute(text("""
|
|
SELECT
|
|
id::text,
|
|
conversation_id,
|
|
contact_id,
|
|
customer_id,
|
|
local_customer_id::text AS local_customer_id,
|
|
customer_email,
|
|
title,
|
|
COALESCE(metadata, '{}'::jsonb) AS metadata
|
|
FROM opportunities
|
|
WHERE id = CAST(:opportunity_id AS UUID)
|
|
"""), {"opportunity_id": opportunity_id}).mappings().first()
|
|
if not opp:
|
|
return {"created": False, "reason": "opportunity_not_found", "action_code": raw_action_code}
|
|
|
|
blocked_action_code = ""
|
|
action_code = raw_action_code
|
|
sensitive_codes = {canonical_action_code(code) for code in RECONSTRUCTED_SENSITIVE_ACTIONS}
|
|
if reconstructed_review_required(opp.get("metadata")) and action_code in sensitive_codes:
|
|
blocked_action_code = action_code
|
|
action_code = "REVIEW_RECONSTRUCTED_PROCESS"
|
|
|
|
if action_code not in MATERIALIZED_ACTIONS:
|
|
return {"created": False, "reason": "action_not_materialized", "action_code": action_code}
|
|
|
|
existing = conn.execute(text("""
|
|
SELECT id::text
|
|
FROM tasks
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND status = 'pending'
|
|
AND action_code = :action_code
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"action_code": action_code,
|
|
}).scalar()
|
|
if existing:
|
|
return {"created": False, "reason": "pending_exists", "task_id": existing, "action_code": action_code}
|
|
|
|
# v4928.1.5.132.2.1: build the canonical task payload before either
|
|
# reactivating a terminal task or inserting a new one. The v132.2
|
|
# pre-insert branch referenced these values before assignment.
|
|
config = get_action_config(action_code)
|
|
default_routes = {
|
|
"SEND_INVOICE": "financeiro",
|
|
"FOLLOW_UP_PAYMENT": "financeiro",
|
|
"PREPARE_ORDER": "operacoes",
|
|
"VALIDATE_PHYSICAL_ORDER": "logistica",
|
|
"CREATE_SHIPMENT": "logistica",
|
|
"REVIEW_RECONSTRUCTED_PROCESS": "rever",
|
|
}
|
|
route = _s(config.get("route")) or default_routes.get(action_code, "financeiro")
|
|
if action_code == "PREPARE_ORDER" and route in {"", "rever", "vendas"}:
|
|
route = "operacoes"
|
|
if action_code in {"VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT"} and route in {"", "rever", "vendas", "financeiro"}:
|
|
route = "logistica"
|
|
if action_code == "REVIEW_RECONSTRUCTED_PROCESS":
|
|
route = "rever"
|
|
|
|
default_labels = {
|
|
"SEND_INVOICE": "Enviar fatura ao cliente",
|
|
"FOLLOW_UP_PAYMENT": "Follow-up pagamento",
|
|
"PREPARE_ORDER": "Preparar encomenda / Odoo",
|
|
"VALIDATE_PHYSICAL_ORDER": "Validar encomenda física",
|
|
"CREATE_SHIPMENT": "Enviar encomenda",
|
|
"REVIEW_RECONSTRUCTED_PROCESS": "Validar processo reconstruído",
|
|
}
|
|
default_descriptions = {
|
|
"SEND_INVOICE": "Fatura criada/associada. Enviar PDF ao cliente e registar evidência.",
|
|
"FOLLOW_UP_PAYMENT": "Encomenda concluída no Odoo/WH-OUT e fatura enviada. Acompanhar pagamento pós-entrega.",
|
|
"PREPARE_ORDER": "Fatura e pagamento confirmados. Criar/validar venda Odoo e preparação da encomenda.",
|
|
"VALIDATE_PHYSICAL_ORDER": "O picking está assigned/reservado no Odoo. Confirmar fisicamente a preparação antes de criar o envio.",
|
|
"CREATE_SHIPMENT": "Encomenda fisicamente validada. Criar envio/tracking ou registar a expedição pelo canal disponível.",
|
|
"REVIEW_RECONSTRUCTED_PROCESS": "Confirmar cliente, documento principal, valor e evidências antes de executar a ação sensível bloqueada.",
|
|
}
|
|
default_label = default_labels.get(action_code, action_code)
|
|
default_description = default_descriptions.get(action_code, "Executar próxima ação operacional.")
|
|
if action_code == "REVIEW_RECONSTRUCTED_PROCESS":
|
|
action_label = _s(config.get("action") or default_label)
|
|
description = default_description
|
|
else:
|
|
action_label = _s(next_action.get("label") or config.get("action") or default_label)
|
|
description = _s(next_action.get("description")) or default_description
|
|
priority = _s(next_action.get("priority")) or "alta"
|
|
if priority not in {"alta", "normal", "baixa"}:
|
|
priority = "alta"
|
|
|
|
idempotency_key = f"task:next_action:{opportunity_id}:{action_code}:{document_key}"
|
|
task_id = str(uuid.uuid4())
|
|
metadata = {
|
|
"source": source,
|
|
"actor": actor,
|
|
"materialized_from_next_action": True,
|
|
"document_id": document_id or None,
|
|
"document_number": document_number or None,
|
|
"next_action": dict(next_action),
|
|
"raw_action_code": raw_action_code,
|
|
"normalized_action_code": action_code,
|
|
"blocked_action_code": blocked_action_code or None,
|
|
"review_type": "reconstructed_process" if action_code == "REVIEW_RECONSTRUCTED_PROCESS" else None,
|
|
}
|
|
|
|
# v4928.1.5.132.2: when an explicit blocker has just been cleared,
|
|
# reactivate the stable materialized task before attempting an INSERT.
|
|
# This avoids depending on the ON CONFLICT branch and makes the transition
|
|
# observable even when an operator previously skipped the task while it
|
|
# was superseded by a review task.
|
|
allowed_reactivation_sources = {
|
|
"reconstructed_review_completion",
|
|
"physical_validation_completion",
|
|
"manual_v132_2_reactivation_repair",
|
|
}
|
|
if reactivate_skipped and source in allowed_reactivation_sources:
|
|
terminal = conn.execute(text("""
|
|
SELECT id::text, status, action_code, COALESCE(metadata, '{}'::jsonb) AS metadata
|
|
FROM tasks
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND action_code = :action_code
|
|
AND status IN ('skipped', 'ignored')
|
|
AND COALESCE((metadata->>'materialized_from_next_action')::boolean, FALSE) = TRUE
|
|
ORDER BY created_at DESC
|
|
LIMIT 1
|
|
FOR UPDATE
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"action_code": action_code,
|
|
}).mappings().first()
|
|
if terminal:
|
|
reactivation_metadata = {
|
|
**metadata,
|
|
"reactivated_after_blocker": True,
|
|
"reactivated_from_status": terminal.get("status"),
|
|
"reactivated_by": actor,
|
|
"reactivated_source": source,
|
|
"reactivated_version": "v4928.1.5.132.2.1",
|
|
}
|
|
reactivated = conn.execute(text("""
|
|
UPDATE tasks
|
|
SET status = 'pending',
|
|
route = :route,
|
|
action = :action,
|
|
note = :note,
|
|
action_required = TRUE,
|
|
safe_to_post = FALSE,
|
|
source_system = 'clientflow_next_action',
|
|
source_event_id = :source_event_id,
|
|
due_at = now(),
|
|
done_at = NULL,
|
|
done_by = NULL,
|
|
priority = :priority,
|
|
metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now()
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
AND status IN ('skipped', 'ignored')
|
|
RETURNING id::text
|
|
"""), {
|
|
"task_id": terminal.get("id"),
|
|
"route": route,
|
|
"action": action_label,
|
|
"note": description,
|
|
"source_event_id": f"{source}:{opportunity_id}:{action_code}:{document_key}:reactivated",
|
|
"priority": priority,
|
|
"metadata": _json(reactivation_metadata),
|
|
}).scalar()
|
|
if reactivated:
|
|
conn.execute(text("""
|
|
INSERT INTO task_events (task_id, event_type, payload, created_by)
|
|
VALUES (CAST(:task_id AS UUID), 'task_reactivated', CAST(:payload AS JSONB), :created_by)
|
|
"""), {
|
|
"task_id": reactivated,
|
|
"payload": _json({
|
|
"action_code": action_code,
|
|
"reason": "explicit_blocker_cleared_preinsert",
|
|
"source": source,
|
|
"previous_status": terminal.get("status"),
|
|
"version": "v4928.1.5.132.2.1",
|
|
}),
|
|
"created_by": actor,
|
|
})
|
|
return {
|
|
"created": False,
|
|
"reactivated": True,
|
|
"reason": "reactivated_after_blocker_preinsert",
|
|
"task_id": reactivated,
|
|
"action_code": action_code,
|
|
}
|
|
|
|
|
|
|
|
row = conn.execute(text("""
|
|
INSERT INTO tasks (
|
|
id,
|
|
opportunity_id,
|
|
conversation_id,
|
|
contact_id,
|
|
customer_id,
|
|
action_code,
|
|
route,
|
|
action,
|
|
note,
|
|
action_required,
|
|
safe_to_post,
|
|
status,
|
|
source_system,
|
|
source_event_id,
|
|
idempotency_key,
|
|
due_at,
|
|
metadata,
|
|
priority,
|
|
created_at,
|
|
updated_at
|
|
) VALUES (
|
|
CAST(:id AS UUID),
|
|
CAST(:opportunity_id AS UUID),
|
|
:conversation_id,
|
|
:contact_id,
|
|
:customer_id,
|
|
:action_code,
|
|
:route,
|
|
:action,
|
|
:note,
|
|
TRUE,
|
|
FALSE,
|
|
'pending',
|
|
'clientflow_next_action',
|
|
:source_event_id,
|
|
:idempotency_key,
|
|
now(),
|
|
CAST(:metadata AS JSONB),
|
|
:priority,
|
|
now(),
|
|
now()
|
|
)
|
|
ON CONFLICT (idempotency_key) DO NOTHING
|
|
RETURNING id::text
|
|
"""), {
|
|
"id": task_id,
|
|
"opportunity_id": opportunity_id,
|
|
"conversation_id": opp.get("conversation_id"),
|
|
"contact_id": opp.get("contact_id"),
|
|
"customer_id": opp.get("local_customer_id") or opp.get("customer_id") or opp.get("customer_email"),
|
|
"action_code": action_code,
|
|
"route": route,
|
|
"action": action_label,
|
|
"note": description,
|
|
"source_event_id": f"{source}:{opportunity_id}:{action_code}:{document_key}",
|
|
"idempotency_key": idempotency_key,
|
|
"metadata": _json(metadata),
|
|
"priority": priority,
|
|
}).fetchone()
|
|
|
|
created_id = row[0] if row else None
|
|
if not created_id:
|
|
existing_after_conflict = conn.execute(text("""
|
|
SELECT id::text, status, action_code, COALESCE(metadata, '{}'::jsonb) AS metadata
|
|
FROM tasks
|
|
WHERE idempotency_key = :idempotency_key
|
|
LIMIT 1
|
|
FOR UPDATE
|
|
"""), {"idempotency_key": idempotency_key}).mappings().first()
|
|
if (
|
|
reactivate_skipped
|
|
and existing_after_conflict
|
|
and _s(existing_after_conflict.get("status")).lower() in {"skipped", "ignored"}
|
|
and canonical_action_code(existing_after_conflict.get("action_code")) == action_code
|
|
):
|
|
reactivated = conn.execute(text("""
|
|
UPDATE tasks
|
|
SET status = 'pending',
|
|
route = :route,
|
|
action = :action,
|
|
note = :note,
|
|
action_required = TRUE,
|
|
safe_to_post = FALSE,
|
|
source_system = 'clientflow_next_action',
|
|
source_event_id = :source_event_id,
|
|
due_at = now(),
|
|
done_at = NULL,
|
|
done_by = NULL,
|
|
priority = :priority,
|
|
metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
|
|
updated_at = now()
|
|
WHERE id = CAST(:task_id AS UUID)
|
|
AND status IN ('skipped', 'ignored')
|
|
RETURNING id::text
|
|
"""), {
|
|
"task_id": existing_after_conflict.get("id"),
|
|
"route": route,
|
|
"action": action_label,
|
|
"note": description,
|
|
"source_event_id": f"{source}:{opportunity_id}:{action_code}:{document_key}:reactivated",
|
|
"priority": priority,
|
|
"metadata": _json({
|
|
**metadata,
|
|
"reactivated_after_blocker": True,
|
|
"reactivated_from_status": existing_after_conflict.get("status"),
|
|
"reactivated_by": actor,
|
|
"reactivated_source": source,
|
|
"reactivated_version": "v4928.1.5.132.2.1",
|
|
}),
|
|
}).scalar()
|
|
if reactivated:
|
|
conn.execute(text("""
|
|
INSERT INTO task_events (task_id, event_type, payload, created_by)
|
|
VALUES (CAST(:task_id AS UUID), 'task_reactivated', CAST(:payload AS JSONB), :created_by)
|
|
"""), {
|
|
"task_id": reactivated,
|
|
"payload": _json({
|
|
"action_code": action_code,
|
|
"reason": "explicit_blocker_cleared",
|
|
"source": source,
|
|
"previous_status": existing_after_conflict.get("status"),
|
|
}),
|
|
"created_by": actor,
|
|
})
|
|
return {
|
|
"created": False,
|
|
"reactivated": True,
|
|
"reason": "reactivated_after_blocker",
|
|
"task_id": reactivated,
|
|
"action_code": action_code,
|
|
}
|
|
return {
|
|
"created": False,
|
|
"reason": "idempotency_conflict",
|
|
"task_id": existing_after_conflict.get("id") if existing_after_conflict else None,
|
|
"existing_status": existing_after_conflict.get("status") if existing_after_conflict else None,
|
|
"action_code": action_code,
|
|
}
|
|
|
|
conn.execute(text("""
|
|
INSERT INTO task_events (task_id, event_type, payload, created_by)
|
|
VALUES (CAST(:task_id AS UUID), 'task_created', CAST(:payload AS JSONB), :created_by)
|
|
"""), {
|
|
"task_id": created_id,
|
|
"payload": _json(metadata),
|
|
"created_by": actor,
|
|
})
|
|
|
|
return {"created": True, "task_id": created_id, "action_code": action_code, "document_number": document_number or None}
|