feat(leads): implement hot lead checkout flow with telegram integration

Added an extended checkout form for hot leads natively to the chat widget frontend. Updated backend schema to capture 'surname', 'delivery_address' and 'lead_type'. Modified RAG engine and intent handlers to propagate 'lead_form_type' parameter. Integrated Telegram service to send checkout alerts to the hot leads topic. Fixed tests to accommodate updated telegram notification signature.

Author: PyDevDeep

app/api/v1/endpoints/leads.py

@@ -26,16 +26,16 @@
     wait=wait_exponential(multiplier=1, min=2, max=10),
     retry=retry_if_exception_type(httpx.RequestError),
 )
-async def send_telegram_notification(lead_id: int, message: str) -> None:
+async def send_telegram_notification(lead_id: int, message: str, alert_type: str) -> None:
     settings = get_settings()
     telegram_service = TelegramService(settings)
-    await telegram_service.send_alert(message, alert_type="lead")
+    await telegram_service.send_alert(message, alert_type=alert_type)
 
 
-async def process_lead_background(lead_id: int, message: str) -> None:
+async def process_lead_background(lead_id: int, message: str, alert_type: str = "lead") -> None:
     async with AsyncSessionLocal() as session:
         try:
-            await send_telegram_notification(lead_id, message)
+            await send_telegram_notification(lead_id, message, alert_type)
             # Оновлюємо статус на sent
             lead = await session.get(Lead, lead_id)
             if lead:
@@ -91,22 +91,34 @@ async def create_lead(request: Request, background_tasks: BackgroundTasks) -> di
     async with AsyncSessionLocal() as session:
         db_lead = Lead(
             name=lead_data.name,
+            surname=lead_data.surname,
             phone_number=lead_data.phone_number,
             contact_method=lead_data.contact_method,
+            lead_type=lead_data.lead_type,
+            delivery_address=lead_data.delivery_address,
             notification_status="pending",
         )
         session.add(db_lead)
         await commit_with_retry(session)
         await session.refresh(db_lead)
         lead_id = int(db_lead.id)  # type: ignore
 
-    message = (
-        f"🚨 <b>Новий лід з форми зв'язку!</b>\n\n"
-        f"👤 <b>Ім'я:</b> {lead_data.name}\n"
-        f"📞 <b>Контакт:</b> <code>{lead_data.phone_number}</code>\n"
-        f"📱 <b>Спосіб:</b> {lead_data.contact_method}"
-    )
-
-    background_tasks.add_task(process_lead_background, lead_id, message)
+    if lead_data.lead_type == "checkout":
+        message = (
+            f"🛒 <b>Нове ЗАМОВЛЕННЯ (Hot Lead)!</b>\n\n"
+            f"👤 <b>Ім'я:</b> {lead_data.name} {lead_data.surname or ''}\n"
+            f"📞 <b>Контакт:</b> <code>{lead_data.phone_number}</code>\n"
+            f"📱 <b>Спосіб:</b> {lead_data.contact_method}\n"
+            f"📍 <b>Адреса доставки:</b> {lead_data.delivery_address or 'Не вказана'}"
+        )
+        background_tasks.add_task(process_lead_background, lead_id, message, "hot_lead")
+    else:
+        message = (
+            f"🚨 <b>Новий лід з форми зв'язку!</b>\n\n"
+            f"👤 <b>Ім'я:</b> {lead_data.name}\n"
+            f"📞 <b>Контакт:</b> <code>{lead_data.phone_number}</code>\n"
+            f"📱 <b>Спосіб:</b> {lead_data.contact_method}"
+        )
+        background_tasks.add_task(process_lead_background, lead_id, message, "lead")
 
     return {"status": "success", "message": "Lead received"}

app/models/lead.py

@@ -10,7 +10,10 @@ class Lead(Base):
 
     id = Column(Integer, primary_key=True, index=True)
     name = Column(String(50), nullable=False)
+    surname = Column(String(50), nullable=True)
     phone_number = Column(String(50), nullable=False)
     contact_method = Column(String(20), nullable=False)
+    lead_type = Column(String(20), default="contact")  # contact, checkout
+    delivery_address = Column(String(255), nullable=True)
     notification_status = Column(String(20), default="pending")  # pending, sent, failed
     created_at = Column(DateTime, default=lambda: datetime.datetime.now(datetime.UTC))

