diff --git a/app/cabinet/routes/branding.py b/app/cabinet/routes/branding.py index 14daae29..7cfaefc4 100644 --- a/app/cabinet/routes/branding.py +++ b/app/cabinet/routes/branding.py @@ -17,7 +17,7 @@ from app.config import settings from app.database.crud.system_setting import get_setting_value from app.database.models import SystemSetting, User -from ..dependencies import get_cabinet_db, require_permission +from ..dependencies import get_cabinet_db, get_current_cabinet_user, require_permission logger = structlog.get_logger(__name__) @@ -291,12 +291,24 @@ class GiftEnabledUpdate(BaseModel): enabled: bool +class OfflineConvGoal(BaseModel): + """Yandex Metrika offline conversion goal descriptor.""" + + name: str + event_id: str + dedup: str + + class AnalyticsCountersResponse(BaseModel): """Analytics counter settings.""" yandex_metrika_id: str = '' google_ads_id: str = '' google_ads_label: str = '' + offline_conv_enabled: bool = False + offline_conv_counter_id: str = '' + offline_conv_measurement_secret_masked: str = '' + offline_conv_goals: list[OfflineConvGoal] = [] class AnalyticsCountersUpdate(BaseModel): @@ -924,10 +936,27 @@ async def get_analytics_counters( google_id = await get_setting_value(db, GOOGLE_ADS_ID_KEY) or '' google_label = await get_setting_value(db, GOOGLE_ADS_LABEL_KEY) or '' + # Yandex Metrika offline conversions snapshot from Settings + oc_enabled = bool(getattr(settings, 'YANDEX_OFFLINE_CONV_ENABLED', False)) + oc_counter = str(getattr(settings, 'YANDEX_OFFLINE_CONV_COUNTER_ID', '') or '') + oc_secret = str(getattr(settings, 'YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET', '') or '') + oc_secret_masked = ('*' * 8 + oc_secret[-4:]) if len(oc_secret) > 4 else ('***' if oc_secret else '') + oc_goals: list[OfflineConvGoal] = [] + if oc_enabled: + oc_goals = [ + OfflineConvGoal(name='Registration', event_id='registration', dedup='user_id'), + OfflineConvGoal(name='Trial', event_id='trial-add', dedup='user_id'), + OfflineConvGoal(name='Purchase', event_id='purchase', dedup='order_id'), + ] + return AnalyticsCountersResponse( yandex_metrika_id=yandex_id, google_ads_id=google_id, google_ads_label=google_label, + offline_conv_enabled=oc_enabled, + offline_conv_counter_id=oc_counter, + offline_conv_measurement_secret_masked=oc_secret_masked, + offline_conv_goals=oc_goals, ) @@ -966,13 +995,56 @@ async def update_analytics_counters( google_id = await get_setting_value(db, GOOGLE_ADS_ID_KEY) or '' google_label = await get_setting_value(db, GOOGLE_ADS_LABEL_KEY) or '' + oc_enabled = bool(getattr(settings, 'YANDEX_OFFLINE_CONV_ENABLED', False)) + oc_counter = str(getattr(settings, 'YANDEX_OFFLINE_CONV_COUNTER_ID', '') or '') + oc_secret = str(getattr(settings, 'YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET', '') or '') + oc_secret_masked = ('*' * 8 + oc_secret[-4:]) if len(oc_secret) > 4 else ('***' if oc_secret else '') + oc_goals: list[OfflineConvGoal] = [] + if oc_enabled: + oc_goals = [ + OfflineConvGoal(name='Registration', event_id='registration', dedup='user_id'), + OfflineConvGoal(name='Trial', event_id='trial-add', dedup='user_id'), + OfflineConvGoal(name='Purchase', event_id='purchase', dedup='order_id'), + ] + return AnalyticsCountersResponse( yandex_metrika_id=yandex_id, google_ads_id=google_id, google_ads_label=google_label, + offline_conv_enabled=oc_enabled, + offline_conv_counter_id=oc_counter, + offline_conv_measurement_secret_masked=oc_secret_masked, + offline_conv_goals=oc_goals, ) +# ============ Yandex CID Sync ============ + + +class YandexCidRequest(BaseModel): + cid: str = Field(max_length=128, pattern=r'^[A-Za-z0-9._:-]{4,128}$') + + +@router.post('/analytics/yandex-cid', status_code=204) +async def store_yandex_cid( + body: YandexCidRequest, + user: User = Depends(get_current_cabinet_user), + db: AsyncSession = Depends(get_cabinet_db), +): + """Store Yandex Metrika ClientID for the authenticated cabinet user.""" + try: + from app.services import yandex_offline_conv_service as yandex_conv + + await yandex_conv.store_cid(db, user.id, body.cid, source='cabinet') + await db.commit() + except Exception as exc: + logger.warning('Failed to store yandex_cid', user_id=user.id, exc=str(exc)) + try: + await db.rollback() + except Exception: + pass + + # ============ Lite Mode Routes ============ diff --git a/app/config.py b/app/config.py index 174b7d88..b0b87c85 100644 --- a/app/config.py +++ b/app/config.py @@ -567,6 +567,21 @@ class Settings(BaseSettings): KASSA_AI_SBERPAY_ENABLED: bool = False # SberPay — payment_system_id=43 KASSA_AI_SBERPAY_DISPLAY_NAME: str = 'SberPay (KassaAI)' + # ── Yandex Metrika offline conversions (server → mc.yandex.ru/collect) ── + YANDEX_OFFLINE_CONV_ENABLED: bool = False + YANDEX_OFFLINE_CONV_COUNTER_ID: str = '' + YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET: str = '' + YANDEX_OFFLINE_CONV_START_PREFIX: str = 'utm_ya_' + YANDEX_OFFLINE_CONV_DL: str = '' + YANDEX_OFFLINE_CONV_DT: str = '' + YANDEX_OFFLINE_CONV_CURRENCY: str = 'RUB' + + # ── S2S Postback (server-to-server affiliate notifications) ── + S2S_POSTBACK_ENABLED: bool = False + S2S_POSTBACK_REGISTRATION_URL: str = '' + S2S_POSTBACK_TRIAL_URL: str = '' + S2S_POSTBACK_PURCHASE_URL: str = '' + # RioPay (api.riopay.online) v2.0.1 RIOPAY_ENABLED: bool = False RIOPAY_API_TOKEN: str | None = None # x-api-token header diff --git a/app/database/crud/yandex_client_id.py b/app/database/crud/yandex_client_id.py new file mode 100644 index 00000000..a5f5d385 --- /dev/null +++ b/app/database/crud/yandex_client_id.py @@ -0,0 +1,108 @@ +"""CRUD operations for yandex_client_id_map table.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import structlog +from sqlalchemy import select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database.models import YandexClientIdMap + + +logger = structlog.get_logger(__name__) + + +async def upsert_cid( + db: AsyncSession, + user_id: int, + cid: str, + source: str = 'web', + counter_id: str | None = None, + subid: str | None = None, +) -> YandexClientIdMap: + """Insert or update Yandex ClientID for a user (race-safe via ON CONFLICT).""" + now = datetime.now(UTC) + values = { + 'yandex_cid': cid, + 'source': source, + 'updated_at': now, + } + if counter_id: + values['counter_id'] = counter_id + if subid: + values['subid'] = subid + + stmt = ( + pg_insert(YandexClientIdMap) + .values(user_id=user_id, yandex_cid=cid, source=source, counter_id=counter_id, subid=subid) + .on_conflict_do_update(index_elements=['user_id'], set_=values) + .returning(YandexClientIdMap) + ) + + result = await db.execute(stmt) + await db.flush() + return result.scalar_one() + + +async def get_cid(db: AsyncSession, user_id: int) -> YandexClientIdMap | None: + """Get Yandex ClientID mapping for a user.""" + result = await db.execute(select(YandexClientIdMap).where(YandexClientIdMap.user_id == user_id)) + return result.scalar_one_or_none() + + +async def mark_registration_sent(db: AsyncSession, user_id: int) -> None: + """Mark registration event as sent for a user.""" + await db.execute( + update(YandexClientIdMap) + .where(YandexClientIdMap.user_id == user_id) + .values(registration_sent=True, updated_at=datetime.now(UTC)) + ) + await db.flush() + + +async def mark_trial_sent(db: AsyncSession, user_id: int) -> None: + """Mark trial event as sent for a user.""" + await db.execute( + update(YandexClientIdMap) + .where(YandexClientIdMap.user_id == user_id) + .values(trial_sent=True, updated_at=datetime.now(UTC)) + ) + await db.flush() + + +async def upsert_subid( + db: AsyncSession, + user_id: int, + subid: str, + source: str = 'web', +) -> None: + """Save subid for a user. Updates existing record or creates with placeholder CID.""" + if not subid or len(subid) > 255: + return + now = datetime.now(UTC) + # Try update first (don't create empty CID records) + result = await db.execute( + update(YandexClientIdMap).where(YandexClientIdMap.user_id == user_id).values(subid=subid, updated_at=now) + ) + if result.rowcount == 0: + # No existing record — create with placeholder + stmt = ( + pg_insert(YandexClientIdMap) + .values(user_id=user_id, yandex_cid='_subid_only', source=source, subid=subid) + .on_conflict_do_update( + index_elements=['user_id'], + set_={'subid': subid, 'updated_at': now}, + ) + ) + await db.execute(stmt) + await db.flush() + logger.info('Subid saved', user_id=user_id, subid=subid, source=source) + + +async def get_subid(db: AsyncSession, user_id: int) -> str | None: + """Get subid for a user.""" + result = await db.execute(select(YandexClientIdMap.subid).where(YandexClientIdMap.user_id == user_id)) + return result.scalar_one_or_none() diff --git a/app/database/models.py b/app/database/models.py index 4f668573..eec6efed 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -3528,6 +3528,10 @@ class GuestPurchase(Base): retry_count = Column(Integer, nullable=False, default=0, server_default='0') receipt_uuid = Column(String(255), nullable=True, index=True) receipt_created_at = Column(AwareDateTime(), nullable=True) + # Yandex Metrika offline conversions: client identifier + traffic source tags + yandex_cid = Column(String(128), nullable=True) + subid = Column(String(255), nullable=True) + referrer = Column(String(500), nullable=True) landing = relationship('LandingPage', back_populates='guest_purchases', lazy='selectin') tariff = relationship('Tariff', lazy='selectin') @@ -3609,3 +3613,26 @@ class NewsTag(Base): def __repr__(self) -> str: return f"" + + +class YandexClientIdMap(Base): + """Yandex Metrika client identifier captured per user. + + Stores the mapping user_id -> yandex_cid so we can fire offline + conversion events to mc.yandex.ru with the right CID even after + the user leaves the landing/web flow. The ``subid`` column carries + a pass-through traffic-source identifier for S2S postbacks. + """ + + __tablename__ = 'yandex_client_id_map' + + id = Column(Integer, primary_key=True, autoincrement=True) + user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'), unique=True, nullable=False) + yandex_cid = Column(String(128), nullable=False) + source = Column(String(20), nullable=False, default='web', server_default='web') + counter_id = Column(String(32), nullable=True) + registration_sent = Column(Boolean, default=False, server_default=text('false'), nullable=False) + trial_sent = Column(Boolean, default=False, server_default=text('false'), nullable=False) + subid = Column(String(255), nullable=True) + created_at = Column(AwareDateTime(), server_default=func.now()) + updated_at = Column(AwareDateTime(), server_default=func.now(), onupdate=func.now()) diff --git a/app/services/s2s_postback_service.py b/app/services/s2s_postback_service.py new file mode 100644 index 00000000..9d9c5b93 --- /dev/null +++ b/app/services/s2s_postback_service.py @@ -0,0 +1,92 @@ +"""S2S Postback Service — sends server-to-server postbacks on events.""" + +import structlog + +from app.config import settings + + +logger = structlog.get_logger(__name__) + +try: + import httpx +except ImportError: + httpx = None + + +def _is_enabled() -> bool: + return getattr(settings, 'S2S_POSTBACK_ENABLED', False) and httpx is not None + + +def _get_url(event: str) -> str | None: + """Get postback URL template for event type.""" + mapping = { + 'registration': getattr(settings, 'S2S_POSTBACK_REGISTRATION_URL', ''), + 'trial': getattr(settings, 'S2S_POSTBACK_TRIAL_URL', ''), + 'purchase': getattr(settings, 'S2S_POSTBACK_PURCHASE_URL', ''), + } + url = mapping.get(event, '') + return url or None + + +async def send_postback( + event: str, + subid: str, + amount: float | None = None, + user_id: int | None = None, +) -> bool: + """Send S2S postback for an event. + + Args: + event: 'registration', 'trial', or 'purchase' + subid: tracking subid from URL + amount: purchase amount in rubles (for purchase event) + user_id: internal user ID for logging + + Returns: + True if sent successfully + """ + if not _is_enabled(): + return False + + if not subid: + return False + + url_template = _get_url(event) + if not url_template: + logger.debug('S2S postback URL not configured', event=event) + return False + + # Replace placeholders (URL-encode subid to prevent injection) + from urllib.parse import quote + + url = url_template.replace('{subid}', quote(subid, safe='')) + url = url.replace('{event}', event) + if amount is not None: + url = url.replace('{amount}', str(round(amount, 2))) + else: + url = url.replace('{amount}', '0') + + url = url.replace('{user_id}', str(user_id) if user_id is not None else '0') + + try: + async with httpx.AsyncClient(timeout=10) as client: + response = await client.get(url) + logger.info( + 'S2S postback sent', + event=event, + subid=subid, + amount=amount, + user_id=user_id, + status_code=response.status_code, + url=url[:100], + ) + return response.status_code < 400 + except Exception as e: + logger.error( + 'S2S postback failed', + event=event, + subid=subid, + error=str(e), + url=url[:100], + ) + return False diff --git a/app/services/yandex_offline_conv_service.py b/app/services/yandex_offline_conv_service.py new file mode 100644 index 00000000..86eb1969 --- /dev/null +++ b/app/services/yandex_offline_conv_service.py @@ -0,0 +1,352 @@ +"""Yandex.Metrika offline conversions service. + +Sends events (registration, trial-add, purchase) to mc.yandex.ru/collect +using the Measurement Protocol. No pageview needed — user has active +Metrika session from the site. yclid is passed via landing page URL, +Metrika matches it automatically. +""" + +from __future__ import annotations + +import asyncio +import re +import time + +import httpx +import structlog +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.database.crud.yandex_client_id import ( + get_cid, + mark_registration_sent, + mark_trial_sent, + upsert_cid, +) +from app.database.database import AsyncSessionLocal + + +logger = structlog.get_logger(__name__) + +COLLECT_URL = 'https://mc.yandex.ru/collect' +TIMEOUT = 10.0 +MAX_RETRIES = 3 +RETRY_DELAY = 1.0 + +_CID_RE = re.compile(r'^[A-Za-z0-9._:-]{4,128}$') +_http_client: httpx.AsyncClient | None = None + + +def _get_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None or _http_client.is_closed: + _http_client = httpx.AsyncClient(timeout=TIMEOUT) + return _http_client + + +def _is_enabled() -> bool: + return bool( + settings.YANDEX_OFFLINE_CONV_ENABLED + and settings.YANDEX_OFFLINE_CONV_COUNTER_ID + and settings.YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET + ) + + +def _normalize_cid(cid: str | None) -> str | None: + if not isinstance(cid, str): + return None + cid = cid.strip() + if not cid or not _CID_RE.match(cid): + return None + return cid + + +def _mask_cid(cid: str) -> str: + if len(cid) <= 4: + return '****' + return '*' * (len(cid) - 4) + cid[-4:] + + +def _base_payload(cid: str) -> dict[str, str]: + return { + 'tid': settings.YANDEX_OFFLINE_CONV_COUNTER_ID, + 'cid': cid, + 'ms': settings.YANDEX_OFFLINE_CONV_MEASUREMENT_SECRET, + } + + +def _pageview_payload(cid: str) -> dict[str, str]: + payload = _base_payload(cid) + payload.update( + { + 't': 'pageview', + 'dl': settings.YANDEX_OFFLINE_CONV_DL or 'https://web.mtrxvps.ru', + 'dt': settings.YANDEX_OFFLINE_CONV_DT or 'Matrixxx VPN', + } + ) + return payload + + +def _event_payload(cid: str, event_action: str) -> dict[str, str]: + payload = _base_payload(cid) + payload.update( + { + 't': 'event', + 'ea': event_action, + } + ) + return payload + + +def _ecommerce_purchase_payload( + cid: str, + amount_rubles: float, + order_id: str = '', + product_name: str = '', + product_category: str = '', +) -> dict[str, str]: + """Build ecommerce:purchase payload for Metrika Measurement Protocol.""" + + service_name = ( + getattr(settings, 'YANDEX_OFFLINE_CONV_DT', '') + or getattr(settings, 'PAYMENT_SERVICE_NAME', '') + or 'Subscription' + ) + currency = getattr(settings, 'YANDEX_OFFLINE_CONV_CURRENCY', '') or 'RUB' + payload = _base_payload(cid) + payload.update( + { + 't': 'event', + 'ea': 'purchase', + 'pa': 'purchase', + 'ti': order_id or str(int(time.time())), + 'tr': str(amount_rubles), + 'cu': currency, + 'ev': str(amount_rubles), + 'pr1id': 'subscription', + 'pr1nm': product_name or service_name, + 'pr1ca': product_category or 'subscription', + 'pr1pr': str(amount_rubles), + 'pr1qt': '1', + } + ) + return payload + + +async def _post_collect(payload: dict[str, str], kind: str, cid: str) -> bool: + """POST to mc.yandex.ru/collect with retries. Returns True on success.""" + masked = _mask_cid(cid) + for attempt in range(1, MAX_RETRIES + 1): + try: + client = _get_client() + resp = await client.post(COLLECT_URL, data=payload) + + if 200 <= resp.status_code < 300: + logger.info('collect sent', kind=kind, cid=masked, status=resp.status_code) + return True + + if 500 <= resp.status_code < 600 and attempt < MAX_RETRIES: + logger.warning( + 'collect server error', + kind=kind, + attempt=attempt, + max=MAX_RETRIES, + cid=masked, + status=resp.status_code, + ) + await asyncio.sleep(RETRY_DELAY) + continue + + logger.error('collect rejected', kind=kind, cid=masked, status=resp.status_code, body=resp.text[:200]) + return False + + except Exception as exc: + logger.warning( + 'collect request error', kind=kind, attempt=attempt, max=MAX_RETRIES, cid=masked, error=str(exc) + ) + if attempt < MAX_RETRIES: + await asyncio.sleep(RETRY_DELAY) + continue + return False + + return False + + +async def _send_event(cid: str, event_action: str) -> bool: + """Send event directly — no pageview needed, user has active Metrika session.""" + return await _post_collect(_event_payload(cid, event_action), event_action, cid) + + +# --- Background task helpers --- + +_background_tasks: set[asyncio.Task] = set() + + +def _task_done(task): + """Log errors from background conversion tasks.""" + _background_tasks.discard(task) + if task.cancelled(): + return + exc = task.exception() + if exc: + logger.error('YandexOfflineConv background task failed', error=str(exc)) + + +def spawn_bg(coro) -> None: + """Spawn a background Yandex conversion task with proper reference tracking. + + Checks _is_enabled() early so callers don't need to. + """ + if not _is_enabled(): + # Close the coroutine to avoid RuntimeWarning + coro.close() + return + task = asyncio.create_task(coro) + _background_tasks.add(task) + task.add_done_callback(_task_done) + + +async def _fire_bg(event_name: str, event_fn, user_id: int, **kwargs) -> None: + """Generic background wrapper: opens a session, calls event_fn, logs errors.""" + try: + async with AsyncSessionLocal() as db: + await event_fn(db, user_id, **kwargs) + except Exception as exc: + logger.warning('YandexOfflineConv background event failed', event=event_name, user_id=user_id, error=str(exc)) + + +async def fire_registration_bg(user_id: int) -> None: + """Fire registration event in background with its own DB session.""" + await _fire_bg('registration', on_registration, user_id) + + +async def fire_trial_bg(user_id: int) -> None: + """Fire trial event in background with its own DB session.""" + await _fire_bg('trial', on_trial, user_id) + + +async def fire_purchase_bg(user_id: int, amount_kopeks: int) -> None: + """Fire purchase event in background with its own DB session.""" + await _fire_bg('purchase', on_purchase, user_id, amount_kopeks=amount_kopeks) + + +# --- Public API --- +async def store_cid( + db: AsyncSession, + user_id: int, + cid: str | None, + source: str = 'web', +) -> bool: + """Store Yandex ClientID for a user. Returns True if stored.""" + normalized = _normalize_cid(cid) + if not normalized: + return False + + try: + await upsert_cid(db, user_id, normalized, source=source, counter_id=settings.YANDEX_OFFLINE_CONV_COUNTER_ID) + logger.info('stored CID', user_id=user_id, source=source) + return True + except Exception as exc: + logger.error('failed to store CID', user_id=user_id, error=str(exc)) + return False + + +async def store_cid_and_fire_registration( + user_id: int, + cid: str | None, + *, + source: str = 'web', +) -> None: + """Store Yandex CID and fire registration conversion in background (best-effort). + + Opens its own DB session so it never interferes with the caller's transaction. + """ + if not cid: + return + try: + async with AsyncSessionLocal() as db: + stored = await store_cid(db, user_id, cid, source=source) + if stored: + await db.commit() + spawn_bg(fire_registration_bg(user_id)) + except Exception as exc: + logger.warning('Failed to store CID and fire registration', user_id=user_id, error=str(exc)) + + +async def on_registration(db: AsyncSession, user_id: int) -> None: + """Fire registration event (once per user).""" + if not _is_enabled(): + return + + try: + row = await get_cid(db, user_id) + if not row or row.registration_sent: + return + if not row.yandex_cid or row.yandex_cid.startswith('_'): + return # placeholder row — real CID not yet received + + success = await _send_event(row.yandex_cid, 'registration') + if success: + await mark_registration_sent(db, user_id) + await db.commit() + logger.info('registration event sent', user_id=user_id) + except Exception as exc: + logger.error('registration event failed', user_id=user_id, error=str(exc)) + + +async def on_trial(db: AsyncSession, user_id: int) -> None: + """Fire trial-add event (once per user).""" + if not _is_enabled(): + return + + try: + row = await get_cid(db, user_id) + if not row or row.trial_sent: + return + if not row.yandex_cid or row.yandex_cid.startswith('_'): + return # placeholder row — real CID not yet received + + success = await _send_event(row.yandex_cid, 'trial-add') + if success: + await mark_trial_sent(db, user_id) + await db.commit() + logger.info('trial-add event sent', user_id=user_id) + except Exception as exc: + logger.error('trial-add event failed', user_id=user_id, error=str(exc)) + + +async def on_purchase(db: AsyncSession, user_id: int, amount_kopeks: int) -> None: + """Fire ecommerce purchase event (every payment).""" + if not _is_enabled(): + return + + try: + row = await get_cid(db, user_id) + if not row: + return + if not row.yandex_cid or row.yandex_cid.startswith('_'): + return # placeholder row — real CID not yet received + + amount_rubles = amount_kopeks / 100 + payload = _ecommerce_purchase_payload(row.yandex_cid, amount_rubles) + success = await _post_collect(payload, 'purchase', row.yandex_cid) + if success: + logger.info('purchase event sent', user_id=user_id, amount=amount_rubles) + except Exception as exc: + logger.error('purchase event failed', user_id=user_id, error=str(exc)) + + +def parse_cid_from_start_param(param: str) -> tuple[str | None, str]: + """Extract Yandex CID from bot start parameter. + + If param starts with the configured prefix (e.g. 'utm_ya_'), + returns (cid, original_param). Otherwise returns (None, original_param). + Original param is always preserved for UTM tracking. + """ + prefix = settings.YANDEX_OFFLINE_CONV_START_PREFIX + if not prefix or not param.startswith(prefix): + return None, param + + cid = param[len(prefix) :] + normalized = _normalize_cid(cid) + return normalized, param # Keep original param for UTM tracking diff --git a/migrations/alembic/versions/0063_add_yandex_client_id_map.py b/migrations/alembic/versions/0063_add_yandex_client_id_map.py new file mode 100644 index 00000000..429046e7 --- /dev/null +++ b/migrations/alembic/versions/0063_add_yandex_client_id_map.py @@ -0,0 +1,99 @@ +"""add yandex_client_id_map table + guest_purchases offline conv columns + +Revision ID: 0063 +Revises: 0062 +Create Date: 2026-04-21 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = '0063' +down_revision: Union[str, None] = '0062' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + conn = op.get_bind() + + # 1) yandex_client_id_map — created idempotently + result = conn.execute( + sa.text("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'yandex_client_id_map')") + ) + if not result.scalar(): + op.create_table( + 'yandex_client_id_map', + sa.Column('id', sa.Integer, primary_key=True, autoincrement=True), + sa.Column( + 'user_id', + sa.Integer, + sa.ForeignKey('users.id', ondelete='CASCADE'), + unique=True, + nullable=False, + ), + sa.Column('yandex_cid', sa.String(128), nullable=False), + sa.Column('source', sa.String(20), nullable=False, server_default='web'), + sa.Column('counter_id', sa.String(32), nullable=True), + sa.Column( + 'registration_sent', + sa.Boolean, + nullable=False, + server_default=sa.text('false'), + ), + sa.Column( + 'trial_sent', + sa.Boolean, + nullable=False, + server_default=sa.text('false'), + ), + sa.Column('subid', sa.String(255), nullable=True), + sa.Column( + 'created_at', + sa.DateTime(timezone=True), + server_default=sa.func.now(), + ), + sa.Column( + 'updated_at', + sa.DateTime(timezone=True), + server_default=sa.func.now(), + ), + ) + + # 2) guest_purchases — add yandex_cid / subid / referrer (idempotent) + for col_name, col_def in ( + ('yandex_cid', sa.Column('yandex_cid', sa.String(128), nullable=True)), + ('subid', sa.Column('subid', sa.String(255), nullable=True)), + ('referrer', sa.Column('referrer', sa.String(500), nullable=True)), + ): + result = conn.execute( + sa.text( + 'SELECT EXISTS (SELECT 1 FROM information_schema.columns ' + "WHERE table_name = 'guest_purchases' AND column_name = :col)" + ), + {'col': col_name}, + ) + if not result.scalar(): + op.add_column('guest_purchases', col_def) + + +def downgrade() -> None: + conn = op.get_bind() + for col_name in ('referrer', 'subid', 'yandex_cid'): + result = conn.execute( + sa.text( + 'SELECT EXISTS (SELECT 1 FROM information_schema.columns ' + "WHERE table_name = 'guest_purchases' AND column_name = :col)" + ), + {'col': col_name}, + ) + if result.scalar(): + op.drop_column('guest_purchases', col_name) + + result = conn.execute( + sa.text("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'yandex_client_id_map')") + ) + if result.scalar(): + op.drop_table('yandex_client_id_map')