Autostage Telegram sales leads

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(
+    if not skip_stage_transition:
-        session,
+        _apply_stage(
-        deal,
+            session,
-        _infer_stage_for_channel(payload.channel_type),
+            deal,
-        actor_type=communication_actor_type,
+            _infer_stage_for_channel(payload.channel_type),
-        actor_id=actor_user,
+            actor_type=communication_actor_type,
-        reason=payload.subject or "communication.started",
+            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:

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)