app/schemas/chat.py

@@ -1,5 +1,6 @@
 import re
 from dataclasses import dataclass
+from typing import Literal
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -37,6 +38,9 @@ class RAGResponse(BaseModel):
     requires_lead: bool = Field(
         default=False, description="Прапорець для активації форми збору контактів на фронтенді"
     )
+    lead_form_type: Literal["contact", "checkout"] | None = Field(
+        default=None, description="Тип форми для відображення на фронтенді"
+    )
 
 
 class ChatResponse(RAGResponse):
@@ -56,6 +60,7 @@ class PipelineContext:
     sources: list[str]
     extracted_links: list[dict[str, str]]
     requires_lead: bool
+    lead_form_type: Literal["contact", "checkout"] | None
     extended_user_message: str
 
 
@@ -65,4 +70,5 @@ class IntentContextResult:
     system_instructions: list[str]
     extracted_links: list[dict[str, str]]
     requires_lead: bool
+    lead_form_type: Literal["contact", "checkout"] | None = None
     new_intent_type: str | None = None

app/schemas/lead.py

@@ -6,8 +6,13 @@
 
 class ContactFormLead(BaseModel):
     name: str = Field(..., max_length=50, pattern=r"^[A-Za-zА-Яа-яЄєІіЇїҐґ\s\-]+$")
+    surname: str | None = Field(
+        default=None, max_length=50, pattern=r"^[A-Za-zА-Яа-яЄєІіЇїҐґ\s\-]+$"
+    )
     phone_number: str = Field(..., max_length=50, description="Телефон")
     contact_method: Literal["telegram", "viber", "phone"]
+    lead_type: Literal["contact", "checkout"] = Field(default="contact")
+    delivery_address: str | None = Field(default=None, max_length=255)
     honeypot: str | None = Field(default=None, max_length=50)
 
     @field_validator("phone_number", mode="before")

app/services/intent_handlers.py

@@ -56,6 +56,7 @@ async def handle(
     ) -> IntentContextResult:
         is_checkout = intent_type == INTENT_CHECKOUT
         requires_lead = False
