diff --git a/services/sales_service/app.py b/services/sales_service/app.py index 613e714..0c9e8a8 100644 --- a/services/sales_service/app.py +++ b/services/sales_service/app.py @@ -237,6 +237,16 @@ def _bool_env(name: str, default: bool = False) -> bool: return raw.strip().lower() in {"1", "true", "yes", "on"} +def _integer_env(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(raw.strip()) + except ValueError: + return default + + def _voice_runtime_enabled() -> bool: return _bool_env("SALES_VOICE_RUNTIME_ENABLED", False) @@ -245,6 +255,10 @@ def _telegram_bridge_enabled() -> bool: return _bool_env("SALES_TELEGRAM_BRIDGE_ENABLED", False) +def _telegram_autostage_message_interval() -> int: + return max(1, _integer_env("SALES_TELEGRAM_STAGE_EVERY_MESSAGES", 3)) + + def _voice_runtime_url() -> str: return os.getenv("AI_VOICE_RUNTIME_SERVICE_URL", "http://localhost:8018").rstrip("/") @@ -1716,6 +1730,89 @@ def _apply_stage_if_needed( ) +def _is_customer_inbound_telegram_message(payload: SalesTelegramSyncIn) -> bool: + direction = str(payload.direction or "").strip().lower() + author_type = str(payload.author_type or "").strip().lower() + return direction == "inbound" and author_type in {"", "customer"} + + +def _next_autostage_pipeline_stage(session, deal: SalesDealRow) -> SalesPipelineStageRow | None: + _, current_stage = _deal_pipeline_stage(session, deal) + if current_stage is None or current_stage.is_terminal: + return None + return session.execute( + select(SalesPipelineStageRow) + .where( + SalesPipelineStageRow.tenant_id == deal.tenant_id, + SalesPipelineStageRow.pipeline_id == deal.pipeline_id, + SalesPipelineStageRow.is_active == True, # noqa: E712 + SalesPipelineStageRow.is_terminal == False, # noqa: E712 + SalesPipelineStageRow.sort_order > current_stage.sort_order, + ) + .order_by(SalesPipelineStageRow.sort_order.asc(), SalesPipelineStageRow.id.asc()) + .limit(1) + ).scalar_one_or_none() + + +def _telegram_customer_message_count(session, deal: SalesDealRow) -> int: + return int( + session.execute( + select(func.count()) + .select_from(SalesMessageRow) + .where( + SalesMessageRow.tenant_id == deal.tenant_id, + SalesMessageRow.deal_id == deal.deal_id, + SalesMessageRow.channel_provider == "telegram", + SalesMessageRow.sender_type == "customer", + SalesMessageRow.delivery_status == "received", + ) + ).scalar_one() + or 0 + ) + + +def _maybe_advance_telegram_deal_stage( + session, + *, + deal: SalesDealRow, + lead: SalesLeadRow | None, + payload: SalesTelegramSyncIn, +) -> None: + if not _is_customer_inbound_telegram_message(payload): + return + interval = _telegram_autostage_message_interval() + message_count = _telegram_customer_message_count(session, deal) + if message_count == 0 or message_count % interval != 0: + return + next_stage = _next_autostage_pipeline_stage(session, deal) + if next_stage is None: + return + _apply_stage( + session, + deal, + next_stage.code, + actor_type="system", + actor_id="telegram-autostage", + reason="telegram.autostage_every_3_messages", + metadata={ + "telegram_message_count": message_count, + "telegram_thread_id": payload.thread_id, + "telegram_chat_id": payload.chat_id, + "interval": interval, + "mode": "deterministic", + }, + force=True, + ) + if lead is not None: + if next_stage.code in {"warm_lead", "hot_lead", "enrichment_required"}: + lead.status = next_stage.code + if next_stage.code == "hot_lead": + lead.lead_temperature = "hot" + elif next_stage.code == "warm_lead": + lead.lead_temperature = "warm" + lead.updated_at = utc_now_iso() + + def _resolve_stage_change_target_code(session, deal: SalesDealRow, payload: SalesDealStageChangeIn) -> str: target_stage_code = str(payload.target_stage_code or payload.stage_code or "").strip() or None target_stage_id = str(payload.target_stage_id or payload.stage_id or "").strip() or None @@ -2216,9 +2313,19 @@ def _touch_deal_contact(deal: SalesDealRow) -> None: deal.updated_at = now -def _start_communication(session, *, deal: SalesDealRow, lead: SalesLeadRow | None, payload: SalesCommunicationStartIn, actor_user: str | None = None, metadata: dict | None = None) -> SalesCommunicationSessionRow: +def _start_communication( + session, + *, + deal: SalesDealRow, + lead: SalesLeadRow | None, + payload: SalesCommunicationStartIn, + actor_user: str | None = None, + metadata: dict | None = None, + skip_stage_transition: bool = False, +) -> SalesCommunicationSessionRow: now = utc_now_iso() metadata_payload = dict(metadata or {}) + skip_stage_transition = skip_stage_transition or bool(metadata_payload.pop("skip_stage_transition", False)) channel_provider = str(metadata_payload.get("channel_provider") or "").strip() if not channel_provider: channel_provider = "voice" if payload.channel_type == "voice" else str(deal.current_channel if deal.current_channel != "voice" else (lead.preferred_channel if lead else "telegram") or "telegram") @@ -2271,14 +2378,15 @@ def _start_communication(session, *, deal: SalesDealRow, lead: SalesLeadRow | No "subject": communication.subject, }, ) - _apply_stage( - session, - deal, - _infer_stage_for_channel(payload.channel_type), - actor_type=communication_actor_type, - actor_id=actor_user, - reason=payload.subject or "communication.started", - ) + if not skip_stage_transition: + _apply_stage( + session, + deal, + _infer_stage_for_channel(payload.channel_type), + actor_type=communication_actor_type, + actor_id=actor_user, + reason=payload.subject or "communication.started", + ) if payload.next_action_type: _schedule_task( session, @@ -2607,6 +2715,8 @@ def _get_or_create_text_communication( metadata: dict | None = None, thread_id: str | None = None, ) -> SalesCommunicationSessionRow: + metadata_payload = dict(metadata or {}) + skip_stage_transition = bool(metadata_payload.pop("skip_stage_transition", False)) if str(thread_id or "").strip(): link = _find_external_link(session, tenant_id=deal.tenant_id, thread_id=thread_id) if link and link.communication_id: @@ -2617,7 +2727,7 @@ def _get_or_create_text_communication( ) ).scalar_one_or_none() if existing is not None: - _merge_communication_metadata(existing, metadata or {}) + _merge_communication_metadata(existing, metadata_payload) return existing existing = session.execute( select(SalesCommunicationSessionRow) @@ -2626,7 +2736,7 @@ def _get_or_create_text_communication( .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc()) ).scalars().first() if existing is not None: - _merge_communication_metadata(existing, metadata or {}) + _merge_communication_metadata(existing, metadata_payload) return existing return _start_communication( session, @@ -2638,7 +2748,8 @@ def _get_or_create_text_communication( agent_type="human" if direction == "outbound" else "text_ai", subject=subject, ), - metadata=metadata, + metadata=metadata_payload, + skip_stage_transition=skip_stage_transition, ) @@ -4986,6 +5097,8 @@ def sync_telegram_thread( subject=(payload.text or payload.display_name or "Telegram conversation")[:120], metadata={ "channel_provider": "telegram", + "skip_stage_transition": True, + "stage_progression_mode": "telegram_message_count", "telegram_thread_id": payload.thread_id, "telegram_chat_id": payload.chat_id, "interaction_id": payload.interaction_id, @@ -5007,6 +5120,7 @@ def sync_telegram_thread( ).scalar_one_or_none() else: message = None + message_created = False if message is None and str(payload.text or "").strip(): message = SalesMessageRow( message_id=new_id("msg"), @@ -5026,6 +5140,7 @@ def sync_telegram_thread( created_at=event_ts, ) session.add(message) + message_created = True _upsert_external_link( session, @@ -5051,6 +5166,9 @@ def sync_telegram_thread( deal.current_channel = "telegram" deal.preferred_channel = "telegram" deal.updated_at = utc_now_iso() + if message_created: + session.flush() + _maybe_advance_telegram_deal_stage(session, deal=deal, lead=lead, payload=payload) session.commit() return _build_workspace(session, deal) finally: diff --git a/tests/test_sales_service.py b/tests/test_sales_service.py index 56aee94..59a5a37 100644 --- a/tests/test_sales_service.py +++ b/tests/test_sales_service.py @@ -223,6 +223,44 @@ def test_sales_internal_telegram_sync_auto_creates_workspace(): assert workspace["messages"][0]["external_message_id"] == "ext_tg_auto_01" +def test_sales_internal_telegram_sync_moves_pipeline_every_three_customer_messages(): + client = TestClient(sales_module.app) + tenant_id = "tenant_tg_autostage" + + def post_message(index: int) -> dict: + response = client.post( + "/internal/sales-sync/telegram", + json={ + "thread_id": "tg-thread-autostage-01", + "chat_id": "tg-chat-autostage-01", + "interaction_id": "int_tg_autostage_01", + "display_name": "Autostage Prospect", + "message_id": f"msg_tg_autostage_{index}", + "external_message_id": f"ext_tg_autostage_{index}", + "text": f"Сообщение клиента {index}", + "direction": "inbound", + "author_type": "customer", + "author_id": "tg-user-autostage", + }, + headers=_headers(tenant_id), + ) + assert response.status_code == 200 + return response.json() + + stage_codes = [post_message(index)["stage"]["code"] for index in range(1, 7)] + + assert stage_codes[0] == "new_qualified_lead" + assert stage_codes[1] == "new_qualified_lead" + assert stage_codes[2] == "warm_lead" + assert stage_codes[3] == "warm_lead" + assert stage_codes[4] == "warm_lead" + assert stage_codes[5] == "hot_lead" + + duplicate = post_message(6) + assert duplicate["stage"]["code"] == "hot_lead" + assert len(duplicate["messages"]) == 6 + + def test_sales_internal_voice_sync_creates_call_and_transcript(): client = TestClient(sales_module.app)