call-center/services/customer_service/app.py

from __future__ import annotations

import json
from datetime import datetime

from fastapi import FastAPI, HTTPException
from sqlalchemy import select

from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.models import (
    CustomerCreate,
    CustomerHistoryEventOut,
    CustomerHistoryOut,
    CustomerHistorySummaryOut,
    CustomerOut,
    HealthResponse,
    InteractionOut,
    TelegramThreadOut,
    VoiceLiveCallOut,
)
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import (
    AsteriskCallLinkRow,
    CallRecordingRow,
    Customer,
    Interaction,
    InteractionTimeline,
    TelegramMessageRow,
    TelegramThreadRow,
    VoiceTranscriptSegmentRow,
)

app = FastAPI(title="customer-service", version="1.0.0")

init_sql_schema()

_CHANNEL_LABELS = {
    "voice": "Голос",
    "telegram": "Telegram",
    "whatsapp": "WhatsApp",
    "webchat": "Веб-чат",
    "email": "Email",
}
_INTERACTION_STATUS_LABELS = {
    "new": "новое",
    "in_progress": "в работе",
    "escalated": "эскалировано",
    "closed": "закрыто",
    "abandoned": "потеряно",
}
_TELEPHONY_STATUS_LABELS = {
    "ringing": "Звонит",
    "claimed": "Взято",
    "connected": "Соединено",
    "ended": "Завершено",
    "failed": "Ошибка",
}
_TIMELINE_ACTION_TITLES = {
    "interaction.created": "Обращение создано",
    "interaction.assigned": "Обращение назначено",
    "interaction.status_changed": "Статус обращения изменён",
    "interaction.escalated": "Обращение эскалировано",
}


def _json_list(raw: str | None) -> list:
    try:
        value = json.loads(raw or "[]")
    except json.JSONDecodeError:
        return []
    return value if isinstance(value, list) else []


def _json_dict(raw: str | None) -> dict:
    try:
        value = json.loads(raw or "{}")
    except json.JSONDecodeError:
        return {}
    return value if isinstance(value, dict) else {}


def _time_value(value: str | None) -> int:
    if not value:
        return 0
    normalized = value.strip()
    if normalized.endswith("Z"):
        normalized = f"{normalized[:-1]}+00:00"
    try:
        return int(datetime.fromisoformat(normalized).timestamp())
    except ValueError:
        return 0


def _channel_label(channel: str | None) -> str:
    return _CHANNEL_LABELS.get(channel or "", channel or "Канал")


def _interaction_status_label(status: str | None) -> str:
    return _INTERACTION_STATUS_LABELS.get(status or "", status or "неизвестно")


def _telephony_status_label(status: str | None) -> str:
    return _TELEPHONY_STATUS_LABELS.get(status or "", status or "неизвестно")


def _timeline_title(action: str) -> str:
    return _TIMELINE_ACTION_TITLES.get(action, action)


def _timeline_body(action: str, metadata: dict) -> str:
    if action == "interaction.assigned":
        assignee = metadata.get("assignee")
        return f"Обращение назначено на {assignee}." if assignee else "Обращение назначено оператору."
    if action == "interaction.status_changed":
        status = metadata.get("status")
        return f"Новый статус: {_interaction_status_label(status)}."
    if action == "interaction.escalated":
        target_queue_id = metadata.get("target_queue_id")
        return (
            f"Обращение передано в очередь {target_queue_id}."
            if target_queue_id
            else "Обращение эскалировано."
        )
    if action == "interaction.created":
        channel = metadata.get("channel")
        return f"Создано новое обращение по каналу {_channel_label(channel)}."
    return "Событие сохранено в ленте обращения."


def _to_out(row: Customer) -> CustomerOut:
    phones = _json_list(row.phones_json)
    tags = _json_list(row.tags_json)
    return CustomerOut(
        customer_id=row.customer_id,
        display_name=row.display_name,
        phones=phones,
        preferred_phone=row.preferred_phone,
        tags=tags,
        created_at=row.created_at,
    )


