from __future__ import annotations import json import os from datetime import datetime, timezone from typing import Any import httpx from fastapi import Depends, FastAPI, Header, HTTPException, Query from sqlalchemy import func, or_, select from sqlalchemy.exc import IntegrityError from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import get_session from services.shared.models import HealthResponse from services.shared.sales_models import ( SalesAutomationTaskOut, SalesCallCompleteIn, SalesCallOut, SalesCallWebhookIn, SalesCommunicationBindExternalIn, SalesChannelSwitchOut, SalesCommunicationOut, SalesCommunicationStartIn, SalesCommunicationSummaryIn, SalesCommunicationSwitchChannelIn, SalesConditionOut, SalesConditionUpsertIn, SalesCounterpartyOut, SalesCounterpartyUpsertIn, SalesDashboardStageOut, SalesDashboardSummaryOut, SalesDealCloseIn, SalesDealCreate, SalesDealNextActionIn, SalesDealOut, SalesDealScenarioIn, SalesDealStageChangeIn, SalesDealUpdate, SalesDocumentCreate, SalesDocumentOut, SalesEscalationCreateIn, SalesEscalationOut, SalesInvoiceCreate, SalesInvoiceOut, SalesLeadCreate, SalesLeadEnrichIn, SalesLeadOut, SalesLeadUpdate, SalesMessageOut, SalesMessageSendIn, SalesMessageWebhookIn, SalesOfferCreate, SalesOfferOut, SalesPaymentOut, SalesPaymentReconcileIn, SalesPaymentWebhookIn, SalesPipelineCreate, SalesPipelineOut, SalesPipelineStageCreate, SalesPipelineStageOut, SalesPipelineStageUpdate, SalesPipelineUpdate, SalesStageHistoryOut, SalesStageRefOut, SalesTelegramSyncIn, SalesTimelineEventOut, SalesTranscriptIn, SalesTranscriptOut, SalesVoiceSyncIn, SalesWorkspaceOut, ) from services.shared.security import get_actor, issue_app_token, require_roles from services.shared.sql_init import init_sql_schema from services.shared.sql_models import Customer from services.shared.sales_sql_models import ( SalesAutomationTaskRow, SalesCallRow, SalesChannelSwitchRow, SalesCommunicationSessionRow, SalesConditionRow, SalesCounterpartyRow, SalesDealRow, SalesDocumentRow, SalesEscalationRow, SalesExternalLinkRow, SalesInvoiceRow, SalesLeadRow, SalesMessageRow, SalesOfferRow, SalesPaymentRow, SalesPipelineRow, SalesPipelineStageRow, SalesStageHistoryRow, SalesTranscriptRow, TenantIntegrationRow, TenantSalesSettingsRow, ) from services.sales_service import sales_events as sales_event_types from services.sales_service.event_publisher import SalesEventPublisher # Import the models module so sales tables are registered on the shared Base metadata. from services.shared import sales_sql_models as _sales_sql_models # noqa: F401 app = FastAPI(title="sales-service", version="1.0.0") init_sql_schema() DEFAULT_SALES_CURRENCY = os.getenv("DEFAULT_SALES_CURRENCY", "KZT").strip() or "KZT" DEFAULT_PIPELINE_CODE = "default_sales" DEFAULT_PIPELINE_NAME = "Базовая воронка продаж" DEFAULT_STAGE_CODE = "new_qualified_lead" LEGACY_STAGE_ALIASES = { "new": "new_qualified_lead", "active_text": "active_text_communication", "invoice": "invoice_sent", "paid": "paid", "won": "won", } DEFAULT_PIPELINE_STAGES = [ {"code": "new_qualified_lead", "name": "Новый квалифицированный лид", "category": "entry", "sort_order": 10}, {"code": "warm_lead", "name": "Теплый лид", "category": "entry", "sort_order": 20}, {"code": "hot_lead", "name": "Горячий лид", "category": "entry", "sort_order": 30}, {"code": "enrichment_required", "name": "Нужно дообогащение", "category": "entry", "sort_order": 40}, {"code": "active_text_communication", "name": "Текстовая коммуникация", "category": "communication", "sort_order": 50}, {"code": "active_voice_communication", "name": "Голосовая коммуникация", "category": "communication", "sort_order": 60}, {"code": "waiting_customer_reply", "name": "Ждем ответ клиента", "category": "communication", "sort_order": 70}, {"code": "need_clarification", "name": "Нужно уточнение", "category": "communication", "sort_order": 80}, {"code": "need_confirmed", "name": "Потребность подтверждена", "category": "communication", "sort_order": 90}, {"code": "offer_selection", "name": "Подбор предложения", "category": "commercial", "sort_order": 100}, {"code": "offer_preparing", "name": "Готовим предложение", "category": "commercial", "sort_order": 110}, {"code": "offer_sent", "name": "Предложение отправлено", "category": "commercial", "sort_order": 120}, {"code": "conditions_negotiation", "name": "Согласование условий", "category": "commercial", "sort_order": 130}, {"code": "counterparty_data_requested", "name": "Запрос реквизитов", "category": "paperwork", "sort_order": 140}, {"code": "counterparty_data_received", "name": "Реквизиты получены", "category": "paperwork", "sort_order": 150}, {"code": "document_preparing", "name": "Готовим документ", "category": "paperwork", "sort_order": 160}, {"code": "document_sent", "name": "Документ отправлен", "category": "paperwork", "sort_order": 170}, {"code": "document_under_review", "name": "Документ на согласовании", "category": "paperwork", "sort_order": 180}, {"code": "document_confirmed", "name": "Документ подтвержден", "category": "paperwork", "sort_order": 190}, {"code": "invoice_preparing", "name": "Готовим счет", "category": "finance", "sort_order": 200}, {"code": "invoice_sent", "name": "Счет отправлен", "category": "finance", "sort_order": 210}, {"code": "payment_expected", "name": "Ожидаем оплату", "category": "finance", "sort_order": 220}, {"code": "partially_paid", "name": "Оплачено частично", "category": "finance", "sort_order": 230}, {"code": "paid", "name": "Оплачено", "category": "finance", "sort_order": 240}, {"code": "payment_overdue", "name": "Оплата просрочена", "category": "finance", "sort_order": 250}, {"code": "won", "name": "Сделка выиграна", "category": "closing", "sort_order": 260, "is_terminal": True}, {"code": "lost", "name": "Сделка проиграна", "category": "closing", "sort_order": 270, "is_terminal": True}, {"code": "postponed", "name": "Отложено", "category": "closing", "sort_order": 280, "is_terminal": True}, {"code": "follow_up_scheduled", "name": "Запланирован follow-up", "category": "closing", "sort_order": 290}, {"code": "transferred_to_execution", "name": "Передано в исполнение", "category": "closing", "sort_order": 300, "is_terminal": True}, {"code": "transferred_to_support", "name": "Передано человеку", "category": "closing", "sort_order": 310, "is_terminal": True}, ] def _tenant_id(actor: dict) -> str: tenant_id = str(actor.get("tenant_id") or "").strip() if not tenant_id: raise HTTPException(status_code=400, detail="Tenant context is required") return tenant_id def _optional_tenant_id(actor: dict) -> str | None: tenant_id = str(actor.get("tenant_id") or "").strip() return tenant_id or None def _now() -> datetime: return datetime.now(timezone.utc).replace(microsecond=0) def _json_dict(raw: str | None) -> dict: try: payload = json.loads(raw or "{}") except json.JSONDecodeError: return {} return payload if isinstance(payload, dict) else {} def _json_list(raw: str | None) -> list: try: payload = json.loads(raw or "[]") except json.JSONDecodeError: return [] return payload if isinstance(payload, list) else [] def _service_headers(subject: str, username: str, provider: str, tenant_id: str | None = None) -> dict[str, str]: token = issue_app_token( subject=subject, username=username, role="admin", auth_source="service", provider=provider, tenant_id=tenant_id, ttl_seconds=300, ) return {"Authorization": f"Bearer {token}"} def _bool_env(name: str, default: bool = False) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _voice_runtime_enabled() -> bool: return _bool_env("SALES_VOICE_RUNTIME_ENABLED", False) def _telegram_bridge_enabled() -> bool: return _bool_env("SALES_TELEGRAM_BRIDGE_ENABLED", False) def _voice_runtime_url() -> str: return os.getenv("AI_VOICE_RUNTIME_SERVICE_URL", "http://localhost:8018").rstrip("/") def _telegram_adapter_url() -> str: return os.getenv("TELEGRAM_ADAPTER_SERVICE_URL", "http://localhost:8007").rstrip("/") def _default_handoff_queue_id() -> str: return os.getenv("SALES_DEFAULT_HANDOFF_QUEUE_ID", "sales_line").strip() or "sales_line" def _infer_text_channel(preferred_channel: str | None) -> str: return "voice" if (preferred_channel or "").strip().lower() == "voice" else "text" def _infer_stage_for_channel(channel: str) -> str: return "active_voice_communication" if channel == "voice" else "active_text_communication" def _normalize_stage_code(value: str | None) -> str: code = str(value or DEFAULT_STAGE_CODE).strip() return LEGACY_STAGE_ALIASES.get(code, code) def _pipeline_to_out(row: SalesPipelineRow) -> SalesPipelineOut: return SalesPipelineOut( pipeline_id=row.pipeline_id, tenant_id=row.tenant_id, code=row.code, name=row.name, description=row.description, is_default=row.is_default, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) def _stage_to_ref(row: SalesPipelineStageRow | None) -> SalesStageRefOut | None: if row is None: return None return SalesStageRefOut( id=row.stage_id, code=row.code, name=row.name, category=row.category, # type: ignore[arg-type] sort_order=row.sort_order, ) def _stage_to_out(row: SalesPipelineStageRow) -> SalesPipelineStageOut: return SalesPipelineStageOut( stage_id=row.stage_id, tenant_id=row.tenant_id, pipeline_id=row.pipeline_id, code=row.code, name=row.name, category=row.category, # type: ignore[arg-type] sort_order=row.sort_order, is_terminal=row.is_terminal, is_system=row.is_system, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) def _get_pipeline(session, pipeline_id: str, tenant_id: str) -> SalesPipelineRow: row = session.execute( select(SalesPipelineRow).where( SalesPipelineRow.pipeline_id == pipeline_id, SalesPipelineRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Pipeline not found") return row def _get_stage(session, stage_id: str, tenant_id: str, pipeline_id: str | None = None) -> SalesPipelineStageRow: stmt = select(SalesPipelineStageRow).where( SalesPipelineStageRow.stage_id == stage_id, SalesPipelineStageRow.tenant_id == tenant_id, ) row = session.execute(stmt).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Stage not found") if pipeline_id and row.pipeline_id != pipeline_id: raise HTTPException(status_code=400, detail="Stage does not belong to pipeline") return row def _set_default_pipeline(session, pipeline: SalesPipelineRow) -> None: now = utc_now_iso() rows = session.execute( select(SalesPipelineRow).where( SalesPipelineRow.tenant_id == pipeline.tenant_id, SalesPipelineRow.is_default == True, # noqa: E712 SalesPipelineRow.pipeline_id != pipeline.pipeline_id, ) ).scalars().all() for row in rows: row.is_default = False row.updated_at = now pipeline.is_default = True pipeline.is_active = True pipeline.updated_at = now settings = session.execute( select(TenantSalesSettingsRow).where(TenantSalesSettingsRow.tenant_id == pipeline.tenant_id) ).scalar_one_or_none() if settings is None: settings = TenantSalesSettingsRow( tenant_id=pipeline.tenant_id, default_pipeline_id=pipeline.pipeline_id, default_language="ru", default_currency=DEFAULT_SALES_CURRENCY, enabled_text_channels_json="[]", enabled_voice_channels_json="[]", payment_provider_settings_json="{}", document_settings_json="{}", created_at=now, updated_at=now, ) session.add(settings) else: settings.default_pipeline_id = pipeline.pipeline_id settings.updated_at = now def ensure_default_stages(session, tenant_id: str, pipeline_id: str) -> list[SalesPipelineStageRow]: now = utc_now_iso() existing = { row.code: row for row in session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.pipeline_id == pipeline_id, ) ).scalars().all() } for spec in DEFAULT_PIPELINE_STAGES: if spec["code"] in existing: continue row = SalesPipelineStageRow( stage_id=new_id("pst"), tenant_id=tenant_id, pipeline_id=pipeline_id, code=spec["code"], name=spec["name"], category=spec["category"], sort_order=int(spec["sort_order"]), is_terminal=bool(spec.get("is_terminal", False)), is_system=True, is_active=True, created_at=now, updated_at=now, ) session.add(row) existing[row.code] = row session.flush() return list(existing.values()) def _backfill_legacy_stage_values( session, *, tenant_id: str, pipeline: SalesPipelineRow, stages: list[SalesPipelineStageRow], ) -> None: stage_by_code = {row.code: row for row in stages} tenant_stages_by_id = { row.stage_id: row for row in session.execute( select(SalesPipelineStageRow).where(SalesPipelineStageRow.tenant_id == tenant_id) ).scalars().all() } tenant_stage_ids = set(tenant_stages_by_id) default_stage = stage_by_code[DEFAULT_STAGE_CODE] legacy_pipeline_values = {"", "sales_default", DEFAULT_PIPELINE_CODE} for deal in session.execute(select(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id)).scalars().all(): if deal.stage_id in tenant_stage_ids: stage = tenant_stages_by_id[deal.stage_id] if deal.pipeline_id != stage.pipeline_id: deal.pipeline_id = stage.pipeline_id deal.updated_at = utc_now_iso() continue code = _normalize_stage_code(deal.stage_id) stage = stage_by_code.get(code, default_stage) deal.stage_id = stage.stage_id if deal.pipeline_id in legacy_pipeline_values or deal.pipeline_id != pipeline.pipeline_id: deal.pipeline_id = pipeline.pipeline_id deal.updated_at = utc_now_iso() for history in session.execute(select(SalesStageHistoryRow).where(SalesStageHistoryRow.tenant_id == tenant_id)).scalars().all(): if history.from_stage_id and history.from_stage_id not in tenant_stage_ids: from_code = _normalize_stage_code(history.from_stage_id) history.from_stage_id = stage_by_code.get(from_code, default_stage).stage_id if history.to_stage_id not in tenant_stage_ids: to_code = _normalize_stage_code(history.to_stage_id) history.to_stage_id = stage_by_code.get(to_code, default_stage).stage_id def ensure_default_pipeline(session, tenant_id: str) -> SalesPipelineRow: pipeline = session.execute( select(SalesPipelineRow) .where( SalesPipelineRow.tenant_id == tenant_id, SalesPipelineRow.is_default == True, # noqa: E712 SalesPipelineRow.is_active == True, # noqa: E712 ) .order_by(SalesPipelineRow.id.asc()) ).scalars().first() if pipeline is None: pipeline = session.execute( select(SalesPipelineRow).where( SalesPipelineRow.tenant_id == tenant_id, SalesPipelineRow.code == DEFAULT_PIPELINE_CODE, ) ).scalar_one_or_none() if pipeline is None: now = utc_now_iso() pipeline = SalesPipelineRow( pipeline_id=new_id("pip"), tenant_id=tenant_id, code=DEFAULT_PIPELINE_CODE, name=DEFAULT_PIPELINE_NAME, description=None, is_default=True, is_active=True, created_at=now, updated_at=now, ) session.add(pipeline) session.flush() if not pipeline.is_default or not pipeline.is_active: _set_default_pipeline(session, pipeline) stages = ensure_default_stages(session, tenant_id, pipeline.pipeline_id) _backfill_legacy_stage_values(session, tenant_id=tenant_id, pipeline=pipeline, stages=stages) return pipeline def get_default_pipeline(session, tenant_id: str) -> SalesPipelineRow: return ensure_default_pipeline(session, tenant_id) def resolve_stage_by_code_or_id( session, *, tenant_id: str, pipeline_id: str, stage_id: str | None = None, stage_code: str | None = None, require_active: bool = True, ) -> SalesPipelineStageRow: ensure_default_pipeline(session, tenant_id) def resolve_one(raw: str | None) -> SalesPipelineStageRow: value = str(raw or DEFAULT_STAGE_CODE).strip() stage = session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.stage_id == value, ) ).scalar_one_or_none() if stage is not None: if stage.pipeline_id != pipeline_id: raise HTTPException(status_code=400, detail="Stage does not belong to pipeline") if require_active and not stage.is_active: raise HTTPException(status_code=400, detail="Stage is inactive") return stage code = _normalize_stage_code(value) stage = session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.pipeline_id == pipeline_id, SalesPipelineStageRow.code == code, ) ).scalar_one_or_none() if stage is None: raise HTTPException(status_code=404, detail="Stage not found") if require_active and not stage.is_active: raise HTTPException(status_code=400, detail="Stage is inactive") return stage by_id_or_code = resolve_one(stage_id or stage_code or DEFAULT_STAGE_CODE) if stage_id and stage_code: by_code = resolve_one(stage_code) if by_id_or_code.stage_id != by_code.stage_id: raise HTTPException(status_code=400, detail="stage_id and stage_code point to different stages") return by_id_or_code def _stage_filter_ids(session, tenant_id: str, value: str) -> list[str]: normalized = str(value or "").strip() if not normalized: return [] direct = session.execute( select(SalesPipelineStageRow.stage_id).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.stage_id == normalized, ) ).scalars().all() if direct: return list(direct) code = _normalize_stage_code(normalized) return list( session.execute( select(SalesPipelineStageRow.stage_id).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.code == code, ) ).scalars().all() ) def _stage_refs_by_id(session, tenant_id: str, stage_ids: list[str | None]) -> dict[str, SalesPipelineStageRow]: ids = [stage_id for stage_id in stage_ids if stage_id] if not ids: return {} rows = session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.stage_id.in_(ids), ) ).scalars().all() return {row.stage_id: row for row in rows} def _deal_pipeline_stage(session, deal: SalesDealRow) -> tuple[SalesPipelineRow | None, SalesPipelineStageRow | None]: pipeline = session.execute( select(SalesPipelineRow).where( SalesPipelineRow.tenant_id == deal.tenant_id, SalesPipelineRow.pipeline_id == deal.pipeline_id, ) ).scalar_one_or_none() stage = session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == deal.tenant_id, SalesPipelineStageRow.stage_id == deal.stage_id, ) ).scalar_one_or_none() return pipeline, stage def _deal_to_out_with_refs(session, deal: SalesDealRow) -> SalesDealOut: pipeline, stage = _deal_pipeline_stage(session, deal) return _deal_to_out(deal, pipeline=pipeline, stage=stage) def _publish_sales_event( session, *, tenant_id: str, event_type: str, aggregate_type: str, aggregate_id: str, payload: dict[str, Any], actor_type: str | None = None, actor_id: str | None = None, correlation_id: str | None = None, causation_id: str | None = None, ) -> None: SalesEventPublisher.publish_sales_event( session, tenant_id=tenant_id, event_type=event_type, aggregate_type=aggregate_type, aggregate_id=aggregate_id, payload=payload, actor_type=actor_type, actor_id=actor_id, correlation_id=correlation_id, causation_id=causation_id, ) def _stage_by_id_or_none(session, tenant_id: str, stage_id: str | None) -> SalesPipelineStageRow | None: if not stage_id: return None return session.execute( select(SalesPipelineStageRow).where( SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.stage_id == stage_id, ) ).scalar_one_or_none() def _publish_lead_entered_event( session, *, lead: SalesLeadRow, deal: SalesDealRow, stage: SalesPipelineStageRow, actor_type: str, actor_id: str | None, ) -> None: _publish_sales_event( session, tenant_id=lead.tenant_id, event_type=sales_event_types.LEAD_ENTERED_CRM, aggregate_type="lead", aggregate_id=lead.lead_id, actor_type=actor_type, actor_id=actor_id, payload={ "lead_id": lead.lead_id, "deal_id": deal.deal_id, "source_type": lead.source_type, "source_channel": lead.source_channel, "lead_temperature": lead.lead_temperature, "initial_stage_id": stage.stage_id, "initial_stage_code": stage.code, }, ) def _lead_to_out(row: SalesLeadRow) -> SalesLeadOut: return SalesLeadOut( lead_id=row.lead_id, tenant_id=row.tenant_id, source_type=row.source_type, source_channel=row.source_channel, source_campaign_id=row.source_campaign_id, full_name=row.full_name, company_name=row.company_name, phone=row.phone, email=row.email, messenger_handles=_json_dict(row.messenger_handles_json), lead_temperature=row.lead_temperature, # type: ignore[arg-type] lead_score=row.lead_score, customer_type=row.customer_type, segment_type=row.segment_type, initial_need_summary=row.initial_need_summary, preferred_channel=row.preferred_channel, # type: ignore[arg-type] assigned_agent_type=row.assigned_agent_type, # type: ignore[arg-type] status=row.status, # type: ignore[arg-type] crm_customer_id=row.crm_customer_id, created_at=row.created_at, updated_at=row.updated_at, ) def _deal_to_out( row: SalesDealRow, *, pipeline: SalesPipelineRow | None = None, stage: SalesPipelineStageRow | None = None, ) -> SalesDealOut: return SalesDealOut( deal_id=row.deal_id, tenant_id=row.tenant_id, lead_id=row.lead_id, customer_id=row.customer_id, pipeline_id=row.pipeline_id, stage_id=row.stage_id, pipeline=_pipeline_to_out(pipeline) if pipeline else None, stage=_stage_to_ref(stage), scenario_type=row.scenario_type, # type: ignore[arg-type] priority=row.priority, title=row.title, need_summary=row.need_summary, product_context=_json_dict(row.product_context_json), estimated_amount=row.estimated_amount, final_amount=row.final_amount, currency=row.currency, payment_model=row.payment_model, document_required=row.document_required, payment_required=row.payment_required, assigned_human_user_id=row.assigned_human_user_id, assigned_ai_orchestrator_id=row.assigned_ai_orchestrator_id, preferred_channel=row.preferred_channel, current_channel=row.current_channel, status=row.status, # type: ignore[arg-type] won_reason=row.won_reason, lost_reason=row.lost_reason, close_reason=row.close_reason, next_action_type=row.next_action_type, next_action_at=row.next_action_at, last_contact_at=row.last_contact_at, closed_at=row.closed_at, created_at=row.created_at, updated_at=row.updated_at, ) def _communication_to_out(row: SalesCommunicationSessionRow) -> SalesCommunicationOut: return SalesCommunicationOut( communication_id=row.communication_id, tenant_id=row.tenant_id, deal_id=row.deal_id, lead_id=row.lead_id, customer_id=row.customer_id, channel_type=row.channel_type, # type: ignore[arg-type] direction=row.direction, # type: ignore[arg-type] agent_type=row.agent_type, # type: ignore[arg-type] started_at=row.started_at, ended_at=row.ended_at, duration_sec=row.duration_sec, subject=row.subject, status=row.status, summary=row.summary, transcript_id=row.transcript_id, next_action_type=row.next_action_type, next_action_at=row.next_action_at, sentiment=row.sentiment, result_code=row.result_code, metadata=_json_dict(row.metadata_json), created_at=row.created_at, updated_at=row.updated_at, ) def _message_to_out(row: SalesMessageRow) -> SalesMessageOut: return SalesMessageOut( message_id=row.message_id, deal_id=row.deal_id, communication_id=row.communication_id, sender_type=row.sender_type, sender_id=row.sender_id, channel_provider=row.channel_provider, external_message_id=row.external_message_id, body=row.body, attachments=_json_list(row.attachments_json), delivery_status=row.delivery_status, read_status=row.read_status, metadata=_json_dict(row.message_metadata_json), sent_at=row.sent_at, created_at=row.created_at, ) def _call_to_out(row: SalesCallRow) -> SalesCallOut: return SalesCallOut( call_id=row.call_id, deal_id=row.deal_id, communication_id=row.communication_id, phone_number=row.phone_number, direction=row.direction, # type: ignore[arg-type] provider=row.provider, external_call_id=row.external_call_id, recording_url=row.recording_url, transcript_status=row.transcript_status, # type: ignore[arg-type] transcript_id=row.transcript_id, call_status=row.call_status, started_at=row.started_at, ended_at=row.ended_at, duration_sec=row.duration_sec, summary=row.summary, created_at=row.created_at, updated_at=row.updated_at, ) def _transcript_to_out(row: SalesTranscriptRow) -> SalesTranscriptOut: return SalesTranscriptOut( transcript_id=row.transcript_id, call_id=row.call_id, language=row.language, transcript_text=row.transcript_text, diarization=_json_dict(row.diarization_json), extracted_entities=_json_dict(row.extracted_entities_json), created_at=row.created_at, updated_at=row.updated_at, ) def _offer_to_out(row: SalesOfferRow) -> SalesOfferOut: return SalesOfferOut( offer_id=row.offer_id, deal_id=row.deal_id, offer_type=row.offer_type, # type: ignore[arg-type] title=row.title, description=row.description, line_items=_json_list(row.line_items_json), pricing=_json_dict(row.pricing_json), total_amount=row.total_amount, currency=row.currency, validity_until=row.validity_until, status=row.status, # type: ignore[arg-type] rendered_document_url=row.rendered_document_url, created_by_type=row.created_by_type, created_at=row.created_at, updated_at=row.updated_at, ) def _condition_to_out(row: SalesConditionRow) -> SalesConditionOut: return SalesConditionOut( condition_id=row.condition_id, deal_id=row.deal_id, product_name=row.product_name, service_name=row.service_name, quantity=row.quantity, unit=row.unit, delivery_mode=row.delivery_mode, execution_date=row.execution_date, start_date=row.start_date, end_date=row.end_date, payment_terms=row.payment_terms, custom_terms=_json_dict(row.custom_terms_json), agreed_price=row.agreed_price, currency=row.currency, confirmed_at=row.confirmed_at, created_at=row.created_at, updated_at=row.updated_at, ) def _counterparty_to_out(row: SalesCounterpartyRow) -> SalesCounterpartyOut: return SalesCounterpartyOut( counterparty_id=row.counterparty_id, deal_id=row.deal_id, customer_id=row.customer_id, company_name=row.company_name, full_name=row.full_name, bin_iin=row.bin_iin, address=row.address, bank_details=_json_dict(row.bank_details_json), signer_name=row.signer_name, signer_role=row.signer_role, signer_basis=row.signer_basis, email_for_docs=row.email_for_docs, phone_for_docs=row.phone_for_docs, completeness_status=row.completeness_status, created_at=row.created_at, updated_at=row.updated_at, ) def _document_to_out(row: SalesDocumentRow) -> SalesDocumentOut: return SalesDocumentOut( document_id=row.document_id, deal_id=row.deal_id, customer_id=row.customer_id, document_type=row.document_type, # type: ignore[arg-type] template_id=row.template_id, version=row.version, status=row.status, # type: ignore[arg-type] file_url=row.file_url, rendered_payload=_json_dict(row.rendered_payload_json), external_sign_provider_id=row.external_sign_provider_id, signed_at=row.signed_at, created_at=row.created_at, updated_at=row.updated_at, ) def _invoice_to_out(row: SalesInvoiceRow) -> SalesInvoiceOut: return SalesInvoiceOut( invoice_id=row.invoice_id, deal_id=row.deal_id, customer_id=row.customer_id, invoice_number=row.invoice_number, basis_document_id=row.basis_document_id, amount=row.amount, currency=row.currency, due_date=row.due_date, status=row.status, # type: ignore[arg-type] payment_link=row.payment_link, line_items=_json_list(row.line_items_json), metadata=_json_dict(row.metadata_json), issued_at=row.issued_at, paid_at=row.paid_at, created_at=row.created_at, updated_at=row.updated_at, ) def _payment_to_out(row: SalesPaymentRow) -> SalesPaymentOut: return SalesPaymentOut( payment_id=row.payment_id, deal_id=row.deal_id, invoice_id=row.invoice_id, payment_provider=row.payment_provider, external_payment_id=row.external_payment_id, amount=row.amount, currency=row.currency, status=row.status, # type: ignore[arg-type] paid_at=row.paid_at, payment_method=row.payment_method, failure_reason=row.failure_reason, metadata=_json_dict(row.metadata_json), created_at=row.created_at, updated_at=row.updated_at, ) def _escalation_to_out(row: SalesEscalationRow) -> SalesEscalationOut: return SalesEscalationOut( escalation_id=row.escalation_id, deal_id=row.deal_id, escalation_type=row.escalation_type, reason=row.reason, severity=row.severity, # type: ignore[arg-type] status=row.status, # type: ignore[arg-type] assigned_to_user_id=row.assigned_to_user_id, created_at=row.created_at, resolved_at=row.resolved_at, ) def _task_to_out(row: SalesAutomationTaskRow) -> SalesAutomationTaskOut: return SalesAutomationTaskOut( task_id=row.task_id, deal_id=row.deal_id, task_type=row.task_type, payload=_json_dict(row.payload_json), run_at=row.run_at, status=row.status, # type: ignore[arg-type] retry_count=row.retry_count, last_error=row.last_error, created_at=row.created_at, updated_at=row.updated_at, ) def _stage_history_to_out(row: SalesStageHistoryRow, stage_refs: dict[str, SalesPipelineStageRow] | None = None) -> SalesStageHistoryOut: refs = stage_refs or {} return SalesStageHistoryOut( history_id=row.history_id, deal_id=row.deal_id, from_stage_id=row.from_stage_id, to_stage_id=row.to_stage_id, from_stage=_stage_to_ref(refs.get(row.from_stage_id or "")), to_stage=_stage_to_ref(refs.get(row.to_stage_id)), changed_by_type=row.changed_by_type, changed_by_id=row.changed_by_id, reason=row.reason, changed_at=row.changed_at, ) def _channel_switch_to_out(row: SalesChannelSwitchRow) -> SalesChannelSwitchOut: return SalesChannelSwitchOut( switch_id=row.switch_id, deal_id=row.deal_id, communication_id=row.communication_id, from_channel=row.from_channel, to_channel=row.to_channel, reason_for_channel_switch=row.reason_for_channel_switch, switched_at=row.switched_at, ) def _get_lead(session, lead_id: str, tenant_id: str) -> SalesLeadRow: row = session.execute( select(SalesLeadRow).where(SalesLeadRow.lead_id == lead_id, SalesLeadRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Lead not found") return row def _get_deal(session, deal_id: str, tenant_id: str) -> SalesDealRow: row = session.execute( select(SalesDealRow).where(SalesDealRow.deal_id == deal_id, SalesDealRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Deal not found") return row def _get_communication(session, communication_id: str, tenant_id: str) -> SalesCommunicationSessionRow: row = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == communication_id, SalesCommunicationSessionRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Communication not found") return row def _find_external_link( session, *, tenant_id: str, thread_id: str | None = None, voice_session_id: str | None = None, external_call_id: str | None = None, interaction_id: str | None = None, ) -> SalesExternalLinkRow | None: conditions = [] if str(thread_id or "").strip(): conditions.append(SalesExternalLinkRow.external_thread_id == str(thread_id).strip()) if str(voice_session_id or "").strip(): conditions.append(SalesExternalLinkRow.voice_session_id == str(voice_session_id).strip()) if str(external_call_id or "").strip(): conditions.append(SalesExternalLinkRow.external_call_id == str(external_call_id).strip()) if str(interaction_id or "").strip(): conditions.append(SalesExternalLinkRow.interaction_id == str(interaction_id).strip()) if not conditions: return None return session.execute( select(SalesExternalLinkRow) .where(SalesExternalLinkRow.tenant_id == tenant_id) .where(or_(*conditions)) .order_by(SalesExternalLinkRow.id.desc()) ).scalars().first() def _metadata_value(metadata: dict | None, *keys: str) -> str | None: payload = metadata or {} for key in keys: value = payload.get(key) normalized = str(value or "").strip() if normalized: return normalized return None def _resolve_provider_tenant_id( session, actor: dict, *, provider_type: str, provider_name: str, metadata: dict | None = None, provider_account_id: str | None = None, external_identifier: str | None = None, ) -> str: tenant_id = _optional_tenant_id(actor) if tenant_id: return tenant_id normalized_provider_name = str(provider_name or "").strip().lower() normalized_provider_type = str(provider_type or "").strip().lower() normalized_account = str(provider_account_id or "").strip() or _metadata_value( metadata, "provider_account_id", "payment_provider_account_id", "merchant_id", "account_id", "integration_id", ) normalized_external = str(external_identifier or "").strip() or _metadata_value( metadata, "external_identifier", "channel_id", "phone_number", "merchant_id", ) base_stmt = select(TenantIntegrationRow).where( func.lower(TenantIntegrationRow.provider_type) == normalized_provider_type, func.lower(TenantIntegrationRow.provider_name) == normalized_provider_name, TenantIntegrationRow.is_active == True, # noqa: E712 ) if normalized_account: row = session.execute( base_stmt.where(TenantIntegrationRow.provider_account_id == normalized_account).order_by(TenantIntegrationRow.id.desc()) ).scalars().first() if row is not None: return row.tenant_id if normalized_external: row = session.execute( base_stmt.where(TenantIntegrationRow.external_identifier == normalized_external).order_by(TenantIntegrationRow.id.desc()) ).scalars().first() if row is not None: return row.tenant_id raise HTTPException(status_code=400, detail="Tenant context is required") def _find_customer_by_customer_id(session, customer_id: str | None) -> Customer | None: normalized = str(customer_id or "").strip() if not normalized: return None return session.execute(select(Customer).where(Customer.customer_id == normalized)).scalar_one_or_none() def _create_lead_and_deal_for_contact( session, *, tenant_id: str, source_channel: str, preferred_channel: str, full_name: str, phone: str | None, customer_id: str | None = None, subject: str | None = None, ) -> tuple[SalesLeadRow, SalesDealRow]: now = _now().isoformat() pipeline = get_default_pipeline(session, tenant_id) stage = resolve_stage_by_code_or_id( session, tenant_id=tenant_id, pipeline_id=pipeline.pipeline_id, stage_id=DEFAULT_STAGE_CODE, ) lead = SalesLeadRow( lead_id=new_id("sld"), tenant_id=tenant_id, source_type="autosync", source_channel=source_channel, source_campaign_id=None, full_name=full_name, company_name=None, phone=phone, email=None, messenger_handles_json="{}", lead_temperature="warm", lead_score=60.0, customer_type=None, segment_type=None, initial_need_summary=subject, preferred_channel=preferred_channel, assigned_agent_type="voice_ai" if preferred_channel == "voice" else "text_ai", status="new_qualified_lead", crm_customer_id=customer_id, created_at=now, updated_at=now, ) deal = SalesDealRow( deal_id=new_id("sde"), tenant_id=tenant_id, lead_id=lead.lead_id, customer_id=customer_id, pipeline_id=pipeline.pipeline_id, stage_id=stage.stage_id, scenario_type="quick_sale", priority=3, title=subject or f"Сделка {full_name}", need_summary=subject, product_context_json="{}", estimated_amount=None, final_amount=None, currency=DEFAULT_SALES_CURRENCY, payment_model=None, document_required=True, payment_required=True, assigned_human_user_id=None, assigned_ai_orchestrator_id=None, preferred_channel=preferred_channel, current_channel=preferred_channel, status="active", won_reason=None, lost_reason=None, close_reason=None, next_action_type=None, next_action_at=None, last_contact_at=None, closed_at=None, created_at=now, updated_at=now, ) session.add(lead) session.add(deal) _record_stage_change( session, deal=deal, from_stage_id=None, to_stage_id=stage.stage_id, changed_by_type="system", changed_by_id=None, reason="autosync.created", ) _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="system", actor_id=None) return lead, deal def _resolve_or_create_sync_deal( session, *, tenant_id: str, source_channel: str, preferred_channel: str, customer_id: str | None = None, phone: str | None = None, display_name: str | None = None, interaction_id: str | None = None, thread_id: str | None = None, voice_session_id: str | None = None, external_call_id: str | None = None, subject: str | None = None, ) -> tuple[SalesLeadRow | None, SalesDealRow]: link = _find_external_link( session, tenant_id=tenant_id, thread_id=thread_id, voice_session_id=voice_session_id, external_call_id=external_call_id, interaction_id=interaction_id, ) if link is not None: deal = _get_deal(session, link.deal_id, tenant_id) lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None return lead, deal normalized_customer_id = str(customer_id or "").strip() or None if normalized_customer_id: deal = session.execute( select(SalesDealRow) .where(SalesDealRow.tenant_id == tenant_id) .where(SalesDealRow.customer_id == normalized_customer_id) .where(SalesDealRow.status == "active") .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc()) ).scalars().first() if deal is not None: lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None return lead, deal normalized_phone = str(phone or "").strip() or None if normalized_phone: lead, deal = _resolve_deal_for_inbound( session, tenant_id=tenant_id, deal_id=None, lead_id=None, phone=normalized_phone, subject=subject, source_channel=source_channel, preferred_channel=preferred_channel, ) if normalized_customer_id and deal.customer_id != normalized_customer_id: deal.customer_id = normalized_customer_id deal.updated_at = utc_now_iso() return lead, deal customer = _find_customer_by_customer_id(session, normalized_customer_id) full_name = str(display_name or (customer.display_name if customer else "") or "").strip() or ( "Telegram customer" if source_channel == "telegram" else "Voice customer" ) lead, deal = _create_lead_and_deal_for_contact( session, tenant_id=tenant_id, source_channel=source_channel, preferred_channel=preferred_channel, full_name=full_name, phone=normalized_phone, customer_id=normalized_customer_id, subject=subject, ) return lead, deal def _find_customer_by_phone(session, phone: str | None) -> Customer | None: normalized = str(phone or "").strip() if not normalized: return None direct = session.execute(select(Customer).where(Customer.preferred_phone == normalized)).scalar_one_or_none() if direct is not None: return direct rows = session.execute(select(Customer).order_by(Customer.id.desc())).scalars().all() for row in rows: phones = _json_list(row.phones_json) if normalized in phones: return row return None def _ensure_crm_customer(session, *, lead: SalesLeadRow | None, deal: SalesDealRow, counterparty: SalesCounterpartyRow | None) -> str | None: if deal.customer_id: return deal.customer_id phone = None display_name = None tags = ["sales"] if counterparty: phone = counterparty.phone_for_docs display_name = counterparty.full_name or counterparty.company_name if counterparty.company_name: tags.append("b2b") if not display_name and lead: display_name = lead.full_name phone = phone or lead.phone if not display_name: return None existing = _find_customer_by_phone(session, phone) if existing is not None: deal.customer_id = existing.customer_id if lead and not lead.crm_customer_id: lead.crm_customer_id = existing.customer_id lead.updated_at = utc_now_iso() return existing.customer_id now = utc_now_iso() customer = Customer( customer_id=new_id("cus"), display_name=display_name, phones_json=json.dumps([phone] if phone else [], ensure_ascii=False), preferred_phone=phone, tags_json=json.dumps(tags, ensure_ascii=False), created_at=now, ) session.add(customer) deal.customer_id = customer.customer_id if lead: lead.crm_customer_id = customer.customer_id lead.updated_at = now return customer.customer_id def _record_stage_change(session, *, deal: SalesDealRow, from_stage_id: str | None, to_stage_id: str, changed_by_type: str, changed_by_id: str | None, reason: str | None = None) -> None: session.add( SalesStageHistoryRow( history_id=new_id("sth"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, from_stage_id=from_stage_id, to_stage_id=to_stage_id, changed_by_type=changed_by_type, changed_by_id=changed_by_id, reason=reason, changed_at=utc_now_iso(), ) ) def _schedule_task(session, *, deal: SalesDealRow, task_type: str, run_at: str | None, payload: dict | None = None) -> SalesAutomationTaskRow: now = utc_now_iso() task = SalesAutomationTaskRow( task_id=new_id("tsk"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, task_type=task_type, payload_json=json.dumps(payload or {}, ensure_ascii=False), run_at=run_at or now, status="pending", retry_count=0, last_error=None, created_at=now, updated_at=now, ) session.add(task) return task def apply_stage_by_id(session, deal: SalesDealRow, stage_id: str, *, actor_type: str, actor_id: str | None, reason: str | None = None) -> SalesPipelineStageRow: stage = _get_stage(session, stage_id, deal.tenant_id, deal.pipeline_id) if not stage.is_active: raise HTTPException(status_code=400, detail="Stage is inactive") if deal.stage_id != stage.stage_id: previous = deal.stage_id previous_stage = _stage_by_id_or_none(session, deal.tenant_id, previous) deal.stage_id = stage.stage_id _record_stage_change( session, deal=deal, from_stage_id=previous, to_stage_id=stage.stage_id, changed_by_type=actor_type, changed_by_id=actor_id, reason=reason, ) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_STAGE_CHANGED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type=actor_type, actor_id=actor_id, payload={ "deal_id": deal.deal_id, "pipeline_id": deal.pipeline_id, "from_stage_id": previous, "from_stage_code": previous_stage.code if previous_stage else None, "to_stage_id": stage.stage_id, "to_stage_code": stage.code, "reason": reason, }, ) deal.updated_at = utc_now_iso() return stage def apply_stage_by_code(session, deal: SalesDealRow, stage_code: str, *, actor_type: str, actor_id: str | None, reason: str | None = None) -> SalesPipelineStageRow: ensure_default_pipeline(session, deal.tenant_id) stage = resolve_stage_by_code_or_id( session, tenant_id=deal.tenant_id, pipeline_id=deal.pipeline_id, stage_id=stage_code, ) return apply_stage_by_id(session, deal, stage.stage_id, actor_type=actor_type, actor_id=actor_id, reason=reason) def _apply_stage(session, deal: SalesDealRow, stage_code: str, *, actor_type: str, actor_id: str | None, reason: str | None = None) -> None: apply_stage_by_code(session, deal, stage_code, actor_type=actor_type, actor_id=actor_id, reason=reason) def _touch_deal_contact(deal: SalesDealRow) -> None: now = utc_now_iso() deal.last_contact_at = now 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: now = utc_now_iso() agent_type = payload.agent_type or ("voice_ai" if payload.channel_type == "voice" else "text_ai") communication = SalesCommunicationSessionRow( communication_id=new_id("com"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, lead_id=lead.lead_id if lead else deal.lead_id, customer_id=deal.customer_id, channel_type=payload.channel_type, direction=payload.direction, agent_type=agent_type, started_at=now, ended_at=None, duration_sec=None, subject=payload.subject, status="active", summary=payload.summary, transcript_id=None, next_action_type=payload.next_action_type, next_action_at=payload.next_action_at, sentiment=None, result_code=None, metadata_json=json.dumps(metadata or {}, ensure_ascii=False), created_at=now, updated_at=now, ) session.add(communication) deal.current_channel = "voice" if payload.channel_type == "voice" else (lead.preferred_channel if lead else deal.current_channel) _touch_deal_contact(deal) communication_actor_type = "system" if agent_type.endswith("_ai") else "human" _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.COMMUNICATION_STARTED, aggregate_type="communication", aggregate_id=communication.communication_id, actor_type=communication_actor_type, actor_id=actor_user, payload={ "deal_id": deal.deal_id, "lead_id": communication.lead_id, "communication_session_id": communication.communication_id, "channel_type": communication.channel_type, "direction": communication.direction, "agent_type": communication.agent_type, "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 payload.next_action_type: _schedule_task( session, deal=deal, task_type=payload.next_action_type, run_at=payload.next_action_at, payload={"source": "communication.start", "communication_id": communication.communication_id}, ) return communication def _resolve_deal_for_inbound( session, *, tenant_id: str, deal_id: str | None, lead_id: str | None, phone: str | None, subject: str | None, source_channel: str, preferred_channel: str, ) -> tuple[SalesLeadRow | None, SalesDealRow]: if deal_id: return None, _get_deal(session, deal_id, tenant_id) if lead_id: lead = _get_lead(session, lead_id, tenant_id) deal = session.execute( select(SalesDealRow) .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id) .order_by(SalesDealRow.id.desc()) ).scalars().first() if deal is None: raise HTTPException(status_code=404, detail="Lead has no deal") return lead, deal normalized_phone = str(phone or "").strip() if normalized_phone: lead = session.execute( select(SalesLeadRow) .where(SalesLeadRow.phone == normalized_phone, SalesLeadRow.tenant_id == tenant_id) .order_by(SalesLeadRow.updated_at.desc(), SalesLeadRow.id.desc()) ).scalars().first() if lead is not None: deal = session.execute( select(SalesDealRow) .where( SalesDealRow.lead_id == lead.lead_id, SalesDealRow.status == "active", SalesDealRow.tenant_id == tenant_id, ) .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc()) ).scalars().first() if deal is not None: return lead, deal crm_customer = _find_customer_by_phone(session, normalized_phone) now = utc_now_iso() pipeline = get_default_pipeline(session, tenant_id) stage = resolve_stage_by_code_or_id( session, tenant_id=tenant_id, pipeline_id=pipeline.pipeline_id, stage_id=DEFAULT_STAGE_CODE, ) lead = SalesLeadRow( lead_id=new_id("sld"), tenant_id=tenant_id, source_type="inbound", source_channel=source_channel, source_campaign_id=None, full_name=crm_customer.display_name if crm_customer else "Новый лид", company_name=None, phone=normalized_phone, email=None, messenger_handles_json="{}", lead_temperature="warm", lead_score=55, customer_type=None, segment_type=None, initial_need_summary=subject, preferred_channel=preferred_channel, assigned_agent_type="voice_ai" if preferred_channel == "voice" else "text_ai", status="new_qualified_lead", crm_customer_id=crm_customer.customer_id if crm_customer else None, created_at=now, updated_at=now, ) deal = SalesDealRow( deal_id=new_id("sde"), tenant_id=tenant_id, lead_id=lead.lead_id, customer_id=crm_customer.customer_id if crm_customer else None, pipeline_id=pipeline.pipeline_id, stage_id=stage.stage_id, scenario_type="quick_sale", priority=3, title=subject or f"Новый лид {normalized_phone}", need_summary=subject, product_context_json="{}", estimated_amount=None, final_amount=None, currency=DEFAULT_SALES_CURRENCY, payment_model=None, document_required=True, payment_required=True, assigned_human_user_id=None, assigned_ai_orchestrator_id=None, preferred_channel=preferred_channel, current_channel=preferred_channel, status="active", won_reason=None, lost_reason=None, close_reason=None, next_action_type=None, next_action_at=None, last_contact_at=None, closed_at=None, created_at=now, updated_at=now, ) session.add(lead) session.add(deal) _record_stage_change( session, deal=deal, from_stage_id=None, to_stage_id=stage.stage_id, changed_by_type="system", changed_by_id=None, reason="lead.entered_crm", ) _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="system", actor_id=None) return lead, deal raise HTTPException(status_code=400, detail="Cannot resolve inbound contact without deal_id, lead_id or phone") def _bridge_voice_runtime(call: SalesCallRow, communication: SalesCommunicationSessionRow, deal: SalesDealRow) -> None: if not _voice_runtime_enabled(): return metadata = _json_dict(communication.metadata_json) if metadata.get("voice_session_id"): return payload = { "call_id": call.external_call_id or call.call_id, "interaction_id": None, "customer_id": deal.customer_id, "queue_id": _default_handoff_queue_id(), "ai_session_id": None, "agent_profile": "voice_support", "language": "ru", "handoff_queue_id": _default_handoff_queue_id(), "greeting_text": f"Здравствуйте! Продолжим работу по сделке: {deal.title}.", } try: with httpx.Client(timeout=8.0) as client: response = client.post( f"{_voice_runtime_url()}/internal/voice-ai/sessions", json=payload, headers=_service_headers("svc:sales-service", "sales-service", "sales-service", deal.tenant_id), ) response.raise_for_status() data = response.json() metadata["voice_session_id"] = data.get("voice_session_id") metadata["voice_runtime"] = "attached" communication.metadata_json = json.dumps(metadata, ensure_ascii=False) communication.updated_at = utc_now_iso() except Exception as exc: # noqa: BLE001 metadata["voice_runtime"] = "degraded" metadata["voice_runtime_error"] = str(exc) communication.metadata_json = json.dumps(metadata, ensure_ascii=False) communication.updated_at = utc_now_iso() def _bridge_telegram_outbound(communication: SalesCommunicationSessionRow, message: SalesMessageRow, deal: SalesDealRow) -> None: if not _telegram_bridge_enabled(): return metadata = _json_dict(communication.metadata_json) thread_id = str(metadata.get("telegram_thread_id") or "").strip() if not thread_id or message.channel_provider != "telegram": return payload = {"text": message.body} try: with httpx.Client(timeout=8.0) as client: response = client.post( f"{_telegram_adapter_url()}/integrations/telegram/threads/{thread_id}/messages", json=payload, headers=_service_headers("svc:sales-service", "sales-service", "sales-service", deal.tenant_id), ) response.raise_for_status() message.delivery_status = "sent" except Exception as exc: # noqa: BLE001 message.delivery_status = "queued" metadata["telegram_bridge_error"] = str(exc) communication.metadata_json = json.dumps(metadata, ensure_ascii=False) communication.updated_at = utc_now_iso() def _counterparty_status(payload: SalesCounterpartyUpsertIn) -> str: required_hits = sum( 1 for value in [ payload.company_name or payload.full_name, payload.email_for_docs or payload.phone_for_docs, payload.bin_iin or payload.signer_name, ] if str(value or "").strip() ) return "completed" if required_hits >= 2 else "requested" def _invoice_number() -> str: return f"INV-{_now().strftime('%Y%m%d')}-{new_id('num').split('_', 1)[1].upper()}" def _merge_communication_metadata(communication: SalesCommunicationSessionRow, values: dict[str, Any]) -> dict: metadata = _json_dict(communication.metadata_json) changed = False for key, value in values.items(): if value is None: if key in metadata: metadata.pop(key, None) changed = True continue if isinstance(value, str): normalized = value.strip() if not normalized: if key in metadata: metadata.pop(key, None) changed = True continue value = normalized if metadata.get(key) != value: metadata[key] = value changed = True if changed: communication.metadata_json = json.dumps(metadata, ensure_ascii=False) communication.updated_at = utc_now_iso() return metadata def _upsert_external_link( session, *, deal: SalesDealRow, channel_provider: str, communication_id: str | None = None, sales_call_id: str | None = None, external_thread_id: str | None = None, external_chat_id: str | None = None, external_call_id: str | None = None, voice_session_id: str | None = None, ai_session_id: str | None = None, interaction_id: str | None = None, customer_id: str | None = None, phone_number: str | None = None, external_status: str | None = None, metadata: dict[str, Any] | None = None, ) -> SalesExternalLinkRow: row = _find_external_link( session, tenant_id=deal.tenant_id, thread_id=external_thread_id, voice_session_id=voice_session_id, external_call_id=external_call_id, interaction_id=interaction_id, ) now = utc_now_iso() if row is None: row = SalesExternalLinkRow( link_id=new_id("xlk"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication_id, sales_call_id=sales_call_id, channel_provider=channel_provider, external_thread_id=external_thread_id, external_chat_id=external_chat_id, external_call_id=external_call_id, voice_session_id=voice_session_id, ai_session_id=ai_session_id, interaction_id=interaction_id, customer_id=customer_id or deal.customer_id, phone_number=phone_number, external_status=external_status, link_metadata_json=json.dumps(metadata or {}, ensure_ascii=False), created_at=now, updated_at=now, last_sync_at=now, ) session.add(row) return row row.deal_id = deal.deal_id row.communication_id = communication_id or row.communication_id row.sales_call_id = sales_call_id or row.sales_call_id row.channel_provider = channel_provider or row.channel_provider row.external_thread_id = str(external_thread_id or "").strip() or row.external_thread_id row.external_chat_id = str(external_chat_id or "").strip() or row.external_chat_id row.external_call_id = str(external_call_id or "").strip() or row.external_call_id row.voice_session_id = str(voice_session_id or "").strip() or row.voice_session_id row.ai_session_id = str(ai_session_id or "").strip() or row.ai_session_id row.interaction_id = str(interaction_id or "").strip() or row.interaction_id row.customer_id = str(customer_id or "").strip() or row.customer_id or deal.customer_id row.phone_number = str(phone_number or "").strip() or row.phone_number row.external_status = str(external_status or "").strip() or row.external_status merged_meta = _json_dict(row.link_metadata_json) merged_meta.update(metadata or {}) row.link_metadata_json = json.dumps(merged_meta, ensure_ascii=False) row.updated_at = now row.last_sync_at = now return row def _get_or_create_text_communication( session, *, deal: SalesDealRow, lead: SalesLeadRow | None, direction: str, subject: str | None, metadata: dict | None = None, thread_id: str | None = None, ) -> SalesCommunicationSessionRow: 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: existing = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == link.communication_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id, ) ).scalar_one_or_none() if existing is not None: _merge_communication_metadata(existing, metadata or {}) return existing existing = session.execute( select(SalesCommunicationSessionRow) .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id) .where(SalesCommunicationSessionRow.channel_type == "text") .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc()) ).scalars().first() if existing is not None: _merge_communication_metadata(existing, metadata or {}) return existing return _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="text", direction=direction, # type: ignore[arg-type] agent_type="human" if direction == "outbound" else "text_ai", subject=subject, ), metadata=metadata, ) def _get_or_create_voice_communication( session, *, deal: SalesDealRow, lead: SalesLeadRow | None, subject: str | None, metadata: dict | None = None, voice_session_id: str | None = None, ) -> SalesCommunicationSessionRow: if str(voice_session_id or "").strip(): link = _find_external_link(session, tenant_id=deal.tenant_id, voice_session_id=voice_session_id) if link and link.communication_id: existing = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == link.communication_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id, ) ).scalar_one_or_none() if existing is not None: _merge_communication_metadata(existing, metadata or {}) return existing existing = session.execute( select(SalesCommunicationSessionRow) .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id) .where(SalesCommunicationSessionRow.channel_type == "voice") .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc()) ).scalars().first() if existing is not None: _merge_communication_metadata(existing, metadata or {}) return existing return _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="voice", direction="inbound", agent_type="voice_ai", subject=subject, ), metadata=metadata, ) def _build_workspace(session, deal: SalesDealRow) -> SalesWorkspaceOut: lead = _get_lead(session, deal.lead_id, deal.tenant_id) if deal.lead_id else None pipeline, stage = _deal_pipeline_stage(session, deal) communications = session.execute( select(SalesCommunicationSessionRow) .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id) .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc()) ).scalars().all() messages = session.execute( select(SalesMessageRow) .where(SalesMessageRow.deal_id == deal.deal_id, SalesMessageRow.tenant_id == deal.tenant_id) .order_by(SalesMessageRow.sent_at.desc(), SalesMessageRow.id.desc()) ).scalars().all() calls = session.execute( select(SalesCallRow) .where(SalesCallRow.deal_id == deal.deal_id, SalesCallRow.tenant_id == deal.tenant_id) .order_by(SalesCallRow.started_at.desc(), SalesCallRow.id.desc()) ).scalars().all() offers = session.execute( select(SalesOfferRow) .where(SalesOfferRow.deal_id == deal.deal_id, SalesOfferRow.tenant_id == deal.tenant_id) .order_by(SalesOfferRow.updated_at.desc(), SalesOfferRow.id.desc()) ).scalars().all() conditions = session.execute( select(SalesConditionRow) .where(SalesConditionRow.deal_id == deal.deal_id, SalesConditionRow.tenant_id == deal.tenant_id) .order_by(SalesConditionRow.updated_at.desc(), SalesConditionRow.id.desc()) ).scalars().all() counterparty = session.execute( select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == deal.tenant_id) ).scalar_one_or_none() documents = session.execute( select(SalesDocumentRow) .where(SalesDocumentRow.deal_id == deal.deal_id, SalesDocumentRow.tenant_id == deal.tenant_id) .order_by(SalesDocumentRow.updated_at.desc(), SalesDocumentRow.id.desc()) ).scalars().all() invoices = session.execute( select(SalesInvoiceRow) .where(SalesInvoiceRow.deal_id == deal.deal_id, SalesInvoiceRow.tenant_id == deal.tenant_id) .order_by(SalesInvoiceRow.updated_at.desc(), SalesInvoiceRow.id.desc()) ).scalars().all() payments = session.execute( select(SalesPaymentRow) .where(SalesPaymentRow.deal_id == deal.deal_id, SalesPaymentRow.tenant_id == deal.tenant_id) .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc()) ).scalars().all() escalations = session.execute( select(SalesEscalationRow) .where(SalesEscalationRow.deal_id == deal.deal_id, SalesEscalationRow.tenant_id == deal.tenant_id) .order_by(SalesEscalationRow.created_at.desc(), SalesEscalationRow.id.desc()) ).scalars().all() tasks = session.execute( select(SalesAutomationTaskRow) .where(SalesAutomationTaskRow.deal_id == deal.deal_id, SalesAutomationTaskRow.tenant_id == deal.tenant_id) .order_by(SalesAutomationTaskRow.run_at.desc(), SalesAutomationTaskRow.id.desc()) ).scalars().all() stage_history = session.execute( select(SalesStageHistoryRow) .where(SalesStageHistoryRow.deal_id == deal.deal_id, SalesStageHistoryRow.tenant_id == deal.tenant_id) .order_by(SalesStageHistoryRow.changed_at.desc(), SalesStageHistoryRow.id.desc()) ).scalars().all() stage_refs = _stage_refs_by_id( session, deal.tenant_id, [stage.stage_id if stage else None] + [row.from_stage_id for row in stage_history] + [row.to_stage_id for row in stage_history], ) channel_switches = session.execute( select(SalesChannelSwitchRow) .where(SalesChannelSwitchRow.deal_id == deal.deal_id, SalesChannelSwitchRow.tenant_id == deal.tenant_id) .order_by(SalesChannelSwitchRow.switched_at.desc(), SalesChannelSwitchRow.id.desc()) ).scalars().all() timeline: list[SalesTimelineEventOut] = [] for row in stage_history: to_stage = stage_refs.get(row.to_stage_id) timeline.append( SalesTimelineEventOut( ts=row.changed_at, kind="stage", title=f"Этап: {to_stage.name if to_stage else row.to_stage_id}", body=row.reason or "Изменение стадии сделки", meta={"from_stage_id": row.from_stage_id, "to_stage_id": row.to_stage_id}, ) ) for row in communications: timeline.append( SalesTimelineEventOut( ts=row.started_at, kind="communication", title=f"Коммуникация: {'звонок' if row.channel_type == 'voice' else 'текст'}", body=row.summary or row.subject or "Открыта новая сессия коммуникации", meta={"communication_id": row.communication_id, "agent_type": row.agent_type}, ) ) for row in offers: timeline.append( SalesTimelineEventOut( ts=row.updated_at, kind="offer", title=f"Оффер: {row.title}", body=f"Статус: {row.status}. Сумма: {row.total_amount:.2f} {row.currency}", meta={"offer_id": row.offer_id}, ) ) for row in documents: timeline.append( SalesTimelineEventOut( ts=row.updated_at, kind="document", title=f"Документ: {row.document_type}", body=f"Статус: {row.status}", meta={"document_id": row.document_id}, ) ) for row in invoices: timeline.append( SalesTimelineEventOut( ts=row.updated_at, kind="invoice", title=f"Счет: {row.invoice_number}", body=f"Статус: {row.status}. Сумма: {row.amount:.2f} {row.currency}", meta={"invoice_id": row.invoice_id}, ) ) for row in payments: timeline.append( SalesTimelineEventOut( ts=row.updated_at, kind="payment", title=f"Оплата: {row.status}", body=f"Сумма: {row.amount:.2f} {row.currency}", meta={"payment_id": row.payment_id}, ) ) for row in channel_switches: timeline.append( SalesTimelineEventOut( ts=row.switched_at, kind="channel_switch", title=f"Смена канала: {row.from_channel} -> {row.to_channel}", body=row.reason_for_channel_switch, meta={"switch_id": row.switch_id}, ) ) timeline.sort(key=lambda item: item.ts, reverse=True) return SalesWorkspaceOut( lead=_lead_to_out(lead) if lead else None, deal=_deal_to_out(deal, pipeline=pipeline, stage=stage), pipeline=_pipeline_to_out(pipeline) if pipeline else None, stage=_stage_to_ref(stage), communications=[_communication_to_out(row) for row in communications], messages=[_message_to_out(row) for row in messages], calls=[_call_to_out(row) for row in calls], offers=[_offer_to_out(row) for row in offers], conditions=[_condition_to_out(row) for row in conditions], counterparty=_counterparty_to_out(counterparty) if counterparty else None, documents=[_document_to_out(row) for row in documents], invoices=[_invoice_to_out(row) for row in invoices], payments=[_payment_to_out(row) for row in payments], escalations=[_escalation_to_out(row) for row in escalations], tasks=[_task_to_out(row) for row in tasks], stage_history=[_stage_history_to_out(row, stage_refs) for row in stage_history], channel_switches=[_channel_switch_to_out(row) for row in channel_switches], timeline=timeline[:40], ) @app.get("/health", response_model=HealthResponse) def health() -> HealthResponse: return HealthResponse(status="ok", service="sales-service") @app.get("/api/v1/dashboard", response_model=SalesDashboardSummaryOut) def sales_dashboard(actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> SalesDashboardSummaryOut: session = get_session() try: tenant_id = _tenant_id(actor) leads_total = session.execute( select(func.count()).select_from(SalesLeadRow).where(SalesLeadRow.tenant_id == tenant_id) ).scalar_one() deals_active = session.execute( select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "active") ).scalar_one() deals_won = session.execute( select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "won") ).scalar_one() deals_lost = session.execute( select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "lost") ).scalar_one() overdue_invoices = session.execute( select(func.count()).select_from(SalesInvoiceRow).where(SalesInvoiceRow.tenant_id == tenant_id, SalesInvoiceRow.status == "overdue") ).scalar_one() payment_expected = session.execute( select(func.coalesce(func.sum(SalesInvoiceRow.amount), 0.0)).where( SalesInvoiceRow.tenant_id == tenant_id, SalesInvoiceRow.status.in_(["issued", "sent", "partially_paid", "overdue"]) ) ).scalar_one() payment_received = session.execute( select(func.coalesce(func.sum(SalesPaymentRow.amount), 0.0)).where( SalesPaymentRow.tenant_id == tenant_id, SalesPaymentRow.status.in_(["success", "partial"]), ) ).scalar_one() voice_sessions = session.execute( select(func.count()).select_from(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.tenant_id == tenant_id, SalesCommunicationSessionRow.channel_type == "voice", ) ).scalar_one() text_sessions = session.execute( select(func.count()).select_from(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.tenant_id == tenant_id, SalesCommunicationSessionRow.channel_type == "text", ) ).scalar_one() channel_switches = session.execute( select(func.count()).select_from(SalesChannelSwitchRow).where(SalesChannelSwitchRow.tenant_id == tenant_id) ).scalar_one() human_escalations = session.execute( select(func.count()).select_from(SalesEscalationRow).where(SalesEscalationRow.tenant_id == tenant_id) ).scalar_one() stage_rows = session.execute( select(SalesDealRow.stage_id, func.count(), func.coalesce(func.sum(SalesDealRow.final_amount), func.sum(SalesDealRow.estimated_amount), 0.0)) .where(SalesDealRow.tenant_id == tenant_id) .group_by(SalesDealRow.stage_id) ).all() stage_refs = _stage_refs_by_id(session, tenant_id, [stage_id for stage_id, _count, _amount in stage_rows]) stage_counts = [ SalesDashboardStageOut( stage_id=stage_id, label=stage_refs[stage_id].name if stage_id in stage_refs else stage_id, stage=_stage_to_ref(stage_refs.get(stage_id)), count=int(count or 0), amount=float(amount or 0.0), ) for stage_id, count, amount in stage_rows ] hottest_rows = session.execute( select(SalesDealRow) .where(SalesDealRow.tenant_id == tenant_id) .order_by(SalesDealRow.priority.asc(), SalesDealRow.updated_at.desc(), SalesDealRow.id.desc()) .limit(5) ).scalars().all() return SalesDashboardSummaryOut( leads_total=int(leads_total or 0), deals_active=int(deals_active or 0), deals_won=int(deals_won or 0), deals_lost=int(deals_lost or 0), overdue_invoices=int(overdue_invoices or 0), payment_expected=float(payment_expected or 0.0), payment_received=float(payment_received or 0.0), voice_sessions=int(voice_sessions or 0), text_sessions=int(text_sessions or 0), channel_switches=int(channel_switches or 0), human_escalations=int(human_escalations or 0), stage_counts=stage_counts, hottest_deals=[_deal_to_out_with_refs(session, row) for row in hottest_rows], ) finally: session.close() @app.get("/api/v1/pipelines", response_model=list[SalesPipelineOut]) def list_pipelines(actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> list[SalesPipelineOut]: session = get_session() try: tenant_id = _tenant_id(actor) ensure_default_pipeline(session, tenant_id) rows = session.execute( select(SalesPipelineRow) .where(SalesPipelineRow.tenant_id == tenant_id) .order_by(SalesPipelineRow.is_default.desc(), SalesPipelineRow.created_at.asc(), SalesPipelineRow.id.asc()) ).scalars().all() session.commit() return [_pipeline_to_out(row) for row in rows] finally: session.close() @app.get("/api/v1/pipelines/{pipeline_id}", response_model=SalesPipelineOut) def get_pipeline(pipeline_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> SalesPipelineOut: session = get_session() try: tenant_id = _tenant_id(actor) ensure_default_pipeline(session, tenant_id) return _pipeline_to_out(_get_pipeline(session, pipeline_id, tenant_id)) finally: session.close() @app.post("/api/v1/pipelines", response_model=SalesPipelineOut) def create_pipeline( payload: SalesPipelineCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPipelineOut: session = get_session() try: tenant_id = _tenant_id(actor) ensure_default_pipeline(session, tenant_id) now = utc_now_iso() pipeline = SalesPipelineRow( pipeline_id=new_id("pip"), tenant_id=tenant_id, code=payload.code.strip(), name=payload.name, description=payload.description, is_default=False, is_active=payload.is_active, created_at=now, updated_at=now, ) session.add(pipeline) if payload.is_default: _set_default_pipeline(session, pipeline) try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException(status_code=409, detail="Pipeline code already exists") from exc return _pipeline_to_out(pipeline) finally: session.close() @app.patch("/api/v1/pipelines/{pipeline_id}", response_model=SalesPipelineOut) def update_pipeline( pipeline_id: str, payload: SalesPipelineUpdate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPipelineOut: session = get_session() try: tenant_id = _tenant_id(actor) pipeline = _get_pipeline(session, pipeline_id, tenant_id) updates = payload.model_dump(exclude_unset=True) if "name" in updates: pipeline.name = updates["name"] if "description" in updates: pipeline.description = updates["description"] if updates.get("is_active") is False and pipeline.is_default: raise HTTPException(status_code=400, detail="Default pipeline cannot be deactivated") if "is_active" in updates: pipeline.is_active = bool(updates["is_active"]) if updates.get("is_default") is True: if not pipeline.is_active: raise HTTPException(status_code=400, detail="Inactive pipeline cannot be default") _set_default_pipeline(session, pipeline) elif updates.get("is_default") is False and pipeline.is_default: raise HTTPException(status_code=400, detail="Use another pipeline set-default action") pipeline.updated_at = utc_now_iso() session.commit() return _pipeline_to_out(pipeline) finally: session.close() @app.post("/api/v1/pipelines/{pipeline_id}/set-default", response_model=SalesPipelineOut) def set_default_pipeline( pipeline_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPipelineOut: session = get_session() try: tenant_id = _tenant_id(actor) pipeline = _get_pipeline(session, pipeline_id, tenant_id) if not pipeline.is_active: raise HTTPException(status_code=400, detail="Inactive pipeline cannot be default") _set_default_pipeline(session, pipeline) session.commit() return _pipeline_to_out(pipeline) finally: session.close() @app.get("/api/v1/pipelines/{pipeline_id}/stages", response_model=list[SalesPipelineStageOut]) def list_pipeline_stages( pipeline_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesPipelineStageOut]: session = get_session() try: tenant_id = _tenant_id(actor) pipeline = _get_pipeline(session, pipeline_id, tenant_id) if pipeline.code == DEFAULT_PIPELINE_CODE: ensure_default_stages(session, tenant_id, pipeline.pipeline_id) rows = session.execute( select(SalesPipelineStageRow) .where(SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.pipeline_id == pipeline_id) .order_by(SalesPipelineStageRow.sort_order.asc(), SalesPipelineStageRow.id.asc()) ).scalars().all() session.commit() return [_stage_to_out(row) for row in rows] finally: session.close() @app.post("/api/v1/pipelines/{pipeline_id}/stages", response_model=SalesPipelineStageOut) def create_pipeline_stage( pipeline_id: str, payload: SalesPipelineStageCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPipelineStageOut: session = get_session() try: tenant_id = _tenant_id(actor) _get_pipeline(session, pipeline_id, tenant_id) now = utc_now_iso() stage = SalesPipelineStageRow( stage_id=new_id("pst"), tenant_id=tenant_id, pipeline_id=pipeline_id, code=payload.code.strip(), name=payload.name, category=payload.category, sort_order=payload.sort_order, is_terminal=payload.is_terminal, is_system=payload.is_system, is_active=payload.is_active, created_at=now, updated_at=now, ) session.add(stage) try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException(status_code=409, detail="Stage code already exists") from exc return _stage_to_out(stage) finally: session.close() @app.patch("/api/v1/pipeline-stages/{stage_id}", response_model=SalesPipelineStageOut) def update_pipeline_stage( stage_id: str, payload: SalesPipelineStageUpdate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPipelineStageOut: session = get_session() try: tenant_id = _tenant_id(actor) stage = _get_stage(session, stage_id, tenant_id) updates = payload.model_dump(exclude_unset=True) if updates.get("is_active") is False and stage.is_active: active_deals = session.execute( select(func.count()) .select_from(SalesDealRow) .where( SalesDealRow.tenant_id == tenant_id, SalesDealRow.stage_id == stage.stage_id, SalesDealRow.status == "active", ) ).scalar_one() if int(active_deals or 0) > 0: raise HTTPException(status_code=400, detail="Stage has active deals") if "name" in updates: stage.name = updates["name"] if "category" in updates: stage.category = updates["category"] if "sort_order" in updates: stage.sort_order = int(updates["sort_order"]) if "is_active" in updates: stage.is_active = bool(updates["is_active"]) stage.updated_at = utc_now_iso() session.commit() return _stage_to_out(stage) finally: session.close() @app.post("/api/v1/leads", response_model=SalesLeadOut) def create_lead( payload: SalesLeadCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesLeadOut: session = get_session() try: tenant_id = _tenant_id(actor) now = utc_now_iso() crm_customer = _find_customer_by_phone(session, payload.phone) pipeline = get_default_pipeline(session, tenant_id) initial_stage_code = payload.status if payload.status != "converted" else DEFAULT_STAGE_CODE try: stage = resolve_stage_by_code_or_id( session, tenant_id=tenant_id, pipeline_id=pipeline.pipeline_id, stage_id=initial_stage_code, ) except HTTPException: stage = resolve_stage_by_code_or_id( session, tenant_id=tenant_id, pipeline_id=pipeline.pipeline_id, stage_id=DEFAULT_STAGE_CODE, ) lead = SalesLeadRow( lead_id=new_id("sld"), tenant_id=tenant_id, source_type=payload.source_type, source_channel=payload.source_channel, source_campaign_id=payload.source_campaign_id, full_name=payload.full_name, company_name=payload.company_name, phone=payload.phone, email=payload.email, messenger_handles_json=json.dumps(payload.messenger_handles, ensure_ascii=False), lead_temperature=payload.lead_temperature, lead_score=payload.lead_score, customer_type=payload.customer_type, segment_type=payload.segment_type, initial_need_summary=payload.initial_need_summary, preferred_channel=payload.preferred_channel, assigned_agent_type=payload.assigned_agent_type, status=payload.status, crm_customer_id=crm_customer.customer_id if crm_customer else None, created_at=now, updated_at=now, ) deal = SalesDealRow( deal_id=new_id("sde"), tenant_id=tenant_id, lead_id=lead.lead_id, customer_id=crm_customer.customer_id if crm_customer else None, pipeline_id=pipeline.pipeline_id, stage_id=stage.stage_id, scenario_type="quick_sale", priority=payload.priority, title=payload.title or f"{payload.full_name} • {payload.source_channel}", need_summary=payload.initial_need_summary, product_context_json="{}", estimated_amount=payload.estimated_amount, final_amount=None, currency=payload.currency or DEFAULT_SALES_CURRENCY, payment_model=None, document_required=True, payment_required=True, assigned_human_user_id=None, assigned_ai_orchestrator_id=None, preferred_channel=payload.preferred_channel, current_channel=payload.preferred_channel, status="active", won_reason=None, lost_reason=None, close_reason=None, next_action_type=None, next_action_at=None, last_contact_at=None, closed_at=None, created_at=now, updated_at=now, ) session.add(lead) session.add(deal) _record_stage_change( session, deal=deal, from_stage_id=None, to_stage_id=stage.stage_id, changed_by_type="human", changed_by_id=actor["user"], reason="lead.entered_crm", ) _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="human", actor_id=actor["user"]) session.commit() return _lead_to_out(lead) finally: session.close() @app.get("/api/v1/leads", response_model=list[SalesLeadOut]) def list_leads( query: str | None = None, status: str | None = None, limit: int = Query(default=100, ge=1, le=200), actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesLeadOut]: session = get_session() try: tenant_id = _tenant_id(actor) stmt = ( select(SalesLeadRow) .where(SalesLeadRow.tenant_id == tenant_id) .order_by(SalesLeadRow.updated_at.desc(), SalesLeadRow.id.desc()) ) if status: stmt = stmt.where(SalesLeadRow.status == status) if query: pattern = f"%{query.strip().lower()}%" stmt = stmt.where( or_( func.lower(func.coalesce(SalesLeadRow.full_name, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.company_name, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.phone, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.email, "")).like(pattern), ) ) rows = session.execute(stmt.limit(limit)).scalars().all() return [_lead_to_out(row) for row in rows] finally: session.close() @app.get("/api/v1/leads/{lead_id}", response_model=SalesLeadOut) def get_lead( lead_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesLeadOut: session = get_session() try: return _lead_to_out(_get_lead(session, lead_id, _tenant_id(actor))) finally: session.close() @app.patch("/api/v1/leads/{lead_id}", response_model=SalesLeadOut) def update_lead( lead_id: str, payload: SalesLeadUpdate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesLeadOut: session = get_session() try: tenant_id = _tenant_id(actor) lead = _get_lead(session, lead_id, tenant_id) updates = payload.model_dump(exclude_unset=True) if "messenger_handles" in updates: lead.messenger_handles_json = json.dumps(updates.pop("messenger_handles") or {}, ensure_ascii=False) for key, value in updates.items(): setattr(lead, key, value) lead.updated_at = utc_now_iso() session.commit() return _lead_to_out(lead) finally: session.close() @app.post("/api/v1/leads/{lead_id}/enrich", response_model=SalesWorkspaceOut) def enrich_lead( lead_id: str, payload: SalesLeadEnrichIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesWorkspaceOut: session = get_session() try: tenant_id = _tenant_id(actor) lead = _get_lead(session, lead_id, tenant_id) if payload.need_summary: lead.initial_need_summary = payload.need_summary if payload.customer_type: lead.customer_type = payload.customer_type if payload.preferred_channel: lead.preferred_channel = payload.preferred_channel lead.updated_at = utc_now_iso() deal = session.execute( select(SalesDealRow) .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id) .order_by(SalesDealRow.id.desc()) ).scalars().first() if deal is None: raise HTTPException(status_code=404, detail="Deal not found for lead") if payload.need_summary: deal.need_summary = payload.need_summary if payload.product_context: deal.product_context_json = json.dumps(payload.product_context, ensure_ascii=False) if payload.preferred_channel: deal.preferred_channel = payload.preferred_channel deal.current_channel = payload.preferred_channel deal.updated_at = utc_now_iso() next_stage = _infer_stage_for_channel(_infer_text_channel(lead.preferred_channel)) _apply_stage( session, deal, next_stage, actor_type="human", actor_id=actor["user"], reason="lead.enrichment_completed", ) _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.LEAD_ENRICHMENT_COMPLETED, aggregate_type="lead", aggregate_id=lead.lead_id, actor_type="human", actor_id=actor["user"], payload={ "lead_id": lead.lead_id, "deal_id": deal.deal_id, "need_summary": lead.initial_need_summary, "customer_type": lead.customer_type, "preferred_channel": lead.preferred_channel, }, ) session.commit() return _build_workspace(session, deal) finally: session.close() @app.post("/api/v1/leads/{lead_id}/convert-to-deal", response_model=SalesDealOut) def convert_lead_to_deal( lead_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: tenant_id = _tenant_id(actor) lead = _get_lead(session, lead_id, tenant_id) lead.status = "converted" lead.updated_at = utc_now_iso() deal = session.execute( select(SalesDealRow) .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id) .order_by(SalesDealRow.id.desc()) ).scalars().first() if deal is None: raise HTTPException(status_code=404, detail="Deal not found for lead") session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals", response_model=SalesDealOut) def create_deal( payload: SalesDealCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: tenant_id = _tenant_id(actor) now = utc_now_iso() if payload.lead_id: _get_lead(session, payload.lead_id, tenant_id) pipeline = get_default_pipeline(session, tenant_id) stage_id_input = payload.stage_id if not stage_id_input and not payload.stage_code: stage_id_input = DEFAULT_STAGE_CODE stage = resolve_stage_by_code_or_id( session, tenant_id=tenant_id, pipeline_id=pipeline.pipeline_id, stage_id=stage_id_input, stage_code=payload.stage_code, ) deal = SalesDealRow( deal_id=new_id("sde"), tenant_id=tenant_id, lead_id=payload.lead_id, customer_id=payload.customer_id, pipeline_id=pipeline.pipeline_id, stage_id=stage.stage_id, scenario_type=payload.scenario_type, priority=payload.priority, title=payload.title, need_summary=payload.need_summary, product_context_json=json.dumps(payload.product_context, ensure_ascii=False), estimated_amount=payload.estimated_amount, final_amount=payload.final_amount, currency=payload.currency or DEFAULT_SALES_CURRENCY, payment_model=payload.payment_model, document_required=payload.document_required, payment_required=payload.payment_required, assigned_human_user_id=None, assigned_ai_orchestrator_id=None, preferred_channel=payload.preferred_channel, current_channel=payload.current_channel, status="active", won_reason=None, lost_reason=None, close_reason=None, next_action_type=None, next_action_at=None, last_contact_at=None, closed_at=None, created_at=now, updated_at=now, ) session.add(deal) _record_stage_change( session, deal=deal, from_stage_id=None, to_stage_id=stage.stage_id, changed_by_type="human", changed_by_id=actor["user"], reason="deal.created", ) session.commit() return _deal_to_out(deal, pipeline=pipeline, stage=stage) finally: session.close() @app.get("/api/v1/deals", response_model=list[SalesDealOut]) def list_deals( query: str | None = None, stage_id: str | None = None, status: str | None = None, limit: int = Query(default=100, ge=1, le=300), actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesDealOut]: session = get_session() try: tenant_id = _tenant_id(actor) stmt = ( select(SalesDealRow) .where(SalesDealRow.tenant_id == tenant_id) .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc()) ) if stage_id: matching_stage_ids = _stage_filter_ids(session, tenant_id, stage_id) stmt = stmt.where(SalesDealRow.stage_id.in_(matching_stage_ids or [""])) if status: stmt = stmt.where(SalesDealRow.status == status) if query: normalized_query = query.strip().lower() pattern = f"%{normalized_query}%" matching_lead_ids = session.execute( select(SalesLeadRow.lead_id).where( SalesLeadRow.tenant_id == tenant_id, or_( func.lower(func.coalesce(SalesLeadRow.full_name, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.company_name, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.phone, "")).like(pattern), func.lower(func.coalesce(SalesLeadRow.email, "")).like(pattern), ) ) ).scalars().all() matching_deal_ids = session.execute( select(SalesExternalLinkRow.deal_id).where( SalesExternalLinkRow.tenant_id == tenant_id, or_( func.lower(func.coalesce(SalesExternalLinkRow.external_thread_id, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.external_chat_id, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.voice_session_id, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.external_call_id, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.interaction_id, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.phone_number, "")).like(pattern), func.lower(func.coalesce(SalesExternalLinkRow.customer_id, "")).like(pattern), ) ) ).scalars().all() stmt = stmt.where( or_( func.lower(func.coalesce(SalesDealRow.title, "")).like(pattern), func.lower(func.coalesce(SalesDealRow.need_summary, "")).like(pattern), func.lower(func.coalesce(SalesDealRow.customer_id, "")).like(pattern), SalesDealRow.lead_id.in_(matching_lead_ids or [""]), SalesDealRow.deal_id.in_(matching_deal_ids or [""]), ) ) rows = session.execute(stmt.limit(limit)).scalars().all() return [_deal_to_out_with_refs(session, row) for row in rows] finally: session.close() @app.get("/api/v1/deals/{deal_id}", response_model=SalesDealOut) def get_deal( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesDealOut: session = get_session() try: return _deal_to_out_with_refs(session, _get_deal(session, deal_id, _tenant_id(actor))) finally: session.close() @app.get("/api/v1/deals/{deal_id}/workspace", response_model=SalesWorkspaceOut) def get_workspace( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesWorkspaceOut: session = get_session() try: return _build_workspace(session, _get_deal(session, deal_id, _tenant_id(actor))) finally: session.close() @app.patch("/api/v1/deals/{deal_id}", response_model=SalesDealOut) def update_deal( deal_id: str, payload: SalesDealUpdate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) updates = payload.model_dump(exclude_unset=True) if "product_context" in updates: deal.product_context_json = json.dumps(updates.pop("product_context") or {}, ensure_ascii=False) for key, value in updates.items(): setattr(deal, key, value) deal.updated_at = utc_now_iso() session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals/{deal_id}/change-stage", response_model=SalesDealOut) def change_stage( deal_id: str, payload: SalesDealStageChangeIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) if not payload.stage_id and not payload.stage_code: raise HTTPException(status_code=400, detail="stage_id or stage_code is required") ensure_default_pipeline(session, deal.tenant_id) stage = resolve_stage_by_code_or_id( session, tenant_id=deal.tenant_id, pipeline_id=deal.pipeline_id, stage_id=payload.stage_id, stage_code=payload.stage_code, ) apply_stage_by_id(session, deal, stage.stage_id, actor_type="human", actor_id=actor["user"], reason=payload.reason) session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals/{deal_id}/select-scenario", response_model=SalesDealOut) def select_scenario( deal_id: str, payload: SalesDealScenarioIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) previous_scenario_type = deal.scenario_type deal.scenario_type = payload.scenario_type deal.updated_at = utc_now_iso() _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_SCENARIO_SELECTED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "scenario_type": deal.scenario_type, "previous_scenario_type": previous_scenario_type, }, ) session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals/{deal_id}/schedule-next-action", response_model=SalesDealOut) def schedule_next_action( deal_id: str, payload: SalesDealNextActionIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) deal.next_action_type = payload.next_action_type deal.next_action_at = payload.next_action_at deal.updated_at = utc_now_iso() _schedule_task( session, deal=deal, task_type=payload.next_action_type, run_at=payload.next_action_at, payload={**payload.payload, "scheduled_by": actor["user"]}, ) _apply_stage( session, deal, "follow_up_scheduled", actor_type="human", actor_id=actor["user"], reason="deal.next_action_scheduled", ) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_NEXT_ACTION_SCHEDULED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "next_action_type": deal.next_action_type, "next_action_at": deal.next_action_at, }, ) session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals/{deal_id}/close", response_model=SalesDealOut) def close_deal( deal_id: str, payload: SalesDealCloseIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDealOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) deal.status = payload.status deal.closed_at = utc_now_iso() deal.close_reason = payload.reason if payload.status == "won": deal.won_reason = payload.reason elif payload.status == "lost": deal.lost_reason = payload.reason _apply_stage( session, deal, payload.status, actor_type="human", actor_id=actor["user"], reason=payload.reason or "deal.closed", ) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_CLOSED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "status": deal.status, "reason": payload.reason, "closed_at": deal.closed_at, }, ) if payload.status == "won": _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_WON, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "final_amount": deal.final_amount, "currency": deal.currency, "won_reason": payload.reason or "deal.closed", }, ) elif payload.status == "lost": _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DEAL_LOST, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "lost_reason": payload.reason or "deal.closed", "currency": deal.currency, }, ) session.commit() return _deal_to_out_with_refs(session, deal) finally: session.close() @app.post("/api/v1/deals/{deal_id}/escalate", response_model=SalesEscalationOut) def escalate_deal( deal_id: str, payload: SalesEscalationCreateIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesEscalationOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) now = utc_now_iso() row = SalesEscalationRow( escalation_id=new_id("esc"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, escalation_type=payload.escalation_type, reason=payload.reason, severity=payload.severity, status="open", assigned_to_user_id=payload.assigned_to_user_id, created_at=now, resolved_at=None, ) session.add(row) deal.assigned_human_user_id = payload.assigned_to_user_id or actor["user"] deal.scenario_type = "custom_human_escalation" _apply_stage( session, deal, "transferred_to_support", actor_type="human", actor_id=actor["user"], reason=payload.reason, ) session.commit() return _escalation_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/communications/text", response_model=SalesCommunicationOut) def start_text_communication( deal_id: str, payload: SalesCommunicationStartIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCommunicationOut: payload = payload.model_copy(update={"channel_type": "text"}) session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, deal_id, tenant_id) lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None row = _start_communication(session, deal=deal, lead=lead, payload=payload, actor_user=actor["user"]) session.commit() return _communication_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/communications/voice", response_model=SalesCommunicationOut) def start_voice_communication( deal_id: str, payload: SalesCommunicationStartIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCommunicationOut: payload = payload.model_copy(update={"channel_type": "voice"}) session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, deal_id, tenant_id) lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None row = _start_communication(session, deal=deal, lead=lead, payload=payload, actor_user=actor["user"]) session.commit() return _communication_to_out(row) finally: session.close() @app.get("/api/v1/deals/{deal_id}/communications", response_model=list[SalesCommunicationOut]) def list_communications( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesCommunicationOut]: session = get_session() try: tenant_id = _tenant_id(actor) _get_deal(session, deal_id, tenant_id) rows = session.execute( select(SalesCommunicationSessionRow) .where(SalesCommunicationSessionRow.deal_id == deal_id, SalesCommunicationSessionRow.tenant_id == tenant_id) .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc()) ).scalars().all() return [_communication_to_out(row) for row in rows] finally: session.close() @app.post("/api/v1/communications/{communication_id}/summary", response_model=SalesCommunicationOut) def summarize_communication( communication_id: str, payload: SalesCommunicationSummaryIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCommunicationOut: session = get_session() try: tenant_id = _tenant_id(actor) row = _get_communication(session, communication_id, tenant_id) row.summary = payload.summary row.result_code = payload.result_code row.sentiment = payload.sentiment row.next_action_type = payload.next_action_type row.next_action_at = payload.next_action_at row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _touch_deal_contact(deal) if payload.next_action_type: deal.next_action_type = payload.next_action_type deal.next_action_at = payload.next_action_at _schedule_task( session, deal=deal, task_type=payload.next_action_type, run_at=payload.next_action_at, payload={"source": "communication.summary", "communication_id": row.communication_id}, ) _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DEAL_NEXT_ACTION_SCHEDULED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "next_action_type": payload.next_action_type, "next_action_at": payload.next_action_at, "communication_session_id": row.communication_id, }, ) _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.COMMUNICATION_SUMMARY_CREATED, aggregate_type="communication", aggregate_id=row.communication_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "communication_session_id": row.communication_id, "summary": row.summary, "result_code": row.result_code, "sentiment": row.sentiment, }, ) session.commit() return _communication_to_out(row) finally: session.close() @app.post("/api/v1/communications/{communication_id}/switch-channel", response_model=SalesChannelSwitchOut) def switch_channel( communication_id: str, payload: SalesCommunicationSwitchChannelIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesChannelSwitchOut: session = get_session() try: tenant_id = _tenant_id(actor) communication = _get_communication(session, communication_id, tenant_id) deal = _get_deal(session, communication.deal_id, tenant_id) from_channel = "voice" if communication.channel_type == "voice" else deal.current_channel now = utc_now_iso() row = SalesChannelSwitchRow( switch_id=new_id("swc"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, from_channel=from_channel, to_channel=payload.to_channel, reason_for_channel_switch=payload.reason_for_channel_switch, switched_at=now, ) session.add(row) deal.current_channel = payload.to_channel deal.updated_at = now _apply_stage( session, deal, _infer_stage_for_channel(_infer_text_channel(payload.to_channel)), actor_type="human", actor_id=actor["user"], reason=payload.reason_for_channel_switch, ) _schedule_task( session, deal=deal, task_type=f"switch_to_{payload.to_channel}", run_at=now, payload={"reason": payload.reason_for_channel_switch, "communication_id": communication.communication_id}, ) session.commit() return _channel_switch_to_out(row) finally: session.close() @app.post("/api/v1/communications/{communication_id}/bind-external", response_model=SalesCommunicationOut) def bind_external_communication( communication_id: str, payload: SalesCommunicationBindExternalIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCommunicationOut: session = get_session() try: tenant_id = _tenant_id(actor) communication = _get_communication(session, communication_id, tenant_id) deal = _get_deal(session, communication.deal_id, tenant_id) metadata_updates = dict(payload.metadata or {}) if payload.channel_provider: metadata_updates["channel_provider"] = payload.channel_provider if payload.telegram_thread_id: metadata_updates["telegram_thread_id"] = payload.telegram_thread_id if payload.telegram_chat_id: metadata_updates["telegram_chat_id"] = payload.telegram_chat_id if payload.voice_session_id: metadata_updates["voice_session_id"] = payload.voice_session_id if payload.external_call_id: metadata_updates["external_call_id"] = payload.external_call_id _merge_communication_metadata(communication, metadata_updates) text_channels = {"telegram", "whatsapp", "webchat", "email"} channel_provider = str(payload.channel_provider or "").strip().lower() if payload.telegram_thread_id or channel_provider in text_channels: next_channel = channel_provider if channel_provider in text_channels else "telegram" deal.current_channel = next_channel deal.preferred_channel = next_channel if communication.channel_type != "voice": _apply_stage( session, deal, "active_text_communication", actor_type="human", actor_id=actor["user"], reason="external.channel_bound", ) if payload.voice_session_id: deal.current_channel = "voice" deal.preferred_channel = "voice" _apply_stage( session, deal, "active_voice_communication", actor_type="human", actor_id=actor["user"], reason="voice.session_bound", ) if payload.external_call_id: call = session.execute( select(SalesCallRow) .where(SalesCallRow.communication_id == communication.communication_id, SalesCallRow.tenant_id == tenant_id) .order_by(SalesCallRow.id.desc()) .limit(1) ).scalar_one_or_none() if call is not None: call.external_call_id = payload.external_call_id.strip() call.updated_at = utc_now_iso() session.commit() return _communication_to_out(communication) finally: session.close() @app.post("/internal/sales-sync/telegram", response_model=SalesWorkspaceOut) def sync_telegram_thread( payload: SalesTelegramSyncIn, actor: dict = Depends(require_roles(Role.ADMIN)), ) -> SalesWorkspaceOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="message", provider_name="telegram", metadata=payload.metadata, external_identifier=payload.thread_id or payload.chat_id, ) lead, deal = _resolve_or_create_sync_deal( session, tenant_id=tenant_id, source_channel="telegram", preferred_channel="telegram", customer_id=payload.customer_id, phone=payload.phone_number, display_name=payload.display_name, interaction_id=payload.interaction_id, thread_id=payload.thread_id, subject=payload.text or payload.display_name or "Telegram inquiry", ) communication = _get_or_create_text_communication( session, deal=deal, lead=lead, direction=payload.direction, subject=(payload.text or payload.display_name or "Telegram conversation")[:120], metadata={ "telegram_thread_id": payload.thread_id, "telegram_chat_id": payload.chat_id, "interaction_id": payload.interaction_id, "queue_id": payload.queue_id, "ai_state": payload.ai_state, "ai_handoff_reason": payload.ai_handoff_reason, **(payload.metadata or {}), }, thread_id=payload.thread_id, ) event_ts = str(payload.happened_at or utc_now_iso()).strip() or utc_now_iso() dedupe_external_message_id = str(payload.external_message_id or payload.message_id or "").strip() or None if dedupe_external_message_id: message = session.execute( select(SalesMessageRow) .where(SalesMessageRow.tenant_id == tenant_id) .where(SalesMessageRow.channel_provider == "telegram") .where(SalesMessageRow.external_message_id == dedupe_external_message_id) ).scalar_one_or_none() else: message = None if message is None and str(payload.text or "").strip(): message = SalesMessageRow( message_id=new_id("msg"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, sender_type=payload.author_type or ("customer" if payload.direction == "inbound" else "human"), sender_id=payload.author_id, channel_provider="telegram", external_message_id=dedupe_external_message_id, body=payload.text.strip(), attachments_json="[]", delivery_status="sent" if payload.direction == "outbound" else "received", read_status="unread" if payload.direction == "inbound" else None, message_metadata_json=json.dumps(payload.metadata or {}, ensure_ascii=False), sent_at=event_ts, created_at=event_ts, ) session.add(message) _upsert_external_link( session, deal=deal, communication_id=communication.communication_id, channel_provider="telegram", external_thread_id=payload.thread_id, external_chat_id=payload.chat_id, interaction_id=payload.interaction_id, customer_id=payload.customer_id or deal.customer_id, phone_number=payload.phone_number, external_status=payload.status or payload.ai_state, metadata={ "display_name": payload.display_name, "queue_id": payload.queue_id, "last_direction": payload.direction, "last_author_type": payload.author_type, "ai_state": payload.ai_state, "ai_handoff_reason": payload.ai_handoff_reason, **(payload.metadata or {}), }, ) deal.current_channel = "telegram" deal.preferred_channel = "telegram" deal.updated_at = utc_now_iso() session.commit() return _build_workspace(session, deal) finally: session.close() @app.post("/internal/sales-sync/voice", response_model=SalesWorkspaceOut) def sync_voice_session( payload: SalesVoiceSyncIn, actor: dict = Depends(require_roles(Role.ADMIN)), ) -> SalesWorkspaceOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="voice", provider_name="voice_runtime", metadata=payload.metadata, external_identifier=payload.voice_session_id or payload.call_id or payload.caller_number, ) lead, deal = _resolve_or_create_sync_deal( session, tenant_id=tenant_id, source_channel="voice", preferred_channel="voice", phone=payload.caller_number, display_name=payload.caller_name, interaction_id=payload.interaction_id, voice_session_id=payload.voice_session_id, external_call_id=payload.call_id, subject=payload.summary or payload.caller_name or "Voice inquiry", ) communication = _get_or_create_voice_communication( session, deal=deal, lead=lead, subject=(payload.summary or payload.caller_name or f"Voice call {payload.call_id}")[:120], metadata={ "voice_session_id": payload.voice_session_id, "ai_session_id": payload.ai_session_id, "interaction_id": payload.interaction_id, "queue_id": payload.queue_id, "queue_code": payload.queue_code, "ai_state": payload.ai_state, "handoff_reason": payload.handoff_reason, **(payload.metadata or {}), }, voice_session_id=payload.voice_session_id, ) call = None existing_link = _find_external_link( session, tenant_id=tenant_id, voice_session_id=payload.voice_session_id, external_call_id=payload.call_id, interaction_id=payload.interaction_id, ) if existing_link and existing_link.sales_call_id: call = session.execute( select(SalesCallRow).where(SalesCallRow.call_id == existing_link.sales_call_id, SalesCallRow.tenant_id == tenant_id) ).scalar_one_or_none() if call is None: call = session.execute( select(SalesCallRow) .where(SalesCallRow.external_call_id == payload.call_id, SalesCallRow.tenant_id == tenant_id) .order_by(SalesCallRow.id.desc()) ).scalars().first() started_at = str(payload.started_at or utc_now_iso()).strip() or utc_now_iso() ended_at = str(payload.ended_at or "").strip() or None if call is None: call = SalesCallRow( call_id=new_id("cal"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, phone_number=str(payload.caller_number or "").strip() or (lead.phone if lead else "") or "unknown", direction="inbound", provider="voice_runtime", external_call_id=payload.call_id, recording_url=None, transcript_status="ready" if str(payload.transcript_text or "").strip() else "pending", transcript_id=None, call_status=payload.call_status or payload.ai_state or "started", summary=payload.summary, started_at=started_at, ended_at=ended_at, duration_sec=None, created_at=started_at, updated_at=utc_now_iso(), ) session.add(call) else: call.communication_id = communication.communication_id call.phone_number = str(payload.caller_number or "").strip() or call.phone_number call.provider = "voice_runtime" call.external_call_id = payload.call_id call.call_status = payload.call_status or payload.ai_state or call.call_status call.summary = payload.summary or call.summary call.started_at = started_at or call.started_at call.ended_at = ended_at or call.ended_at call.transcript_status = "ready" if str(payload.transcript_text or "").strip() else call.transcript_status call.updated_at = utc_now_iso() if str(payload.transcript_text or "").strip(): transcript = None if call.transcript_id: transcript = session.execute( select(SalesTranscriptRow).where( SalesTranscriptRow.transcript_id == call.transcript_id, SalesTranscriptRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if transcript is None: transcript = SalesTranscriptRow( transcript_id=new_id("trn"), tenant_id=deal.tenant_id, call_id=call.call_id, language="ru", transcript_text=payload.transcript_text.strip(), diarization_json="{}", extracted_entities_json=json.dumps(payload.metadata or {}, ensure_ascii=False), created_at=utc_now_iso(), updated_at=utc_now_iso(), ) session.add(transcript) call.transcript_id = transcript.transcript_id else: transcript.transcript_text = payload.transcript_text.strip() transcript.extracted_entities_json = json.dumps(payload.metadata or {}, ensure_ascii=False) transcript.updated_at = utc_now_iso() _upsert_external_link( session, deal=deal, communication_id=communication.communication_id, sales_call_id=call.call_id, channel_provider="voice", external_call_id=payload.call_id, voice_session_id=payload.voice_session_id, ai_session_id=payload.ai_session_id, interaction_id=payload.interaction_id, customer_id=deal.customer_id, phone_number=payload.caller_number, external_status=payload.call_status or payload.ai_state or payload.telephony_status, metadata={ "caller_name": payload.caller_name, "queue_id": payload.queue_id, "queue_code": payload.queue_code, "telephony_status": payload.telephony_status, "ai_state": payload.ai_state, "handoff_reason": payload.handoff_reason, **(payload.metadata or {}), }, ) deal.current_channel = "voice" deal.preferred_channel = "voice" deal.updated_at = utc_now_iso() session.commit() return _build_workspace(session, deal) finally: session.close() @app.post("/api/v1/messages/inbound-webhook", response_model=SalesMessageOut) def inbound_message( payload: SalesMessageWebhookIn, actor: dict = Depends(get_actor), ) -> SalesMessageOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="message", provider_name=payload.channel_provider, metadata=payload.metadata, external_identifier=payload.phone, ) lead, deal = _resolve_deal_for_inbound( session, tenant_id=tenant_id, deal_id=payload.deal_id, lead_id=payload.lead_id, phone=payload.phone, subject=payload.body[:120], source_channel=payload.channel_provider, preferred_channel=payload.channel_provider if payload.channel_provider in {"telegram", "whatsapp", "webchat", "email"} else "telegram", ) communication = _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="text", direction="inbound", agent_type="text_ai", subject=payload.body[:120], summary=None, ), metadata=payload.metadata, ) message = SalesMessageRow( message_id=new_id("msg"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, sender_type="customer", sender_id=payload.sender_id, channel_provider=payload.channel_provider, external_message_id=payload.external_message_id, body=payload.body, attachments_json=json.dumps(payload.attachments, ensure_ascii=False), delivery_status="received", read_status="unread", message_metadata_json=json.dumps(payload.metadata, ensure_ascii=False), sent_at=utc_now_iso(), created_at=utc_now_iso(), ) session.add(message) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.MESSAGE_RECEIVED, aggregate_type="message", aggregate_id=message.message_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "communication_session_id": communication.communication_id, "message_id": message.message_id, "channel_type": communication.channel_type, "channel_provider": message.channel_provider, "external_message_id": message.external_message_id, }, ) session.commit() return _message_to_out(message) finally: session.close() @app.post("/api/v1/messages/outbound", response_model=SalesMessageOut) def outbound_message( payload: SalesMessageSendIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesMessageOut: session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, payload.deal_id, tenant_id) communication = None if payload.communication_id: communication = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == payload.communication_id, SalesCommunicationSessionRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if communication is not None and communication.deal_id != deal.deal_id: raise HTTPException(status_code=400, detail="Communication does not belong to deal") if communication is None: lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None communication = _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="text", direction="outbound", agent_type=payload.sender_type if payload.sender_type == "human" else "text_ai", subject=payload.body[:120], ), metadata=payload.metadata, ) message = SalesMessageRow( message_id=new_id("msg"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, sender_type=payload.sender_type, sender_id=payload.sender_id, channel_provider=payload.channel_provider, external_message_id=None, body=payload.body, attachments_json=json.dumps(payload.attachments, ensure_ascii=False), delivery_status="queued", read_status=None, message_metadata_json=json.dumps(payload.metadata, ensure_ascii=False), sent_at=utc_now_iso(), created_at=utc_now_iso(), ) session.add(message) _touch_deal_contact(deal) _bridge_telegram_outbound(communication, message, deal) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.MESSAGE_SENT, aggregate_type="message", aggregate_id=message.message_id, actor_type="human" if payload.sender_type == "human" else "system", actor_id=payload.sender_id or actor["user"], payload={ "deal_id": deal.deal_id, "communication_session_id": communication.communication_id, "message_id": message.message_id, "channel_type": communication.channel_type, "channel_provider": message.channel_provider, "delivery_status": message.delivery_status, }, ) session.commit() return _message_to_out(message) finally: session.close() @app.get("/api/v1/deals/{deal_id}/messages", response_model=list[SalesMessageOut]) def list_messages( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesMessageOut]: session = get_session() try: tenant_id = _tenant_id(actor) _get_deal(session, deal_id, tenant_id) rows = session.execute( select(SalesMessageRow) .where(SalesMessageRow.deal_id == deal_id, SalesMessageRow.tenant_id == tenant_id) .order_by(SalesMessageRow.sent_at.desc(), SalesMessageRow.id.desc()) ).scalars().all() return [_message_to_out(row) for row in rows] finally: session.close() @app.post("/api/v1/calls/inbound-webhook", response_model=SalesCallOut) def inbound_call( payload: SalesCallWebhookIn, actor: dict = Depends(get_actor), ) -> SalesCallOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="voice", provider_name=payload.provider, metadata={"external_identifier": payload.external_call_id, "phone_number": payload.phone_number}, external_identifier=payload.external_call_id or payload.phone_number, ) lead, deal = _resolve_deal_for_inbound( session, tenant_id=tenant_id, deal_id=payload.deal_id, lead_id=payload.lead_id, phone=payload.phone_number, subject=payload.subject, source_channel="voice", preferred_channel="voice", ) communication = _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="voice", direction="inbound", agent_type="voice_ai", subject=payload.subject or f"Входящий звонок {payload.phone_number}", ), metadata={"provider": payload.provider}, ) row = SalesCallRow( call_id=new_id("cal"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, phone_number=payload.phone_number, direction="inbound", provider=payload.provider, external_call_id=payload.external_call_id, recording_url=None, transcript_status="pending", transcript_id=None, call_status="started", summary=None, started_at=utc_now_iso(), ended_at=None, duration_sec=None, created_at=utc_now_iso(), updated_at=utc_now_iso(), ) session.add(row) _bridge_voice_runtime(row, communication, deal) _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.CALL_RECEIVED, aggregate_type="call", aggregate_id=row.call_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "communication_session_id": communication.communication_id, "call_id": row.call_id, "phone_number": row.phone_number, "provider": row.provider, "external_call_id": row.external_call_id, }, ) session.commit() return _call_to_out(row) finally: session.close() @app.post("/api/v1/calls/outbound", response_model=SalesCallOut) def outbound_call( payload: SalesCallWebhookIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCallOut: if not payload.deal_id: raise HTTPException(status_code=400, detail="deal_id is required for outbound calls") session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, payload.deal_id, tenant_id) lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None communication = _start_communication( session, deal=deal, lead=lead, payload=SalesCommunicationStartIn( channel_type="voice", direction="outbound", agent_type="voice_ai", subject=payload.subject or f"Исходящий звонок {payload.phone_number}", ), metadata={"provider": payload.provider}, ) row = SalesCallRow( call_id=new_id("cal"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, communication_id=communication.communication_id, phone_number=payload.phone_number, direction="outbound", provider=payload.provider, external_call_id=payload.external_call_id, recording_url=None, transcript_status="pending", transcript_id=None, call_status="started", summary=None, started_at=utc_now_iso(), ended_at=None, duration_sec=None, created_at=utc_now_iso(), updated_at=utc_now_iso(), ) session.add(row) _bridge_voice_runtime(row, communication, deal) session.commit() return _call_to_out(row) finally: session.close() @app.post("/api/v1/calls/{call_id}/complete", response_model=SalesCallOut) def complete_call( call_id: str, payload: SalesCallCompleteIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCallOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesCallRow).where(SalesCallRow.call_id == call_id, SalesCallRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Call not found") row.summary = payload.summary row.recording_url = payload.recording_url row.call_status = payload.result_code or "completed" row.ended_at = utc_now_iso() if row.started_at: started = datetime.fromisoformat(row.started_at.replace("Z", "+00:00")) row.duration_sec = max(int((_now() - started).total_seconds()), 0) row.updated_at = utc_now_iso() communication = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == row.communication_id, SalesCommunicationSessionRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if communication is not None: communication.ended_at = row.ended_at communication.duration_sec = row.duration_sec communication.summary = payload.summary or communication.summary communication.result_code = payload.result_code communication.status = "completed" communication.updated_at = row.updated_at deal = _get_deal(session, row.deal_id, tenant_id) _touch_deal_contact(deal) _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.CALL_COMPLETED, aggregate_type="call", aggregate_id=row.call_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "call_id": row.call_id, "communication_session_id": row.communication_id, "duration_sec": row.duration_sec, "result_code": payload.result_code, "transcript_id": row.transcript_id, }, ) if payload.summary: _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.COMMUNICATION_SUMMARY_CREATED, aggregate_type="communication", aggregate_id=row.communication_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "call_id": row.call_id, "communication_session_id": row.communication_id, "duration_sec": row.duration_sec, "transcript_id": row.transcript_id, "summary": payload.summary, }, ) session.commit() return _call_to_out(row) finally: session.close() @app.get("/api/v1/deals/{deal_id}/calls", response_model=list[SalesCallOut]) def list_calls( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesCallOut]: session = get_session() try: tenant_id = _tenant_id(actor) _get_deal(session, deal_id, tenant_id) rows = session.execute( select(SalesCallRow) .where(SalesCallRow.deal_id == deal_id, SalesCallRow.tenant_id == tenant_id) .order_by(SalesCallRow.started_at.desc(), SalesCallRow.id.desc()) ).scalars().all() return [_call_to_out(row) for row in rows] finally: session.close() @app.post("/api/v1/calls/{call_id}/transcript", response_model=SalesTranscriptOut) def attach_transcript( call_id: str, payload: SalesTranscriptIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesTranscriptOut: session = get_session() try: tenant_id = _tenant_id(actor) call = session.execute( select(SalesCallRow).where(SalesCallRow.call_id == call_id, SalesCallRow.tenant_id == tenant_id) ).scalar_one_or_none() if call is None: raise HTTPException(status_code=404, detail="Call not found") now = utc_now_iso() row = SalesTranscriptRow( transcript_id=new_id("trn"), tenant_id=tenant_id, call_id=call.call_id, language=payload.language, transcript_text=payload.transcript_text, diarization_json=json.dumps(payload.diarization, ensure_ascii=False), extracted_entities_json=json.dumps(payload.extracted_entities, ensure_ascii=False), created_at=now, updated_at=now, ) session.add(row) call.transcript_id = row.transcript_id call.transcript_status = "ready" call.updated_at = now communication = session.execute( select(SalesCommunicationSessionRow).where( SalesCommunicationSessionRow.communication_id == call.communication_id, SalesCommunicationSessionRow.tenant_id == tenant_id, ) ).scalar_one_or_none() if communication is not None: communication.transcript_id = row.transcript_id communication.updated_at = now session.commit() return _transcript_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/offers", response_model=SalesOfferOut) def create_offer( deal_id: str, payload: SalesOfferCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesOfferOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) now = utc_now_iso() row = SalesOfferRow( offer_id=new_id("off"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, offer_type=payload.offer_type, title=payload.title, description=payload.description, line_items_json=json.dumps(payload.line_items, ensure_ascii=False), pricing_json=json.dumps(payload.pricing, ensure_ascii=False), total_amount=payload.total_amount, currency=payload.currency, validity_until=payload.validity_until, status="draft", rendered_document_url=payload.rendered_document_url, created_by_type=actor["user"], created_at=now, updated_at=now, ) session.add(row) _apply_stage(session, deal, "offer_preparing", actor_type="human", actor_id=actor["user"], reason="offer.created") _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.OFFER_CREATED, aggregate_type="offer", aggregate_id=row.offer_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type, "total_amount": row.total_amount, "currency": row.currency, }, ) session.commit() return _offer_to_out(row) finally: session.close() @app.get("/api/v1/offers/{offer_id}", response_model=SalesOfferOut) def get_offer( offer_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesOfferOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Offer not found") return _offer_to_out(row) finally: session.close() @app.post("/api/v1/offers/{offer_id}/send", response_model=SalesOfferOut) def send_offer( offer_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesOfferOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Offer not found") row.status = "sent" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "offer_sent", actor_type="system", actor_id=None, reason="offer.sent") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.OFFER_SENT, aggregate_type="offer", aggregate_id=row.offer_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type, "total_amount": row.total_amount, "currency": row.currency, }, ) session.commit() return _offer_to_out(row) finally: session.close() @app.post("/api/v1/offers/{offer_id}/accept", response_model=SalesOfferOut) def accept_offer( offer_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesOfferOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Offer not found") row.status = "accepted" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "conditions_negotiation", actor_type="system", actor_id=None, reason="offer.accepted") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.OFFER_ACCEPTED, aggregate_type="offer", aggregate_id=row.offer_id, actor_type="system", actor_id=None, payload={"deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type}, ) session.commit() return _offer_to_out(row) finally: session.close() @app.post("/api/v1/offers/{offer_id}/reject", response_model=SalesOfferOut) def reject_offer( offer_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesOfferOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Offer not found") row.status = "rejected" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "need_clarification", actor_type="system", actor_id=None, reason="offer.rejected") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.OFFER_REJECTED, aggregate_type="offer", aggregate_id=row.offer_id, actor_type="system", actor_id=None, payload={"deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type}, ) session.commit() return _offer_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/conditions", response_model=SalesConditionOut) def upsert_conditions( deal_id: str, payload: SalesConditionUpsertIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesConditionOut: session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, deal_id, tenant_id) row = session.execute( select(SalesConditionRow).where(SalesConditionRow.deal_id == deal.deal_id, SalesConditionRow.tenant_id == tenant_id) ).scalar_one_or_none() now = utc_now_iso() if row is None: row = SalesConditionRow( condition_id=new_id("con"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, product_name=payload.product_name, service_name=payload.service_name, quantity=payload.quantity, unit=payload.unit, delivery_mode=payload.delivery_mode, execution_date=payload.execution_date, start_date=payload.start_date, end_date=payload.end_date, payment_terms=payload.payment_terms, custom_terms_json=json.dumps(payload.custom_terms, ensure_ascii=False), agreed_price=payload.agreed_price, currency=payload.currency, confirmed_at=now, created_at=now, updated_at=now, ) session.add(row) else: row.product_name = payload.product_name row.service_name = payload.service_name row.quantity = payload.quantity row.unit = payload.unit row.delivery_mode = payload.delivery_mode row.execution_date = payload.execution_date row.start_date = payload.start_date row.end_date = payload.end_date row.payment_terms = payload.payment_terms row.custom_terms_json = json.dumps(payload.custom_terms, ensure_ascii=False) row.agreed_price = payload.agreed_price row.currency = payload.currency row.confirmed_at = now row.updated_at = now if payload.agreed_price is not None: deal.final_amount = payload.agreed_price _apply_stage(session, deal, "conditions_negotiation", actor_type="human", actor_id=actor["user"], reason="deal.conditions_confirmed") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DEAL_CONDITIONS_CONFIRMED, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "condition_id": row.condition_id, "agreed_price": row.agreed_price, "currency": row.currency, "payment_terms": row.payment_terms, }, ) session.commit() return _condition_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut) def create_counterparty( deal_id: str, payload: SalesCounterpartyUpsertIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCounterpartyOut: return patch_counterparty(deal_id, payload, actor) @app.patch("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut) def patch_counterparty( deal_id: str, payload: SalesCounterpartyUpsertIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesCounterpartyOut: session = get_session() try: tenant_id = _tenant_id(actor) deal = _get_deal(session, deal_id, tenant_id) row = session.execute( select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == tenant_id) ).scalar_one_or_none() now = utc_now_iso() status = _counterparty_status(payload) if row is None: row = SalesCounterpartyRow( counterparty_id=new_id("ctp"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, customer_id=deal.customer_id, company_name=payload.company_name, full_name=payload.full_name, bin_iin=payload.bin_iin, address=payload.address, bank_details_json=json.dumps(payload.bank_details, ensure_ascii=False), signer_name=payload.signer_name, signer_role=payload.signer_role, signer_basis=payload.signer_basis, email_for_docs=payload.email_for_docs, phone_for_docs=payload.phone_for_docs, completeness_status=status, created_at=now, updated_at=now, ) session.add(row) else: row.company_name = payload.company_name row.full_name = payload.full_name row.bin_iin = payload.bin_iin row.address = payload.address row.bank_details_json = json.dumps(payload.bank_details, ensure_ascii=False) row.signer_name = payload.signer_name row.signer_role = payload.signer_role row.signer_basis = payload.signer_basis row.email_for_docs = payload.email_for_docs row.phone_for_docs = payload.phone_for_docs row.completeness_status = status row.updated_at = now row.customer_id = _ensure_crm_customer( session, lead=_get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None, deal=deal, counterparty=row, ) _apply_stage( session, deal, "counterparty_data_received" if status == "completed" else "counterparty_data_requested", actor_type="human", actor_id=actor["user"], reason="counterparty.completed" if status == "completed" else "counterparty.requested", ) if status == "completed": _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.COUNTERPARTY_COMPLETED, aggregate_type="counterparty", aggregate_id=row.counterparty_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "counterparty_id": row.counterparty_id, "customer_id": row.customer_id, "completeness_status": row.completeness_status, }, ) session.commit() return _counterparty_to_out(row) finally: session.close() @app.get("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut) def get_counterparty( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesCounterpartyOut: session = get_session() try: tenant_id = _tenant_id(actor) _get_deal(session, deal_id, tenant_id) row = session.execute( select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal_id, SalesCounterpartyRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Counterparty not found") return _counterparty_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/documents", response_model=SalesDocumentOut) def create_document( deal_id: str, payload: SalesDocumentCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDocumentOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) now = utc_now_iso() row = SalesDocumentRow( document_id=new_id("doc"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, customer_id=deal.customer_id, document_type=payload.document_type, template_id=payload.template_id, version=payload.version, status="draft", file_url=payload.file_url, rendered_payload_json=json.dumps(payload.rendered_payload, ensure_ascii=False), external_sign_provider_id=None, signed_at=None, created_at=now, updated_at=now, ) session.add(row) _apply_stage(session, deal, "document_preparing", actor_type="human", actor_id=actor["user"], reason="document.created") _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.DOCUMENT_CREATED, aggregate_type="document", aggregate_id=row.document_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "document_id": row.document_id, "document_type": row.document_type, "status": row.status, }, ) session.commit() return _document_to_out(row) finally: session.close() @app.get("/api/v1/documents/{document_id}", response_model=SalesDocumentOut) def get_document( document_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesDocumentOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Document not found") return _document_to_out(row) finally: session.close() @app.post("/api/v1/documents/{document_id}/send", response_model=SalesDocumentOut) def send_document( document_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDocumentOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Document not found") row.status = "sent" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "document_sent", actor_type="system", actor_id=None, reason="document.sent") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DOCUMENT_SENT, aggregate_type="document", aggregate_id=row.document_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "document_id": row.document_id, "document_type": row.document_type, "status": row.status, }, ) session.commit() return _document_to_out(row) finally: session.close() @app.post("/api/v1/documents/{document_id}/confirm", response_model=SalesDocumentOut) def confirm_document( document_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesDocumentOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Document not found") row.status = "confirmed" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "document_confirmed", actor_type="system", actor_id=None, reason="document.confirmed") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DOCUMENT_CONFIRMED, aggregate_type="document", aggregate_id=row.document_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "document_id": row.document_id, "document_type": row.document_type, "status": row.status, }, ) session.commit() return _document_to_out(row) finally: session.close() @app.post("/api/v1/documents/{document_id}/sign-status-webhook", response_model=SalesDocumentOut) def sign_document( document_id: str, x_provider_account_id: str | None = Header(default=None, alias="X-Provider-Account-ID"), x_provider_name: str | None = Header(default=None, alias="X-Provider-Name"), actor: dict = Depends(get_actor), ) -> SalesDocumentOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="document_sign", provider_name=x_provider_name or "generic", provider_account_id=x_provider_account_id, ) row = session.execute( select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Document not found") row.status = "signed" row.signed_at = utc_now_iso() row.updated_at = row.signed_at _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DOCUMENT_SIGNED, aggregate_type="document", aggregate_id=row.document_id, actor_type="system", actor_id=None, payload={ "deal_id": row.deal_id, "document_id": row.document_id, "document_type": row.document_type, "status": row.status, "signed_at": row.signed_at, }, ) session.commit() return _document_to_out(row) finally: session.close() @app.post("/api/v1/deals/{deal_id}/invoices", response_model=SalesInvoiceOut) def create_invoice( deal_id: str, payload: SalesInvoiceCreate, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesInvoiceOut: session = get_session() try: deal = _get_deal(session, deal_id, _tenant_id(actor)) now = utc_now_iso() row = SalesInvoiceRow( invoice_id=new_id("inv"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, customer_id=deal.customer_id, invoice_number=_invoice_number(), basis_document_id=payload.basis_document_id, amount=payload.amount, currency=payload.currency, due_date=payload.due_date, status="draft", payment_link=payload.payment_link, line_items_json=json.dumps(payload.line_items, ensure_ascii=False), metadata_json=json.dumps(payload.metadata, ensure_ascii=False), issued_at=None, paid_at=None, created_at=now, updated_at=now, ) session.add(row) _apply_stage(session, deal, "invoice_preparing", actor_type="human", actor_id=actor["user"], reason="invoice.created") _publish_sales_event( session, tenant_id=deal.tenant_id, event_type=sales_event_types.INVOICE_CREATED, aggregate_type="invoice", aggregate_id=row.invoice_id, actor_type="human", actor_id=actor["user"], payload={ "deal_id": deal.deal_id, "invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "amount": row.amount, "currency": row.currency, "due_date": row.due_date, }, ) session.commit() return _invoice_to_out(row) finally: session.close() @app.get("/api/v1/invoices/{invoice_id}", response_model=SalesInvoiceOut) def get_invoice( invoice_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> SalesInvoiceOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Invoice not found") return _invoice_to_out(row) finally: session.close() @app.post("/api/v1/invoices/{invoice_id}/send", response_model=SalesInvoiceOut) def send_invoice( invoice_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesInvoiceOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Invoice not found") row.status = "sent" row.issued_at = utc_now_iso() row.updated_at = row.issued_at deal = _get_deal(session, row.deal_id, tenant_id) deal.next_action_type = "payment_follow_up" deal.next_action_at = row.due_date _schedule_task( session, deal=deal, task_type="invoice_follow_up", run_at=row.due_date, payload={"invoice_id": row.invoice_id, "invoice_number": row.invoice_number}, ) _apply_stage(session, deal, "invoice_sent", actor_type="system", actor_id=None, reason="invoice.sent") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.INVOICE_SENT, aggregate_type="invoice", aggregate_id=row.invoice_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "amount": row.amount, "currency": row.currency, "due_date": row.due_date, }, ) session.commit() return _invoice_to_out(row) finally: session.close() @app.post("/api/v1/invoices/{invoice_id}/mark-overdue", response_model=SalesInvoiceOut) def mark_invoice_overdue( invoice_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesInvoiceOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Invoice not found") row.status = "overdue" row.updated_at = utc_now_iso() deal = _get_deal(session, row.deal_id, tenant_id) _apply_stage(session, deal, "payment_overdue", actor_type="system", actor_id=None, reason="invoice.overdue") _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.INVOICE_OVERDUE, aggregate_type="invoice", aggregate_id=row.invoice_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "amount": row.amount, "currency": row.currency, "due_date": row.due_date, }, ) session.commit() return _invoice_to_out(row) finally: session.close() @app.post("/api/v1/payments/webhook", response_model=SalesPaymentOut) def payment_webhook( payload: SalesPaymentWebhookIn, actor: dict = Depends(get_actor), ) -> SalesPaymentOut: session = get_session() try: tenant_id = _resolve_provider_tenant_id( session, actor, provider_type="payment", provider_name=payload.payment_provider, metadata=payload.metadata, external_identifier=payload.external_payment_id, ) deal = _get_deal(session, payload.deal_id, tenant_id) invoice = None if payload.invoice_id: invoice = session.execute( select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == payload.invoice_id, SalesInvoiceRow.tenant_id == tenant_id) ).scalar_one_or_none() if invoice is None: raise HTTPException(status_code=404, detail="Invoice not found") if invoice.deal_id != deal.deal_id: raise HTTPException(status_code=400, detail="Invoice does not belong to deal") now = utc_now_iso() row = None previous_payment_status = None if payload.external_payment_id: row = session.execute( select(SalesPaymentRow).where( SalesPaymentRow.tenant_id == tenant_id, SalesPaymentRow.payment_provider == payload.payment_provider, SalesPaymentRow.external_payment_id == payload.external_payment_id, ) ).scalar_one_or_none() previous_payment_status = row.status if row is not None else None previous_invoice_status = invoice.status if invoice is not None else None previous_deal_status = deal.status if row is None: row = SalesPaymentRow( payment_id=new_id("pay"), tenant_id=deal.tenant_id, deal_id=deal.deal_id, invoice_id=payload.invoice_id, payment_provider=payload.payment_provider, external_payment_id=payload.external_payment_id, amount=payload.amount, currency=payload.currency, status=payload.status, paid_at=payload.paid_at or (now if payload.status in {"success", "partial"} else None), payment_method=payload.payment_method, failure_reason=payload.failure_reason, metadata_json=json.dumps(payload.metadata, ensure_ascii=False), created_at=now, updated_at=now, ) session.add(row) else: row.deal_id = deal.deal_id row.invoice_id = payload.invoice_id row.amount = payload.amount row.currency = payload.currency row.status = payload.status row.paid_at = payload.paid_at or row.paid_at or (now if payload.status in {"success", "partial"} else None) row.payment_method = payload.payment_method row.failure_reason = payload.failure_reason row.metadata_json = json.dumps(payload.metadata, ensure_ascii=False) row.updated_at = now should_publish_payment_received = payload.status in {"success", "partial"} and previous_payment_status != payload.status if should_publish_payment_received: _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.PAYMENT_RECEIVED, aggregate_type="payment", aggregate_id=row.payment_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "invoice_id": row.invoice_id, "payment_id": row.payment_id, "payment_provider": row.payment_provider, "external_payment_id": row.external_payment_id, "amount": row.amount, "currency": row.currency, "status": row.status, }, ) lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None counterparty = session.execute( select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == tenant_id) ).scalar_one_or_none() if invoice is not None: if payload.status == "success": invoice.status = "paid" invoice.paid_at = row.paid_at invoice.updated_at = now deal.status = "won" deal.closed_at = row.paid_at deal.final_amount = payload.amount _apply_stage(session, deal, "won", actor_type="system", actor_id=None, reason="payment.received") _ensure_crm_customer(session, lead=lead, deal=deal, counterparty=counterparty) if previous_invoice_status != "paid": _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.INVOICE_PAID, aggregate_type="invoice", aggregate_id=invoice.invoice_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "invoice_id": invoice.invoice_id, "paid_amount": payload.amount, "currency": payload.currency, "paid_at": row.paid_at, }, ) if previous_deal_status != "won": _publish_sales_event( session, tenant_id=tenant_id, event_type=sales_event_types.DEAL_WON, aggregate_type="deal", aggregate_id=deal.deal_id, actor_type="system", actor_id=None, payload={ "deal_id": deal.deal_id, "invoice_id": invoice.invoice_id, "payment_id": row.payment_id, "final_amount": deal.final_amount, "currency": deal.currency, "won_reason": "payment_received", }, ) elif payload.status == "partial": invoice.status = "partially_paid" invoice.updated_at = now _apply_stage(session, deal, "partially_paid", actor_type="system", actor_id=None, reason="payment.partial") elif payload.status == "failed": invoice.updated_at = now session.commit() return _payment_to_out(row) finally: session.close() @app.get("/api/v1/deals/{deal_id}/payments", response_model=list[SalesPaymentOut]) def list_payments( deal_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)), ) -> list[SalesPaymentOut]: session = get_session() try: tenant_id = _tenant_id(actor) _get_deal(session, deal_id, tenant_id) rows = session.execute( select(SalesPaymentRow) .where(SalesPaymentRow.deal_id == deal_id, SalesPaymentRow.tenant_id == tenant_id) .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc()) ).scalars().all() return [_payment_to_out(row) for row in rows] finally: session.close() @app.post("/api/v1/payments/{payment_id}/reconcile", response_model=SalesPaymentOut) def reconcile_payment( payment_id: str, payload: SalesPaymentReconcileIn, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> SalesPaymentOut: session = get_session() try: tenant_id = _tenant_id(actor) row = session.execute( select(SalesPaymentRow).where(SalesPaymentRow.payment_id == payment_id, SalesPaymentRow.tenant_id == tenant_id) ).scalar_one_or_none() if row is None: raise HTTPException(status_code=404, detail="Payment not found") row.status = payload.status row.failure_reason = payload.failure_reason row.metadata_json = json.dumps(payload.metadata, ensure_ascii=False) row.updated_at = utc_now_iso() session.commit() return _payment_to_out(row) finally: session.close()