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

"""Cian session cookie management — load/save/verify encrypted cookies.

Used by cian_valuation.py (Stage 7) to authenticate Valuation Calculator API.
Cookies stored encrypted (pgp_sym_encrypt) in cian_session_cookies table.
"""
from __future__ import annotations

import json
import logging
from typing import Any

import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services.scrapers.cian_state_parser import extract_state

logger = logging.getLogger(__name__)

# Cookies критичные для Cian auth — фильтр перед сохранением.
# Список обновлён по реальному DevTools-дампу из logged-in сессии cian.ru (2026-05-23).
# Старые записи оставлены как fallback (backward compat).
CIAN_REQUIRED_COOKIES: set[str] = {
    # --- Cian auth & session (critical) ---
    "DMIR_AUTH",                   # Cian auth token (httpOnly) — основной auth-сигнал
    "_CIAN_GK",                    # Cian session GUID

    # --- Yandex Metrika (обычно присутствуют, не critical) ---
    "_ym_uid",
    "_ym_d",
    "_ym_isad",
    "_ym_visorc",
    "_yasc",                       # Yandex anti-spam

    # --- Cian UX state ---
    "uxfb_card_satisfaction",

    # --- Cian session IDs (fallback / legacy deployments) ---
    "_cian_visitor_session_id",
    "_cian_app_session_id",
    "_cian_uid",

    # --- Misc (legacy / optional) ---
    "_ga",                         # Google Analytics
    "tlsr_id",                     # legacy fingerprint
    "cf_clearance",                # Cloudflare bot check
    "session-id",                  # generic session cookie
    "anti_bot",
    "csrftoken",
}

_VALUATION_TEST_URL = (
    "https://www.cian.ru/kalkulator-nedvizhimosti/"
    "?address=%D0%A1%D0%B2%D0%B5%D1%80%D0%B4%D0%BB%D0%BE%D0%B2%D1%81%D0%BA%D0%B0%D1%8F"
    "%20%D0%BE%D0%B1%D0%BB%2C%20%D0%95%D0%BA%D0%B0%D1%82%D0%B5%D1%80%D0%B8%D0%BD%D0%B1%D1%83%D1%80%D0%B3"
    "&totalArea=40&roomsCount=1&valuationType=sale"
    "&floor%5B0%5D=floorOther&repairType%5B0%5D=repairTypeCosmetic"
)

_MFE_VALUATION = "valuation-for-agent-frontend"


async def verify_session(cookies: dict[str, str]) -> dict[str, Any] | None:
    """Hit Cian Valuation Calculator with cookies — return parsed state if authenticated.

    Returns state dict containing user.isAuthenticated + userId, or None if
    cookies are invalid/expired or state extraction fails.
    Никогда не логирует сырые значения cookies.
    """
    try:
        async with httpx.AsyncClient(
            cookies=cookies,
            timeout=20.0,
            follow_redirects=True,
        ) as client:
            resp = await client.get(_VALUATION_TEST_URL)
            resp.raise_for_status()
            state = extract_state(resp.text, mfe=_MFE_VALUATION, key="initialState")
            if state is None:
                logger.warning(
                    "Cian cookies verify: state extraction failed (mfe=%s)", _MFE_VALUATION
                )
                return None
            user = state.get("user", {})
            if not user.get("isAuthenticated"):
                logger.warning("Cian cookies verify: user.isAuthenticated=false")
                return None
            logger.info("Cian cookies verified — userId=%s", user.get("userId"))
            return state
    except httpx.HTTPStatusError as exc:
        logger.warning(
            "Cian cookies verify HTTP error: %s %s",
            exc.response.status_code,
            exc.request.url,
        )
        return None
    except Exception as exc:
        logger.warning("Cian cookies verify failed: %s", exc)
        return None


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

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


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

    Returns dict[cookie_name, cookie_value] или None если нет валидной session.
    Выбирает только записи где expires_at_estimate > NOW() и сессия не
    была инвалидирована после последнего upload'а.
    """
    row = db.execute(
        text("""
            SELECT
                account_user_id,
                pgp_sym_decrypt(cookies_encrypted, :key)::text AS cookies_json,
                expires_at_estimate
            FROM cian_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 Cian session cookies in DB")
        return None

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

    # Обновляем last_used_at — не критично, игнорируем ошибки.
    try:
        db.execute(
            text(
                "UPDATE cian_session_cookies SET last_used_at = NOW()"
                " WHERE account_user_id = CAST(:uid AS bigint)"
            ),
            {"uid": row["account_user_id"]},
        )
        db.commit()
    except Exception as exc:
        logger.warning(
            "Failed to update last_used_at for userId=%s: %s", row["account_user_id"], exc
        )

    logger.info(
        "Cian cookies loaded for userId=%s (count=%d)",
        row["account_user_id"],
        len(cookies),
    )
    return cookies


def mark_session_invalid(db: Session, account_user_id: int) -> None:
    """Flag session как expired/invalid (например после 401 во время scrape)."""
    db.execute(
        text(
            "UPDATE cian_session_cookies SET last_invalid_at = NOW()"
            " WHERE account_user_id = CAST(:uid AS bigint)"
        ),
        {"uid": account_user_id},
    )
    db.commit()
    logger.warning("Cian session marked invalid for userId=%s", account_user_id)