gendesign/tradein-mvp/backend/app/api/v1/lead.py

"""Trade-in lead capture endpoint (issue #2376, sub-issue родителя #1971).

POST /api/v1/trade-in/lead — контактная заявка с результата оценки:
телефон + явное согласие на обработку персональных данных. Persist в
trade_in_leads.

Уведомление ответственному (#1971): после ответа клиенту (BackgroundTasks) лид
уходит сообщением в support-топик Telegram — тот же бот и топик, что у веб-чата
поддержки (`app.api.v1.support`). До этого заявки только ложились в таблицу, и
никто о них не узнавал. Успешная отправка проставляет `notified_at`; отказ
Telegram не трогает ни ответ, ни лид — остаётся лог и пустой `notified_at` как
видимый след недоставки. Телефона в сообщении НЕТ: копия в Telegram не стирается
механизмом удаления ПДн (`data_erasure.py`), поэтому туда идут только id —
телефон оператор берёт из trade_in_leads.

IDOR-фикс (security-audit): `estimate_id` раньше только проверялся на
СУЩЕСТВОВАНИЕ (`SELECT 1 ... WHERE id = ...`), без проверки владельца — любой
аутентифицированный пилот мог привязать свою заявку к чужой оценке (утечка через
последующий просмотр лида: чужой адрес/телефон/оценка в заявке, которую видит не
её владелец). Гвард переиспользует `_assert_estimate_access` из
`app.api.v1.trade_in` — тот же owner-or-admin подход, что и `GET /estimate/{id}`
(#690, `tests/test_estimate_idor.py`): 401 без `X-Authenticated-User`, 403 —
неизвестная роль, 404 — оценка не найдена ИЛИ принадлежит не этому пользователю
(существование чужой оценки не подтверждаем).
"""

from __future__ import annotations

import logging
import re
from datetime import UTC, datetime, timedelta
from typing import Annotated, Any, Literal
from uuid import UUID

from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.api.v1.support import _bot_configured
from app.api.v1.trade_in import _assert_estimate_access
from app.core.config import settings
from app.core.db import SessionLocal, get_db
from app.observability.metrics import LEADS
from app.services.tgbot.shared import get_telegram_client

logger = logging.getLogger(__name__)

router = APIRouter()

# Простая маска телефона (RU/международная): опциональный "+", цифры/пробелы/
# скобки/дефисы/точки, 5-32 символа. Полная нормализация в E.164 — вне scope MVP,
# см. #2376 DoD ("простая regex, не EmailStr-подобное").
_PHONE_PATTERN = r"^[+]?[\d\s().-]{5,32}$"
# Маска выше делает цифры ОПЦИОНАЛЬНЫМИ: "(()) -- .." её проходит (0 цифр).
# Поэтому дополнительно требуем правдоподобное число реальных цифр. RU-мобильный =
# 11 цифр; берём лениентный диапазон 10-15 (нац. номер без/с кодом страны).
_PHONE_MIN_DIGITS = 10
_PHONE_MAX_DIGITS = 15

# Версия политики обработки ПДн (152-ФЗ), под которую собрано согласие. Персистится
# per-row в trade_in_leads.consent_policy_version (migration 182) — до неё писалась
# только в audit-лог (#2497 TODO, теперь закрыт).
#
# Значение = дата утверждения политики (PRIVACY_APPROVAL в frontend/src/app/
# mera-public/content.ts: «приказом директора № 2 от 10 сентября 2026 г.» →
# "2026-09-10"), а не дата этого коммита — версия обязана указывать на редакцию
# ДОКУМЕНТА, на который согласие фактически ссылается (чекбокс теперь линкует
# именно на /mera-public/privacy). test_consent_text_frontend_sync.py проверяет
# это соответствие автоматически, так что рассинхронизация здесь падает в CI.
#
# Редакция № 2 от 10.09.2026: в политику добавлен раздел 9 «Файлы cookie и
# веб-аналитика» (на публичный сайт поставлены Яндекс.Метрика и GA4), прежний
# раздел «Контакты» стал десятым. Согласия, собранные до этой даты, ссылаются
# на редакцию № 1 и остаются с версией "2026-08-13" — в этом и смысл хранить
# версию per-row: снимок согласия должен указывать на тот документ, который
# человек видел, а не на текущий.
_CONSENT_POLICY_VERSION = "2026-09-10"

# Снимок точного текста согласия, который видит пользователь при отправке лида
# (ПЛОСКИЙ текст — без разметки ссылки на политику, которая в LeadForm.tsx рядом
# с этой фразой). Должен ДОСЛОВНО совпадать с чекбоксом в LeadForm.tsx (frontend/
# src/components/trade-in/v2/LeadForm.tsx) — если текст меняется, здесь нужно
# поднять _CONSENT_POLICY_VERSION И обновить этот снимок в одном PR, иначе новые
# строки будут нести устаревший snapshot под новой version-меткой.
_CONSENT_TEXT_SNAPSHOT = (
    "Согласен(-на) на обработку персональных данных в соответствии с "
    "Политикой обработки персональных данных"
)


class TradeInLeadInput(BaseModel):
    phone: str = Field(min_length=5, max_length=32, pattern=_PHONE_PATTERN)
    estimate_id: UUID | None = None
    consent: Literal[True]
    # "landing"-воронка недостижима: /api/v1/trade-in/lead закрыт rbac_guard
    # (main.py — путь не в _PUBLIC_PATHS => 401 без X-Authenticated-User), а
    # публичного лендинг-роута нет. Убрали мёртвый литерал, чтобы контракт не
    # обещал невозможную воронку (#2376). Вернуть, если появится public-роут.
    source: Literal["result"] = "result"

    @field_validator("phone")
    @classmethod
    def _phone_has_enough_digits(cls, value: str) -> str:
        digits = len(re.sub(r"\D", "", value))
        if not (_PHONE_MIN_DIGITS <= digits <= _PHONE_MAX_DIGITS):
            raise ValueError(f"phone must contain {_PHONE_MIN_DIGITS}-{_PHONE_MAX_DIGITS} digits")
        return value