def _interaction_to_out(row: Interaction) -> InteractionOut:
    return InteractionOut(
        interaction_id=row.interaction_id,
        channel=row.channel,
        subject=row.subject,
        customer_id=row.customer_id,
        queue_id=row.queue_id,
        priority=row.priority,
        status=row.status,
        assigned_to=row.assigned_to,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _telegram_thread_to_out(row: TelegramThreadRow) -> TelegramThreadOut:
    return TelegramThreadOut(
        thread_id=row.thread_id,
        chat_id=row.chat_id,
        interaction_id=row.interaction_id,
        telegram_user_id=row.telegram_user_id,
        username=row.username,
        display_name=row.display_name,
        queue_id=row.queue_id,
        status=row.status,  # type: ignore[arg-type]
        claimed_by_user=row.claimed_by_user,
        claimed_at=row.claimed_at,
        ai_session_id=row.ai_session_id,
        ai_state=row.ai_state,  # type: ignore[arg-type]
        ai_handoff_reason=row.ai_handoff_reason,
        ai_last_model_at=row.ai_last_model_at,
        last_message_at=row.last_message_at,
        last_message_preview=row.last_message_preview,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _voice_call_to_out(row: AsteriskCallLinkRow, *, has_recording: bool) -> VoiceLiveCallOut:
    return VoiceLiveCallOut(
        call_id=row.call_id,
        interaction_id=row.interaction_id,
        queue_id=row.queue_id,
        queue_code=row.queue_code,
        caller_number=row.caller_number,
        caller_name=row.caller_name,
        status=row.status,
        telephony_status=row.telephony_status,  # type: ignore[arg-type]
        claimed_by_user=row.claimed_by_user,
        claimed_at=row.claimed_at,
        operator_extension=row.operator_extension,
        channel_name=row.channel_name,
        started_at=row.started_at,
        connected_at=row.connected_at,
        ended_at=row.ended_at,
        updated_at=row.updated_at,
        last_transition_at=row.ended_at or row.connected_at or row.updated_at,
        hangup_cause=None,
        terminal_action=None,
        terminal_target=None,
        voice_session_id=row.voice_session_id,
        ai_session_id=row.ai_session_id,
        ai_state=row.ai_state,  # type: ignore[arg-type]
        ai_handoff_reason=row.ai_handoff_reason,
        ai_last_model_at=row.ai_last_model_at,
        has_recording=has_recording,
    )


def _history_event_sort_key(event: CustomerHistoryEventOut) -> tuple[int, str]:
    return (_time_value(event.timestamp), event.note)


def _build_customer_history(
    customer: Customer,
    interactions: list[Interaction],
    timelines: list[InteractionTimeline],
    telegram_threads: list[TelegramThreadRow],
    telegram_messages: list[TelegramMessageRow],
    live_calls: list[AsteriskCallLinkRow],
    transcript_segments: list[VoiceTranscriptSegmentRow],
    recording_call_ids: set[str],
) -> CustomerHistoryOut:
    customer_out = _to_out(customer)
    interaction_out = [_interaction_to_out(row) for row in interactions]
    telegram_thread_out = [_telegram_thread_to_out(row) for row in telegram_threads]
    live_call_out = [_voice_call_to_out(row, has_recording=row.call_id in recording_call_ids) for row in live_calls]
    interaction_by_id = {row.interaction_id: row for row in interactions}
    thread_by_id = {row.thread_id: row for row in telegram_threads}
    events: list[CustomerHistoryEventOut] = [
        CustomerHistoryEventOut(
            timestamp=customer.created_at,
            kind="profile",
            title="Профиль клиента активен",
            body=(
                f"Контакт доступен по номеру {customer_out.phones[0]}."
                if customer_out.phones
                else "Контакт добавлен в клиентскую базу."
            ),
            note=" • ".join(filter(None, [customer.customer_id, customer_out.preferred_phone])),
        )
    ]

    for row in interactions:
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.updated_at or row.created_at,
                kind=row.channel or "case",
                title=f"{_channel_label(row.channel)} • {_interaction_status_label(row.status)}",
                body=row.subject or "Обращение без темы",
                note=" • ".join(
                    filter(
                        None,
                        [
                            row.interaction_id,
                            f"оператор {row.assigned_to}" if row.assigned_to else "",
                            f"queue {row.queue_id}" if row.queue_id else "",
                        ],
                    )
                ),
                interaction_id=row.interaction_id,
            )
        )

    for row in timelines:
        interaction = interaction_by_id.get(row.interaction_id)
        metadata = _json_dict(row.metadata_json)
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.timestamp,
                kind=interaction.channel if interaction else "case",
                title=_timeline_title(row.action),
                body=_timeline_body(row.action, metadata),
                note=" • ".join(
                    filter(
                        None,
                        [
                            row.interaction_id,
                            f"оператор {interaction.assigned_to}" if interaction and interaction.assigned_to else "",
                        ],
                    )
                ),
                interaction_id=row.interaction_id,
            )
        )

    for row in telegram_threads:
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.last_message_at or row.updated_at or row.created_at,
                kind="telegram",
                title="Telegram диалог",
                body=row.last_message_preview or "Последнее сообщение недоступно.",
                note=" • ".join(
                    filter(
                        None,
                        [
                            row.display_name or row.username or f"chat {row.chat_id}",
                            f"оператор {row.claimed_by_user}" if row.claimed_by_user else "",
                            _interaction_status_label(row.status),
                        ],
                    )
                ),
                interaction_id=row.interaction_id,
                thread_id=row.thread_id,
            )
        )

    for row in telegram_messages:
        thread = thread_by_id.get(row.thread_id or "")
        direction = {
            "outbound": "Исходящее сообщение",
            "system": "Системное сообщение",
        }.get(row.direction, "Входящее сообщение")
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.created_at,
                kind="telegram",
                title=direction,
                body=row.text or "Сообщение без текста",
                note=" • ".join(
                    filter(
                        None,
                        [
                            thread.display_name or thread.username if thread else "",
                            f"оператор {row.operator_user}" if row.operator_user else "",
                            row.delivery_status or "",
                        ],
                    )
                ),
                interaction_id=row.interaction_id,
                thread_id=row.thread_id,
            )
        )

    for row in live_calls:
        completed = row.telephony_status == "ended" or bool(row.ended_at)
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.ended_at or row.connected_at or row.updated_at or row.started_at,
                kind="voice",
                title=(
                    f"Звонок • {_telephony_status_label('ended')}"
                    if completed
                    else f"Звонок • {_telephony_status_label(row.telephony_status)}"
                ),
                body=f"Номер клиента: {row.caller_number or 'не определён'}",
                note=" • ".join(
                    filter(
                        None,
                        [
                            row.call_id,
                            f"внутр. {row.operator_extension}" if row.operator_extension else "",
                            f"оператор {row.claimed_by_user}" if row.claimed_by_user else "",
                        ],
                    )
                ),
                interaction_id=row.interaction_id,
                call_id=row.call_id,
            )
        )
        if row.ai_state or row.ai_handoff_reason:
            events.append(
                CustomerHistoryEventOut(
                    timestamp=row.ai_last_model_at or row.updated_at or row.started_at,
                    kind="voice",
                    title=f"Голосовой AI • {row.ai_state or 'active'}",
                    body=row.ai_handoff_reason or "AI обновил контекст разговора для оператора.",
                    note=" • ".join(
                        filter(
                            None,
                            [
                                row.call_id,
                                f"voice {row.voice_session_id}" if row.voice_session_id else "",
                                f"ai {row.ai_session_id}" if row.ai_session_id else "",
                            ],
                        )
                    ),
                    interaction_id=row.interaction_id,
                    call_id=row.call_id,
                )
            )

    for row in transcript_segments:
        speaker = "AI" if row.speaker == "assistant" else "Клиент"
        events.append(
            CustomerHistoryEventOut(
                timestamp=row.created_at,
                kind="voice",
                title=f"Транскрипт • {speaker}",
                body=row.text or "Реплика без текста",
                note=" • ".join(filter(None, [row.call_id, f"seq {row.sequence_no}"])),
                interaction_id=row.interaction_id,
                call_id=row.call_id,
            )
        )

    events.sort(key=_history_event_sort_key, reverse=True)
    active_channels = sorted(
        {
            *[row.channel for row in interactions if row.channel],
            *(["telegram"] if telegram_threads else []),
            *(["voice"] if live_calls else []),
        }
    )
    latest_event = events[0] if events else None
    primary_thread = max(
        telegram_threads,
        key=lambda row: _time_value(row.last_message_at or row.updated_at or row.created_at),
        default=None,
    )
    summary = CustomerHistorySummaryOut(
        contact_points=len(interactions) + len(telegram_threads) + len(live_calls),
        open_cases=sum(1 for row in interactions if row.status != "closed"),
        active_channels=active_channels,
        latest_event_at=latest_event.timestamp if latest_event else None,
        latest_event_title=latest_event.title if latest_event else None,
        primary_phone=customer_out.phones[0] if customer_out.phones else customer_out.preferred_phone,
        primary_telegram_thread_id=primary_thread.thread_id if primary_thread else None,
    )
    return CustomerHistoryOut(
        customer=customer_out,
        summary=summary,
        interactions=interaction_out,
        telegram_threads=telegram_thread_out,
        live_calls=live_call_out,
        history=events[:25],
    )


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return HealthResponse(status="ok", service="customer-service")