+        lead_form_type = None
         new_intent_type = intent_type
 
         logger.info(
@@ -74,6 +75,7 @@ async def handle(
                 system_instructions=system_instructions,
                 extracted_links=extracted_links,
                 requires_lead=requires_lead,
+                lead_form_type=lead_form_type,
                 new_intent_type=INTENT_SEARCH,
             )
 
@@ -111,6 +113,7 @@ async def handle(
 
         elif is_checkout:
             requires_lead = True
+            lead_form_type = "checkout"
             if result.woo_url:
                 extracted_links.append({"text": LINK_CHECKOUT, "url": result.woo_url})
             extracted_links.append(
@@ -156,6 +159,7 @@ async def handle(
             system_instructions=system_instructions,
             extracted_links=extracted_links,
             requires_lead=requires_lead,
+            lead_form_type=lead_form_type,
             new_intent_type=new_intent_type,
         )
 
@@ -173,6 +177,7 @@ async def handle(
         extracted_links: list[dict[str, str]],
     ) -> IntentContextResult:
         requires_lead = False
+        lead_form_type = None
         strict_term = str(
             intent_data.get("strict_query")
             or intent_data.get("search_term")
@@ -269,4 +274,5 @@ async def handle(
             system_instructions=system_instructions,
             extracted_links=extracted_links,
             requires_lead=requires_lead,
+            lead_form_type=lead_form_type,
         )

app/services/rag_engine.py

@@ -152,6 +152,7 @@ async def _get_intent_context(
         product_facts: list[str] = []
         extracted_links: list[dict[str, str]] = []
         requires_lead: bool = False
+        lead_form_type = None
 
         intent_type = intent_data.get("intent", INTENT_FAQ)
         product_name = intent_data.get("product_name")
@@ -191,6 +192,7 @@ async def _get_intent_context(
             system_instructions = res.system_instructions
             extracted_links = res.extracted_links
             requires_lead = res.requires_lead
+            lead_form_type = res.lead_form_type
             intent_type = res.new_intent_type or intent_type
 
             if intent_type == INTENT_SEARCH:
@@ -209,12 +211,14 @@ async def _get_intent_context(
             system_instructions = res.system_instructions
             extracted_links = res.extracted_links
             requires_lead = res.requires_lead
+            lead_form_type = res.lead_form_type
 
         return IntentContextResult(
             product_facts=product_facts,
             system_instructions=system_instructions,
             extracted_links=extracted_links,
             requires_lead=requires_lead,
+            lead_form_type=lead_form_type,
         )
 
     async def _try_capture_lead(
@@ -244,6 +248,7 @@ async def _try_capture_lead(
                     sources=[],
                     extracted_links=links,
                     requires_lead=False,
+                    lead_form_type=None,
                     extended_user_message="",
                 )
             except Exception:
@@ -255,6 +260,7 @@ async def _try_capture_lead(
                     sources=[],
                     extracted_links=[],
                     requires_lead=False,
+                    lead_form_type=None,
                     extended_user_message="",
                 )
         return False, None
@@ -265,7 +271,7 @@ async def _prepare_rag_pipeline(
         """
         Executes the shared pipeline for both sync and stream methods.
         Returns:
-            is_valid, fallback_response, final_context, sources, links, requires_lead, extended_message
+            is_valid, fallback_response, final_context, sources, links, requires_lead, lead_form_type, extended_message
         """
         if not self.guardrails_service.validate_input(question, client_ip=client_ip):
             return PipelineContext(
@@ -275,6 +281,7 @@ async def _prepare_rag_pipeline(
                 sources=[],
                 extracted_links=[],
                 requires_lead=False,
+                lead_form_type=None,
                 extended_user_message="",
             )
 
@@ -342,11 +349,13 @@ async def fetch_vectors() -> list[
         if isinstance(intent_results, BaseException):
             logger.error(f"Intent context retrieval failed: {intent_results}")
             product_facts, system_instructions, extracted_links, requires_lead = [], [], [], False
+            lead_form_type = None
         else:
             product_facts = intent_results.product_facts
             system_instructions = intent_results.system_instructions
             extracted_links = intent_results.extracted_links
             requires_lead = intent_results.requires_lead
+            lead_form_type = intent_results.lead_form_type
 
         prepended_context: list[str] = []
         if system_instructions:
@@ -365,6 +374,7 @@ async def fetch_vectors() -> list[
             sources=list(sources),
             extracted_links=extracted_links,
             requires_lead=requires_lead,
+            lead_form_type=lead_form_type,
             extended_user_message=extended_user_message,
         )
 
@@ -384,6 +394,7 @@ async def process_query(
                     has_context=False,
                     links=[],
                     requires_lead=False,
+                    lead_form_type=None,
                 )
             else:
                 return RAGResponse(
@@ -392,6 +403,7 @@ async def process_query(
                     has_context=False,
                     links=[LinkItem(**link) for link in ctx.extracted_links],
                     requires_lead=False,
+                    lead_form_type=None,
                 )
 
         if not ctx.final_context:
@@ -402,6 +414,7 @@ async def process_query(
                 has_context=False,
                 links=[],
                 requires_lead=True,
+                lead_form_type=None,
             )
 
         try:
@@ -424,6 +437,7 @@ async def process_query(
                     "has_context": False,
                     "links": [],
                     "requires_lead": True,
+                    "lead_form_type": None,
                 }
             )
 
@@ -437,6 +451,7 @@ async def process_query(
                 "has_context": True,
                 "links": ctx.extracted_links,
                 "requires_lead": ctx.requires_lead,
+                "lead_form_type": ctx.lead_form_type,
             }
         )
 
@@ -450,14 +465,20 @@ async def process_query_stream(
 
         if not ctx.is_valid:
             meta_payload = json.dumps(
-                {"links": ctx.extracted_links, "requires_lead": False}, ensure_ascii=False
+                {"links": ctx.extracted_links, "requires_lead": False, "lead_form_type": None},
+                ensure_ascii=False,
             )
             yield f"[METADATA] {meta_payload}"
             yield json.dumps({"token": ctx.fallback_response}, ensure_ascii=False)
             return
 
         meta_payload = json.dumps(
-            {"links": ctx.extracted_links, "requires_lead": ctx.requires_lead}, ensure_ascii=False
+            {
+                "links": ctx.extracted_links,
+                "requires_lead": ctx.requires_lead,
+                "lead_form_type": ctx.lead_form_type,
+            },
+            ensure_ascii=False,
         )
         yield f"[METADATA] {meta_payload}"