gendesign/tradein-mvp/backend/app/services/yandex_session.py

"""Yandex.Nedvizhimost session cookie management — load/save/invalidate encrypted cookies.

⚠️ ОБОСНОВАНИЕ ЭТОГО МОДУЛЯ ОПРОВЕРГНУТО (#3192, замер 2026-08-28, разбор в тикете).
Модуль строился под гипотезу «авторизованная сессия раскрывает контакты продавца».
Повторный замер на ПРОД-транспорте (curl_cffi + прокси, тот же путь, что у
yandex_detail_backfill, а не сайдкар) её опроверг:

  - offerCard.card.author — целевая карточка — не несёт phones/phoneNumbers ни в
    одном из 12 проверенных объявлений (6 AGENCY, 6 DEVELOPER), одинаково с куками
    и без; есть только encryptedPhones (1 токен) и redirectPhones;
  - phoneNumbers во всём INITIAL_STATE встречается ТОЛЬКО под
    offerCard.visitedOffers[*].author — это история просмотров НАШЕЙ учётки;
    анонимно она пуста, с куками содержит 9-10 записей;
  - первоначальный вывод «0 → 3,4,5,6 телефонов» считал именно рост этой истории:
    +1 запись на каждый фетч, а не свойство карточек;
  - authorStats.phones (коммутатор застройщика) отдаётся анонимно — тот же номер
    в обеих ветках.

То есть измеримой разницы по данным объявления авторизация не даёт. Хранилище
работает и покрыто тестами, но пока не имеет доказанного применения и НЕ
заполнено. Судьба модуля — за владельцем (#3192, label needs-human); удалять
таблицу самовольно нельзя. Ниже — исходное описание устройства, оно актуально.

Cookies хранятся зашифрованно (pgp_sym_encrypt) в yandex_session_cookies —
зеркалит app.services.cian_session / app.services.domclick_session, но:

  - ключ записи — account_label (text), НЕ числовой id: у Яндекса нет внутреннего
    числового id аккаунта, доступного из cookie-дампа;
  - хранится ПОЛНЫЙ CDP cookie-объект (name/value/domain/path/httpOnly/secure/
    expires), а не плоский dict[name, value] как у Cian/DomClick — инъекция кук в
    browser context (следующая часть #3192) требует domain/path, не только value;
  - НЕТ allowlist-фильтра на сохранение (см. filter_analytics_cookies docstring) —
    Cian/DomClick фильтруют payload по списку известных cookie-имён перед
    сохранением, здесь так делать нельзя: замеры делались на ПОЛНОМ наборе из
    44 cookies, и какая из них за что отвечает — неизвестно.

MVP: без verify_session (аналогично domclick_session) — верификация Yandex-сессии
требует реального browser-фетча, не простого curl_cffi-запроса; вне рамок этой
итерации.
"""

from __future__ import annotations

import json
import logging
from datetime import datetime
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings

logger = logging.getLogger(__name__)

# За сколько дней до протухания кук предупреждать (зеркалит #2658/#2674). Обновление
# кук — РУЧНАЯ операция (залить дамп через админку), человеку нужен запас: сигнал по
# факту протухания приходит, когда сбор уже встал. save_session ставит ttl 30 дней.
COOKIE_EXPIRY_WARN_DAYS = 5

# Cookies, реально наблюдаемые в авторизованной Yandex-сессии (замер #3192,
# 2026-08-28), все HttpOnly кроме отмеченных.
#
# ВАЖНО: это НЕ allowlist для фильтрации того, что сохранять (в отличие от
# CIAN_REQUIRED_COOKIES / DOMCLICK_REQUIRED_COOKIES) — используется ТОЛЬКО в
# has_auth_marker() для проверки "дамп вообще похож на авторизованную сессию".
# Замеры #3192 делались на ПОЛНОМ наборе из 44 cookies, за что отвечает каждая —
# неизвестно, поэтому save_session хранит весь дамп целиком (за вычетом аналитики),
# а не только эти имена. НЕ сужай набор, который реально сохраняется, до этого
# списка.
YANDEX_CRITICAL_COOKIES: set[str] = {
    "Session_id",
    "sessionid2",
    "sessar",
    "i",
    "L",
    "sessguard",
    "yandexuid",  # не HttpOnly
    "yandex_login",  # не HttpOnly
}

# Ключевая auth-cookie, которую ставит Яндекс.Паспорт при логине — её отсутствие
# в дампе означает "это анонимная сессия", независимо от того, сколько прочих
# cookies присутствует.
AUTH_MARKER_COOKIE = "Session_id"

# Префиксы имён cookies, которые ВСЕГДА аналитика (Яндекс.Метрика / РСЯ /
# анти-спам виджет), а не auth-состояние — единственное, что выкидываем из
# дампа перед сохранением. Всё остальное сохраняется как пришло.
ANALYTICS_COOKIE_PREFIXES: tuple[str, ...] = ("_ym_", "yabs-", "_yasc")


