Раздел 2: Polling

from celery.schedules import crontab

# Расписание с распределённым смещением
# Интервал 5 минут, смещение ~50 секунд между задачами
beat_schedule = {
    # Wildberries Reviews: */5 :00
    "poll-wb-reviews": {
        "task": "tasks.reputation_tasks.poll_wb_reviews",
        "schedule": crontab(minute="*/5"),
        "options": {"queue": "default"}
    },
    # Wildberries Questions: */5 :50
    "poll-wb-questions": {
        "task": "tasks.reputation_tasks.poll_wb_questions",
        "schedule": crontab(minute="0,5,10,15,20,25,30,35,40,45,50,55"),
        "options": {"queue": "default", "countdown": 50}
    },
    
    # Ozon Reviews: */5 1:40
    "poll-ozon-reviews": {
        "task": "tasks.reputation_tasks.poll_ozon_reviews",
        "schedule": crontab(minute="1,6,11,16,21,26,31,36,41,46,51,56"),
        "options": {"queue": "default", "countdown": 40}
    },
    # Ozon Questions: */5 2:30
    "poll-ozon-questions": {
        "task": "tasks.reputation_tasks.poll_ozon_questions",
        "schedule": crontab(minute="2,7,12,17,22,27,32,37,42,47,52,57"),
        "options": {"queue": "default", "countdown": 30}
    },
    
    # Яндекс.Маркет Reviews: */5 3:20
    "poll-ym-reviews": {
        "task": "tasks.reputation_tasks.poll_ym_reviews",
        "schedule": crontab(minute="3,8,13,18,23,28,33,38,43,48,53,58"),
        "options": {"queue": "default", "countdown": 20}
    },
    # Яндекс.Маркет Questions: */5 4:10
    "poll-ym-questions": {
        "task": "tasks.reputation_tasks.poll_ym_questions",
        "schedule": crontab(minute="4,9,14,19,24,29,34,39,44,49,54,59"),
        "options": {"queue": "default", "countdown": 10}
    },
}

Минута 0:  WB Reviews  ─────────────────────────────────────────
           :00         :50
Минута 1:              ───── WB Questions ─────
                             1:40
Минута 2:                    ───── Ozon Reviews ─────
                                   2:30
Минута 3:                          ───── Ozon Questions ─────
                                         3:20
Минута 4:                                ───── YM Reviews ─────
                                               4:10
Минута 5:  WB Reviews  ────────────────────── YM Questions ─────

# Wildberries
WB_API_TOKEN=xxx
WB_API_TOKEN_EXPIRES=2026-06-01T00:00:00Z

# Ozon
OZON_CLIENT_ID=xxx
OZON_API_KEY=xxx

# Яндекс.Маркет
YM_OAUTH_TOKEN=xxx
YM_CAMPAIGN_ID=xxx

# config/marketplace_credentials.py
import os
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class WBCredentials:
    api_token: str
    expires_at: Optional[datetime] = None

@dataclass
class OzonCredentials:
    client_id: str
    api_key: str

@dataclass
class YMCredentials:
    oauth_token: str
    campaign_id: str

def get_wb_credentials() -> WBCredentials:
    expires_str = os.getenv("WB_API_TOKEN_EXPIRES")
    expires_at = datetime.fromisoformat(expires_str) if expires_str else None
    return WBCredentials(
        api_token=os.environ["WB_API_TOKEN"],
        expires_at=expires_at
    )

def get_ozon_credentials() -> OzonCredentials:
    return OzonCredentials(
        client_id=os.environ["OZON_CLIENT_ID"],
        api_key=os.environ["OZON_API_KEY"]
    )

def get_ym_credentials() -> YMCredentials:
    return YMCredentials(
        oauth_token=os.environ["YM_OAUTH_TOKEN"],
        campaign_id=os.environ["YM_CAMPAIGN_ID"]
    )

Client-Id: {OZON_CLIENT_ID}
Api-Key: {OZON_API_KEY}
Content-Type: application/json

{
  "filter": {
    "interaction_status": "NEED_ANSWER",
    "published_at_from": "2026-01-01T00:00:00Z"
  },
  "last_id": "",
  "limit": 100,
  "sort_by": "PUBLISHED_AT",
  "sort_dir": "DESC"
}

Authorization: Bearer {YM_OAUTH_TOKEN}
Content-Type: application/json

# services/rate_limiter.py
import redis
import time
from typing import Optional