@app.post("/customers", response_model=CustomerOut)
def create_customer(payload: CustomerCreate) -> CustomerOut:
    if payload.preferred_phone and payload.preferred_phone not in payload.phones:
        raise HTTPException(status_code=400, detail="preferred_phone must be in phones")

    session = get_session()
    try:
        row = Customer(
            customer_id=new_id("cus"),
            display_name=payload.display_name,
            phones_json=json.dumps(payload.phones, ensure_ascii=False),
            preferred_phone=payload.preferred_phone,
            tags_json=json.dumps(payload.tags, ensure_ascii=False),
            created_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return _to_out(row)
    finally:
        session.close()


@app.get("/customers/{customer_id}", response_model=CustomerOut)
def get_customer(customer_id: str) -> CustomerOut:
    session = get_session()
    try:
        row = session.execute(select(Customer).where(Customer.customer_id == customer_id)).scalar_one_or_none()
        if not row:
            raise HTTPException(status_code=404, detail="Customer not found")
        return _to_out(row)
    finally:
        session.close()


@app.get("/customers", response_model=list[CustomerOut])
def search_customers(query: str | None = None, limit: int = 100) -> list[CustomerOut]:
    session = get_session()
    try:
        rows = session.execute(select(Customer).order_by(Customer.id.desc())).scalars().all()
        out = [_to_out(r) for r in rows]
        if query:
            q = query.strip().lower()
            out = [
                c
                for c in out
                if q in c.display_name.lower()
                or any(q in p.lower() for p in c.phones)
                or any(q in t.lower() for t in c.tags)
            ]
        return out[:limit]
    finally:
        session.close()


@app.get("/customers/{customer_id}/history", response_model=CustomerHistoryOut)
def customer_history(customer_id: str) -> CustomerHistoryOut:
    session = get_session()
    try:
        customer = session.execute(
            select(Customer).where(Customer.customer_id == customer_id)
        ).scalar_one_or_none()
        if not customer:
            raise HTTPException(status_code=404, detail="Customer not found")

        interactions = session.execute(
            select(Interaction)
            .where(Interaction.customer_id == customer_id)
            .order_by(Interaction.updated_at.desc(), Interaction.id.desc())
        ).scalars().all()
        interaction_ids = [row.interaction_id for row in interactions]

        timelines: list[InteractionTimeline] = []
        telegram_threads: list[TelegramThreadRow] = []
        telegram_messages: list[TelegramMessageRow] = []
        live_calls: list[AsteriskCallLinkRow] = []
        transcript_segments: list[VoiceTranscriptSegmentRow] = []
        recording_call_ids: set[str] = set()

        if interaction_ids:
            timelines = session.execute(
                select(InteractionTimeline)
                .where(InteractionTimeline.interaction_id.in_(interaction_ids))
                .order_by(InteractionTimeline.timestamp.desc(), InteractionTimeline.id.desc())
            ).scalars().all()

            telegram_threads = session.execute(
                select(TelegramThreadRow)
                .where(TelegramThreadRow.interaction_id.in_(interaction_ids))
                .order_by(TelegramThreadRow.last_message_at.desc(), TelegramThreadRow.id.desc())
            ).scalars().all()

            live_calls = session.execute(
                select(AsteriskCallLinkRow)
                .where(AsteriskCallLinkRow.interaction_id.in_(interaction_ids))
                .order_by(AsteriskCallLinkRow.updated_at.desc(), AsteriskCallLinkRow.id.desc())
            ).scalars().all()

        thread_ids = [row.thread_id for row in telegram_threads if row.thread_id]
        if thread_ids:
            telegram_messages = session.execute(
                select(TelegramMessageRow)
                .where(TelegramMessageRow.thread_id.in_(thread_ids))
                .order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
                .limit(8)
            ).scalars().all()

        call_ids = [row.call_id for row in live_calls if row.call_id]
        if call_ids:
            recording_call_ids = set(
                session.execute(
                    select(CallRecordingRow.call_id).where(CallRecordingRow.call_id.in_(call_ids))
                ).scalars().all()
            )
            transcript_segments = session.execute(
                select(VoiceTranscriptSegmentRow)
                .where(
                    VoiceTranscriptSegmentRow.call_id.in_(call_ids),
                    VoiceTranscriptSegmentRow.is_final.is_(True),
                )
                .order_by(VoiceTranscriptSegmentRow.created_at.desc(), VoiceTranscriptSegmentRow.id.desc())
                .limit(8)
            ).scalars().all()

        return _build_customer_history(
            customer,
            interactions,
            timelines,
            telegram_threads,
            telegram_messages,
            live_calls,
            transcript_segments,
            recording_call_ids,
        )
    finally:
        session.close()