def filter_analytics_cookies(cookies: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Drop только заведомую аналитику (_ym_*, yabs-*, _yasc*) — остальное как есть.

    В отличие от Cian/DomClick здесь нет allowlist на сохранение: замеры #3192
    делались на ПОЛНОМ наборе из 44 cookies, значимость конкретной cookie
    неизвестна. Сужать этот фильтр до allowlist известных "критичных" имён —
    значит потерять часть замера у следующего инженера, который решит его
    "оптимизировать". YANDEX_CRITICAL_COOKIES используется только в
    has_auth_marker(), не здесь.
    """
    return [c for c in cookies if not str(c.get("name", "")).startswith(ANALYTICS_COOKIE_PREFIXES)]


def has_auth_marker(cookies: list[dict[str, Any]]) -> bool:
    """True если в дампе присутствует ключевая auth-cookie (Session_id).

    Используется как дешёвая пре-проверка "это вообще похоже на авторизованный
    дамп", НЕ полноценная верификация (см. модульный docstring — verify_session
    для Yandex вне рамок этой итерации).
    """
    names = {c.get("name") for c in cookies}
    return AUTH_MARKER_COOKIE in names


def save_session(
    db: Session,
    account_label: str,
    cookies: list[dict[str, Any]],
    ttl_days: int = 30,
) -> None:
    """Encrypt cookies via pgp_sym_encrypt and UPSERT into yandex_session_cookies.

    cookies — список полных CDP cookie-объектов (name/value/domain/path/httpOnly/
    secure/expires), НЕ плоский dict[name, value] как у Cian/DomClick — инъекция
    кук в browser context требует domain/path.

    Uses settings.cookie_encryption_key as the encryption secret.
    Никогда не логирует сырые значения cookies.
    """
    cookies_json = json.dumps(cookies)
    db.execute(
        text("""
            INSERT INTO yandex_session_cookies (
                account_label,
                cookies_encrypted,
                expires_at_estimate,
                uploaded_at
            ) VALUES (
                :label,
                pgp_sym_encrypt(:cookies_json, :key),
                NOW() + (CAST(:ttl_days AS int) || ' days')::interval,
                NOW()
            )
            ON CONFLICT (account_label) DO UPDATE SET
                cookies_encrypted    = EXCLUDED.cookies_encrypted,
                expires_at_estimate  = EXCLUDED.expires_at_estimate,
                uploaded_at          = NOW(),
                last_invalid_at      = NULL
        """),
        {
            "label": account_label,
            "cookies_json": cookies_json,
            "key": settings.cookie_encryption_key,
            "ttl_days": ttl_days,
        },
    )
    db.commit()
    logger.info(
        "Yandex cookies saved for label=%s (count=%d, ttl=%d days)",
        account_label,
        len(cookies),
        ttl_days,
    )


def load_session(db: Session) -> list[dict[str, Any]] | None:
    """Load most-recently-uploaded valid Yandex cookies (decrypt).

    Returns список CDP cookie-объектов или None если нет валидной session.
    Выбирает только записи где expires_at_estimate > NOW() и сессия не была
    инвалидирована после последнего upload'а.
    """
    row = (
        db.execute(
            text("""
            SELECT
                account_label,
                pgp_sym_decrypt(cookies_encrypted, :key)::text AS cookies_json,
                expires_at_estimate
            FROM yandex_session_cookies
            WHERE expires_at_estimate > NOW()
              AND (last_invalid_at IS NULL OR last_invalid_at < uploaded_at)
            ORDER BY uploaded_at DESC
            LIMIT 1
        """),
            {"key": settings.cookie_encryption_key},
        )
        .mappings()
        .first()
    )

    if row is None:
        logger.warning("No valid Yandex session cookies in DB")
        return None

    cookies: list[dict[str, Any]] = json.loads(row["cookies_json"])

    # Обновляем last_used_at — не критично, игнорируем ошибки.
    try:
        db.execute(
            text(
                "UPDATE yandex_session_cookies SET last_used_at = NOW()"
                " WHERE account_label = :label"
            ),
            {"label": row["account_label"]},
        )
        db.commit()
    except Exception as exc:
        # Ошибка на execute() оставляет сессию в aborted-транзакции (psycopg/PG:
        # "current transaction is aborted") — если db переживёт этот вызов (caller
        # держит ту же сессию дальше, напр. admin.py session-status делает
        # session_expires_at(db, ...) сразу следом), последующие запросы иначе все
        # падали бы с той же ошибкой (зеркалит app.services.proxy_egress паттерн).
        try:
            db.rollback()
        except Exception:
            logger.warning("Failed to rollback after last_used_at update failure", exc_info=True)
        logger.warning("Failed to update last_used_at for label=%s: %s", row["account_label"], exc)

    logger.info(
        "Yandex cookies loaded for label=%s (count=%d)",
        row["account_label"],
        len(cookies),
    )
    return cookies


def session_expires_at(db: Session, *, valid_only: bool = False) -> datetime | None:
    """Когда протухают самые свежезагруженные куки (зеркалит cian_session #2658).

    `load_session` отбирает только ещё валидные записи (expires_at_estimate > NOW()) и на
    протухших отдаёт None — вызывающий не мог отличить «кук никогда не загружали» от
    «протухли позавчера» и не мог предупредить ЗАРАНЕЕ.

    valid_only=False (диагностика после None от load_session) — свежайшая запись любая:
    валидных по определению нет, нужен именно срок протухшей. valid_only=True — та же
    запись, которую взял бы load_session: для предупреждения «скоро протухнут» нужен срок
    ИМЕННО используемых кук, иначе при нескольких аккаунтах посчитаем по чужой строке.
    """
    row = db.execute(
        text(
            """
            SELECT expires_at_estimate FROM yandex_session_cookies
            WHERE NOT CAST(:valid_only AS boolean)
               OR (expires_at_estimate > NOW()
                   AND (last_invalid_at IS NULL OR last_invalid_at < uploaded_at))
            ORDER BY uploaded_at DESC
            LIMIT 1
            """
        ),
        {"valid_only": valid_only},
    ).first()
    if row is None:
        return None
    expires_at: datetime | None = row[0]
    return expires_at


def mark_session_invalid(db: Session, account_label: str) -> None:
    """Flag session как expired/invalid (например после логаута во время scrape)."""
    db.execute(
        text(
            "UPDATE yandex_session_cookies SET last_invalid_at = NOW() WHERE account_label = :label"
        ),
        {"label": account_label},
    )
    db.commit()
    logger.warning("Yandex session marked invalid for label=%s", account_label)