async def _notify_new_lead(lead_id: str, estimate_id: UUID | None, username: str | None) -> None:
    """Сообщает о заявке в support-топик и отмечает `notified_at` (#1971).

    Идёт после ответа клиенту, поэтому ретраи клиента — штатные воркерные, без
    интерактивного бюджета (как `app.tasks.glitchtip_alert_retry`). Сессия своя:
    сессия запроса к этому моменту уже закрыта. Любой отказ — только в лог.
    """
    message = (
        "Новая заявка на трейд-ин\n"
        f"id: {lead_id}\n"
        f"пользователь: {username or '—'}\n"
        f"оценка: {estimate_id or 'без привязки'}\n"
        "Телефон — в trade_in_leads по id."
    )
    try:
        await get_telegram_client().send_message(
            chat_id=settings.telegram_support_chat_id,
            text=message,
            message_thread_id=settings.telegram_support_topic_id or None,
        )
    except Exception:
        logger.exception("trade_in_lead: уведомление не доставлено id=%s", lead_id)
        return

    db = SessionLocal()
    try:
        db.execute(
            text("UPDATE trade_in_leads SET notified_at = now() WHERE id = CAST(:id AS uuid)"),
            {"id": lead_id},
        )
        db.commit()
    except Exception:
        logger.exception("trade_in_lead: уведомление ушло, notified_at не записан id=%s", lead_id)
    finally:
        db.close()


@router.post("/lead")
async def create_trade_in_lead(
    payload: TradeInLeadInput,
    request: Request,
    db: Annotated[Session, Depends(get_db)],
    background_tasks: BackgroundTasks,
    x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
) -> dict[str, Any]:
    """Сохраняет лид (телефон + согласие) в trade_in_leads."""
    if payload.estimate_id is not None:
        estimate_row = db.execute(
            text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
            {"id": str(payload.estimate_id)},
        ).fetchone()
        if estimate_row is None:
            raise HTTPException(status_code=404, detail="estimate not found")
        # IDOR guard (security-audit, зеркалит #690): нельзя привязать лид к
        # чужой оценке. 404 и на "не найдено", и на "чужая" — не подтверждаем
        # существование чужого estimate_id.
        _assert_estimate_access(estimate_row.created_by, x_authenticated_user)

    user_agent = request.headers.get("user-agent")
    # 152-ФЗ audit trail: реальный клиентский IP из X-Forwarded-For (его ставит
    # фронтящий Caddy), fallback — прямой peer. Персистится per-row в
    # trade_in_leads.client_ip (migration 182), а не только в лог.
    client_ip = request.headers.get("x-forwarded-for")
    if client_ip:
        client_ip = client_ip.split(",")[0].strip()
    elif request.client is not None:
        client_ip = request.client.host

    # 152-ФЗ proof-of-consent: client_ip / consent_policy_version /
    # consent_text_snapshot теперь durable-колонки на trade_in_leads (migration 182,
    # ранее — только audit-лог, #2497 TODO). client_ip может быть None (нет
    # X-Forwarded-For и request.client) — колонка nullable, CAST(NULL AS inet) валиден.
    #
    # ЭТАП 4 B2C: expires_at (migration 231) — раньше лид хранился бессрочно
    # (никакого TTL вообще не было, в отличие от trade_in_estimates.expires_at).
    # Считаем на insert-time тем же паттерном, что estimator.py делает для
    # trade_in_estimates — retention-период вынесен в settings, не хардкод.
    expires_at = datetime.now(tz=UTC) + timedelta(days=settings.trade_in_lead_retention_days)

    row = (
        db.execute(
            text(
                """
            INSERT INTO trade_in_leads (
                estimate_id, phone, consent, source, user_agent,
                client_ip, consent_policy_version, consent_text_snapshot,
                expires_at
            )
            VALUES (
                CAST(:estimate_id AS uuid), :phone, :consent, :source, :user_agent,
                CAST(:client_ip AS inet), :consent_policy_version, :consent_text_snapshot,
                :expires_at
            )
            RETURNING CAST(id AS text), created_at
            """
            ),
            {
                "estimate_id": str(payload.estimate_id) if payload.estimate_id else None,
                "phone": payload.phone,
                "consent": payload.consent,
                "source": payload.source,
                "user_agent": user_agent,
                "client_ip": client_ip,
                "consent_policy_version": _CONSENT_POLICY_VERSION,
                "consent_text_snapshot": _CONSENT_TEXT_SNAPSHOT,
                "expires_at": expires_at,
            },
        )
        .mappings()
        .one()
    )

    db.commit()
    LEADS.inc()

    logger.info(
        "trade_in_lead saved id=%s estimate_id=%s source=%s ip=%s policy=%s",
        row["id"],
        payload.estimate_id,
        payload.source,
        client_ip,
        _CONSENT_POLICY_VERSION,
    )

    if _bot_configured():
        background_tasks.add_task(
            _notify_new_lead, row["id"], payload.estimate_id, x_authenticated_user
        )
    else:
        logger.warning(
            "trade_in_lead: бот не настроен, уведомление о id=%s не отправлено", row["id"]
        )

    return {
        "id": row["id"],
        "created_at": row["created_at"].isoformat(),
        "status": "received",
    }