import re from datetime import datetime, timedelta from fastapi import HTTPException from app.services.orchestration.orchestrator_config import ( LAST_REVIEW_PACKAGE_TTL_MINUTES, PENDING_REVIEW_DRAFT_TTL_MINUTES, REVIEW_REQUIRED_FIELDS, ) # Esse mixin concentra os fluxos incrementais de revisao e pos-venda. class ReviewFlowMixin: def _decision_intent(self, turn_decision: dict | None) -> str: return str((turn_decision or {}).get("intent") or "").strip().lower() def _log_review_flow_source( self, source: str, payload: dict | None = None, missing_fields: list[str] | None = None, ) -> None: if not hasattr(self, "_log_turn_event"): return self._log_turn_event( "review_flow_progress", review_flow_source=source, payload_keys=sorted((payload or {}).keys()), missing_fields=list(missing_fields or []), ) def _active_domain(self, user_id: int | None) -> str: if user_id is None or not hasattr(self, "_get_user_context"): return "general" context = self._get_user_context(user_id) if not isinstance(context, dict): return "general" return str(context.get("active_domain") or "general").strip().lower() def _supplement_review_fields_from_message(self, message: str, payload: dict) -> None: if not isinstance(payload, dict): return normalized_message = self._normalize_text(message).strip() if "placa" not in payload: for token in str(message or "").split(): normalized_plate = self.normalizer.normalize_plate(token) if normalized_plate: payload["placa"] = normalized_plate break if "data_hora" not in payload: normalized_datetime = self._normalize_review_datetime_text(message) if normalized_datetime and normalized_datetime != str(message or "").strip(): payload["data_hora"] = normalized_datetime if "km" not in payload: km_match = re.search(r"(? str | None: normalized_message = self._normalize_text(message).strip() management_fields = self._normalize_review_management_fields(extracted_fields) has_protocol = bool(management_fields.get("protocolo") or self._extract_review_protocol_from_text(message)) if any(term in normalized_message for term in {"agendamento", "agendamentos"}) and any( term in normalized_message for term in {"listar", "liste", "mostrar", "mostre", "ver", "consultar"} ): return "list" if not has_protocol: return None if any(term in normalized_message for term in {"remarcar", "reagendar", "alterar data", "mudar data", "trocar data"}): return "reschedule" if any(term in normalized_message for term in {"cancelar", "cancelamento", "desmarcar"}): return "cancel" return None def _should_bootstrap_review_from_active_context(self, message: str, payload: dict | None = None) -> bool: normalized_message = self._normalize_text(message).strip() normalized_payload = payload if isinstance(payload, dict) else {} if normalized_payload: return True explicit_review_terms = { "agendar revisao", "marcar revisao", "nova revisao", "revisao agora", "revisao", } return any(term in normalized_message for term in explicit_review_terms) async def _try_handle_review_management( self, message: str, user_id: int | None, extracted_fields: dict | None = None, intents: dict | None = None, turn_decision: dict | None = None, ) -> str | None: if user_id is None: return None normalized_intents = self._normalize_intents(intents) draft = self.state.get_entry("pending_review_management_drafts", user_id, expire=True) schedule_draft = self.state.get_entry("pending_review_drafts", user_id, expire=True) pending_reuse = self.state.get_entry("pending_review_reuse_confirmations", user_id, expire=True) decision_intent = self._decision_intent(turn_decision) inferred_action = self._infer_review_management_action(message=message, extracted_fields=extracted_fields) normalized_fields = self._normalize_review_management_fields(extracted_fields) protocol_in_message = normalized_fields.get("protocolo") or self._extract_review_protocol_from_text(message) open_schedule_context = bool(schedule_draft or pending_reuse) has_list_intent = ( decision_intent == "review_list" or normalized_intents.get("review_list", False) or inferred_action == "list" ) has_cancel_intent = ( decision_intent == "review_cancel" or normalized_intents.get("review_cancel", False) or inferred_action == "cancel" ) has_reschedule_intent = ( decision_intent == "review_reschedule" or normalized_intents.get("review_reschedule", False) or inferred_action == "reschedule" ) if open_schedule_context and not protocol_in_message and inferred_action is None: return None if (decision_intent == "review_schedule" or normalized_intents.get("review_schedule", False)) and inferred_action is None: if draft is not None: self.state.pop_entry("pending_review_management_drafts", user_id) draft = None return None if has_list_intent: self._reset_pending_review_states(user_id=user_id) try: tool_result = await self.tool_executor.execute( "listar_agendamentos_revisao", {"limite": 20}, user_id=user_id, ) except HTTPException as exc: return self._http_exception_detail(exc) return self._fallback_format_tool_result("listar_agendamentos_revisao", tool_result) if not has_cancel_intent and not has_reschedule_intent and draft is None: return None if draft is None: action = "reschedule" if has_reschedule_intent else "cancel" draft = { "action": action, "payload": {}, "expires_at": datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES), } else: if has_reschedule_intent: draft["action"] = "reschedule" elif has_cancel_intent: draft["action"] = "cancel" extracted = self._normalize_review_management_fields(extracted_fields) if "protocolo" not in extracted: inferred_protocol = self._extract_review_protocol_from_text(message) if inferred_protocol: extracted["protocolo"] = inferred_protocol action = draft.get("action", "cancel") if ( action == "cancel" and "motivo" not in extracted and draft["payload"].get("protocolo") and not has_cancel_intent ): free_text = str(message or "").strip() if free_text and len(free_text) >= 4 and not self._is_affirmative_message(free_text): extracted["motivo"] = free_text draft["payload"].update(extracted) draft["expires_at"] = datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES) self.state.set_entry("pending_review_management_drafts", user_id, draft) if action == "reschedule": missing = [field for field in ("protocolo", "nova_data_hora") if field not in draft["payload"]] if missing: return self._render_missing_review_reschedule_fields_prompt(missing) try: tool_result = await self.tool_executor.execute( "editar_data_revisao", { "protocolo": draft["payload"]["protocolo"], "nova_data_hora": draft["payload"]["nova_data_hora"], }, user_id=user_id, ) except HTTPException as exc: error = self.tool_executor.coerce_http_error(exc) if error.get("retryable") and error.get("field"): draft["payload"].pop(str(error["field"]), None) draft["expires_at"] = datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES) self.state.set_entry("pending_review_management_drafts", user_id, draft) return self._http_exception_detail(exc) self.state.pop_entry("pending_review_management_drafts", user_id) return self._fallback_format_tool_result("editar_data_revisao", tool_result) missing = [field for field in ("protocolo",) if field not in draft["payload"]] if missing: return self._render_missing_review_cancel_fields_prompt(missing) try: tool_result = await self.tool_executor.execute( "cancelar_agendamento_revisao", { "protocolo": draft["payload"]["protocolo"], "motivo": draft["payload"].get("motivo"), }, user_id=user_id, ) except HTTPException as exc: error = self.tool_executor.coerce_http_error(exc) if error.get("retryable") and error.get("field"): draft["payload"].pop(str(error["field"]), None) draft["expires_at"] = datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES) self.state.set_entry("pending_review_management_drafts", user_id, draft) return self._http_exception_detail(exc) self.state.pop_entry("pending_review_management_drafts", user_id) return self._fallback_format_tool_result("cancelar_agendamento_revisao", tool_result) def _render_missing_review_fields_prompt(self, missing_fields: list[str]) -> str: labels = { "placa": "a placa do veiculo", "data_hora": "a data e hora desejada para a revisao", "modelo": "o modelo do veiculo", "ano": "o ano do veiculo", "km": "a quilometragem atual (km)", "revisao_previa_concessionaria": "se ja fez revisao na concessionaria (sim/nao)", } itens = [f"- {labels[field]}" for field in missing_fields] return "Para agendar sua revisao, preciso dos dados abaixo:\n" + "\n".join(itens) def _render_missing_review_cancel_fields_prompt(self, missing_fields: list[str]) -> str: labels = { "protocolo": "o protocolo da revisao (ex.: REV-20260310-ABC12345)", } itens = [f"- {labels[field]}" for field in missing_fields] return "Para cancelar o agendamento de revisao, preciso dos dados abaixo:\n" + "\n".join(itens) def _render_missing_review_reschedule_fields_prompt(self, missing_fields: list[str]) -> str: labels = { "protocolo": "o protocolo da revisao (ex.: REV-20260310-ABC12345)", "nova_data_hora": "a nova data e hora desejada para a revisao", } itens = [f"- {labels[field]}" for field in missing_fields] return "Para remarcar sua revisao, preciso dos dados abaixo:\n" + "\n".join(itens) def _render_review_reuse_question(self, payload: dict | None = None) -> str: package = payload if isinstance(payload, dict) else {} plate = str(package.get("placa") or "").strip() model = str(package.get("modelo") or "").strip() vehicle_label = "" if plate and model: vehicle_label = f" do ultimo veiculo ({model}, placa {plate})" elif plate: vehicle_label = f" do ultimo veiculo (placa {plate})" elif model: vehicle_label = f" do ultimo veiculo ({model})" return ( f"Posso reutilizar os dados{vehicle_label} e voce me passa so a nova data/hora da revisao? " "(sim/nao)" ) def _store_last_review_package(self, user_id: int | None, payload: dict | None) -> None: if user_id is None or not isinstance(payload, dict): return # Guarda um pacote reutilizavel do ultimo veiculo informado # para reduzir repeticao em novos agendamentos. package = { "placa": payload.get("placa"), "modelo": payload.get("modelo"), "ano": payload.get("ano"), "km": payload.get("km"), "revisao_previa_concessionaria": payload.get("revisao_previa_concessionaria"), } sanitized = {k: v for k, v in package.items() if v is not None} required = {"placa", "modelo", "ano", "km", "revisao_previa_concessionaria"} if not required.issubset(sanitized.keys()): return self.state.set_entry( "last_review_packages", user_id, { "payload": sanitized, "expires_at": datetime.utcnow() + timedelta(minutes=LAST_REVIEW_PACKAGE_TTL_MINUTES), }, ) def _get_last_review_package(self, user_id: int | None) -> dict | None: if user_id is None: return None cached = self.state.get_entry("last_review_packages", user_id, expire=True) if not cached: return None payload = cached.get("payload") return dict(payload) if isinstance(payload, dict) else None async def _try_collect_and_schedule_review( self, message: str, user_id: int | None, extracted_fields: dict | None = None, intents: dict | None = None, turn_decision: dict | None = None, ) -> str | None: if user_id is None: return None normalized_intents = self._normalize_intents(intents) decision_intent = self._decision_intent(turn_decision) has_intent = decision_intent == "review_schedule" or normalized_intents.get("review_schedule", False) has_management_intent = ( decision_intent in {"review_list", "review_cancel", "review_reschedule"} or normalized_intents.get("review_list", False) or normalized_intents.get("review_cancel", False) or normalized_intents.get("review_reschedule", False) ) if self._infer_review_management_action(message=message, extracted_fields=extracted_fields): return None if has_management_intent: self.state.pop_entry("pending_review_drafts", user_id) self.state.pop_entry("pending_review_reuse_confirmations", user_id) return None draft = self.state.get_entry("pending_review_drafts", user_id, expire=True) extracted = self._normalize_review_fields(extracted_fields) pending_reuse = self.state.get_entry("pending_review_reuse_confirmations", user_id, expire=True) pending_confirmation = self.state.get_entry("pending_review_confirmations", user_id, expire=True) active_review_context = self._active_domain(user_id) == "review" review_flow_source = "draft" if draft else None if has_intent and draft is None and pending_confirmation and not self._is_affirmative_message(message): self.state.pop_entry("pending_review_confirmations", user_id) pending_confirmation = None if pending_reuse: should_reuse = False if self._is_negative_message(message): self.state.pop_entry("pending_review_reuse_confirmations", user_id) pending_reuse = None elif self._is_affirmative_message(message) or "data_hora" in extracted: should_reuse = True else: self._log_review_flow_source(source="last_review_package", payload=pending_reuse.get("payload")) return self._render_review_reuse_question(pending_reuse.get("payload")) if should_reuse: seed_payload = dict(pending_reuse.get("payload") or {}) if draft is None: draft = { "payload": seed_payload, "expires_at": datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES), } else: for key, value in seed_payload.items(): draft["payload"].setdefault(key, value) self.state.pop_entry("pending_review_reuse_confirmations", user_id) review_flow_source = "last_review_package" if "data_hora" not in extracted: self.state.set_entry("pending_review_drafts", user_id, draft) self._log_review_flow_source(source=review_flow_source, payload=draft["payload"], missing_fields=["data_hora"]) return "Perfeito. Me informe apenas a data e hora desejada para a revisao." if has_intent and draft is None and not extracted: last_package = self._get_last_review_package(user_id=user_id) if last_package: self.state.set_entry( "pending_review_reuse_confirmations", user_id, { "payload": last_package, "expires_at": datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES), }, ) self._log_review_flow_source(source="last_review_package", payload=last_package) return self._render_review_reuse_question(last_package) if ( draft and not has_intent and ( decision_intent in {"order_create", "order_cancel"} or normalized_intents.get("order_create", False) or normalized_intents.get("order_cancel", False) ) and not extracted ): self.state.pop_entry("pending_review_drafts", user_id) return None bootstrap_payload = dict(extracted) self._supplement_review_fields_from_message(message=message, payload=bootstrap_payload) self._try_prefill_review_fields_from_memory(user_id=user_id, payload=bootstrap_payload) should_bootstrap_from_context = ( active_review_context and self._should_bootstrap_review_from_active_context(message=message, payload=bootstrap_payload) ) if not has_intent and draft is None and not should_bootstrap_from_context: return None if draft is None: # Cria um draft com TTL para permitir coleta do agendamento # em varias mensagens sem perder o progresso. review_flow_source = "active_domain_fallback" if should_bootstrap_from_context and not has_intent else "intent_bootstrap" draft = { "payload": dict(bootstrap_payload), "expires_at": datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES), } draft["payload"].update(extracted) self._supplement_review_fields_from_message(message=message, payload=draft["payload"]) self._try_prefill_review_fields_from_memory(user_id=user_id, payload=draft["payload"]) if ( "revisao_previa_concessionaria" not in draft["payload"] and draft["payload"] and not extracted ): if self._is_affirmative_message(message): draft["payload"]["revisao_previa_concessionaria"] = True elif self._is_negative_message(message): draft["payload"]["revisao_previa_concessionaria"] = False draft["expires_at"] = datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES) self.state.set_entry("pending_review_drafts", user_id, draft) missing = [field for field in REVIEW_REQUIRED_FIELDS if field not in draft["payload"]] if missing: self._log_review_flow_source(source=review_flow_source or "draft", payload=draft["payload"], missing_fields=missing) return self._render_missing_review_fields_prompt(missing) try: tool_result = await self.tool_executor.execute( "agendar_revisao", draft["payload"], user_id=user_id, ) except HTTPException as exc: error = self.tool_executor.coerce_http_error(exc) self._capture_review_confirmation_suggestion( tool_name="agendar_revisao", arguments=draft["payload"], exc=exc, user_id=user_id, ) if error.get("retryable") and error.get("field"): draft["payload"].pop(str(error["field"]), None) draft["expires_at"] = datetime.utcnow() + timedelta(minutes=PENDING_REVIEW_DRAFT_TTL_MINUTES) self.state.set_entry("pending_review_drafts", user_id, draft) self._log_review_flow_source(source=review_flow_source or "draft", payload=draft["payload"]) return self._http_exception_detail(exc) self.state.pop_entry("pending_review_drafts", user_id) self._store_last_review_package(user_id=user_id, payload=draft["payload"]) self._log_review_flow_source(source=review_flow_source or "draft", payload=draft["payload"]) return self._fallback_format_tool_result("agendar_revisao", tool_result)