class RateLimiter:
    """Token bucket rate limiter с Redis backend."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def acquire(
        self, 
        key: str, 
        max_tokens: int, 
        refill_rate: float,
        timeout: float = 30.0
    ) -> bool:
        """
        Попытка получить токен.
        
        Args:
            key: Уникальный ключ лимита (например, 'wb:feedbacks')
            max_tokens: Максимальное количество токенов
            refill_rate: Скорость пополнения (токенов/сек)
            timeout: Максимальное время ожидания
        
        Returns:
            True если токен получен, False если timeout
        """
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            tokens = self._get_tokens(key, max_tokens, refill_rate)
            
            if tokens >= 1:
                self._consume_token(key)
                return True
            
            # Ждём пополнения
            wait_time = 1 / refill_rate
            time.sleep(min(wait_time, 1.0))
        
        return False
    
    def _get_tokens(self, key: str, max_tokens: int, refill_rate: float) -> float:
        """Получить текущее количество токенов."""
        pipe = self.redis.pipeline()
        now = time.time()
        
        # Получить последнее обновление и количество токенов
        data = self.redis.hgetall(f"ratelimit:{key}")
        
        if not data:
            # Инициализация: полный bucket
            self.redis.hset(f"ratelimit:{key}", mapping={
                "tokens": max_tokens,
                "last_update": now
            })
            return max_tokens
        
        tokens = float(data.get(b"tokens", max_tokens))
        last_update = float(data.get(b"last_update", now))
        
        # Пополнение токенов
        elapsed = now - last_update
        tokens = min(max_tokens, tokens + elapsed * refill_rate)
        
        # Обновить состояние
        self.redis.hset(f"ratelimit:{key}", mapping={
            "tokens": tokens,
            "last_update": now
        })
        
        return tokens
    
    def _consume_token(self, key: str):
        """Использовать один токен."""
        self.redis.hincrbyfloat(f"ratelimit:{key}", "tokens", -1)

RATE_LIMITS = {
    "wb:feedbacks": {"max_tokens": 100, "refill_rate": 100/60},
    "wb:questions": {"max_tokens": 100, "refill_rate": 100/60},
    "ozon:reviews": {"max_tokens": 60, "refill_rate": 60/60},
    "ozon:questions": {"max_tokens": 60, "refill_rate": 60/60},
    "ym:feedbacks": {"max_tokens": 100, "refill_rate": 100/60},
    "ym:questions": {"max_tokens": 100, "refill_rate": 100/60},
}

# services/circuit_breaker.py
from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    """Circuit breaker для защиты от каскадных сбоев."""
    
    def __init__(
        self,
        failure_threshold: int = 5,
        reset_timeout: int = 60,
        db_session = None
    ):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.db = db_session
    
    def get_state(self, platform: str, item_type: str) -> CircuitState:
        """Получить состояние circuit breaker."""
        state = self.db.query(PollingState).filter(
            PollingState.platform == platform,
            PollingState.item_type == item_type
        ).first()
        
        if not state:
            return CircuitState.CLOSED
        
        if state.circuit_breaker_open:
            if datetime.utcnow() > state.circuit_breaker_until:
                return CircuitState.HALF_OPEN
            return CircuitState.OPEN
        
        return CircuitState.CLOSED
    
    def record_success(self, platform: str, item_type: str):
        """Записать успешный запрос."""
        self.db.query(PollingState).filter(
            PollingState.platform == platform,
            PollingState.item_type == item_type
        ).update({
            "consecutive_errors": 0,
            "circuit_breaker_open": False,
            "circuit_breaker_until": None,
            "last_poll_status": "success"
        })
        self.db.commit()
    
    def record_failure(self, platform: str, item_type: str, error: str):
        """Записать неудачный запрос."""
        state = self.db.query(PollingState).filter(
            PollingState.platform == platform,
            PollingState.item_type == item_type
        ).first()
        
        state.consecutive_errors += 1
        state.last_poll_status = "error"
        state.last_poll_error = error
        
        if state.consecutive_errors >= self.failure_threshold:
            state.circuit_breaker_open = True
            state.circuit_breaker_until = datetime.utcnow() + timedelta(seconds=self.reset_timeout)
        
        self.db.commit()

# Структура лога polling
{
    "event": "polling_completed",
    "platform": "wb",
    "item_type": "review",
    "duration_ms": 1234,
    "items_fetched": 15,
    "items_new": 10,
    "items_duplicate": 5,
    "errors": 0,
    "timestamp": "2026-01-10T14:30:00Z"
}