Раздел 6: Celery Tasks

Проект: Интеллектуальная система управления интернет-магазином
Модуль: Shop / Celery
Версия: 1.0
Дата: Январь 2026

6.1 Обзор задач

Назначение

Celery в модуле Shop обеспечивает:

Периодический polling событий из WooCommerce (каждую минуту)
Детектирование новых заказов, отмен, возвратов
Генерацию уведомлений через Notifications

Реестр задач

Задача	Тип	Периодичность	Описание
`shop.poll_orders`	periodic	/1 * * *	Проверка новых событий
`shop.send_notification`	async	по событию	Отправка уведомления

Архитектура

6.2 Task: poll_orders

Назначение

Периодическая проверка WooCommerce на наличие новых событий:

Новые заказы (статус processing)
Отмены (статус cancelled)
Возвраты (статус refunded)
Ошибки оплаты (статус failed)

Конфигурация

# config/celery.py

CELERY_BEAT_SCHEDULE = {
    'shop-poll-orders': {
        'task': 'shop.poll_orders',
        'schedule': crontab(minute='*'),  # каждую минуту
        'options': {'queue': 'default'}
    },
}

Реализация

# tasks/shop_tasks.py

from celery import shared_task
from datetime import datetime, timedelta
import redis
import logging

from services.mcp_client import MCPClient
from services.notifications import NotificationPublisher

logger = logging.getLogger(__name__)

REDIS_KEY_LAST_ORDER_ID = "shop:polling:last_order_id"
REDIS_KEY_LAST_POLL_TIME = "shop:polling:last_poll_time"
REDIS_KEY_KNOWN_ORDERS = "shop:polling:known_orders"

@shared_task(
    name="shop.poll_orders",
    bind=True,
    max_retries=3,
    default_retry_delay=30,
    autoretry_for=(Exception,),
    retry_backoff=True
)
def poll_orders(self):
    """
    Периодическая проверка заказов в WooCommerce.
    Запускается каждую минуту через Celery Beat.
    """
    logger.info("Starting shop.poll_orders")
    
    redis_client = redis.Redis.from_url(settings.REDIS_URL)
    mcp_client = MCPClient()
    notification_publisher = NotificationPublisher()
    
    try:
        # 1. Получение состояния последнего polling
        last_poll_time = redis_client.get(REDIS_KEY_LAST_POLL_TIME)
        if last_poll_time:
            last_poll_time = datetime.fromisoformat(last_poll_time.decode())
        else:
            # Первый запуск — берём заказы за последний час
            last_poll_time = datetime.utcnow() - timedelta(hours=1)
        
        # 2. Запрос к WooCommerce MCP
        orders = mcp_client.orders_list(
            after=last_poll_time.isoformat(),
            per_page=100,
            orderby="date",
            order="asc"
        )
        
        logger.info(f"Fetched {len(orders)} orders since {last_poll_time}")
        
        # 3. Обработка каждого заказа
        for order in orders:
            process_order_event(order, redis_client, notification_publisher)
        
        # 4. Обновление состояния
        redis_client.set(
            REDIS_KEY_LAST_POLL_TIME,
            datetime.utcnow().isoformat()
        )
        
        logger.info("shop.poll_orders completed successfully")
        
    except Exception as e:
        logger.error(f"shop.poll_orders failed: {e}")
        raise


def process_order_event(order: dict, redis_client, notification_publisher):
    """Обработка события по заказу."""
    
    order_id = order["id"]
    status = order["status"]
    
    # Проверка, обрабатывали ли мы этот заказ ранее
    known_status = redis_client.hget(REDIS_KEY_KNOWN_ORDERS, str(order_id))
    known_status = known_status.decode() if known_status else None
    
    if known_status == status:
        # Статус не изменился — пропускаем
        return
    
    # Определение типа события
    event_type = None
    level = "info"
    title = ""
    message = ""
    
    if status == "processing" and known_status is None:
        # Новый заказ
        event_type = "shop.order.new"
        title = "Новый заказ"
        message = f"Заказ #{order_id} на сумму {order['total']} ₽"
        level = "info"
        
    elif status == "cancelled":
        # Отмена
        event_type = "shop.order.cancelled"
        title = "Заказ отменён"
        message = f"Клиент отменил заказ #{order_id}"
        level = "warning"
        
    elif status == "refunded":
        # Возврат
        event_type = "shop.order.refunded"
        title = "Возврат средств"
        message = f"Оформлен возврат по заказу #{order_id}"
        level = "warning"
        
    elif status == "failed":
        # Ошибка оплаты
        event_type = "shop.order.failed"
        title = "Ошибка оплаты"
        message = f"Не удалась оплата заказа #{order_id}"
        level = "error"
    
    # Отправка уведомления
    if event_type:
        send_notification.delay(
            event_type=event_type,
            title=title,
            message=message,
            level=level,
            order_id=order_id,
            order_data={
                "total": order.get("total"),
                "customer": order.get("billing", {}).get("first_name"),
                "status": status
            }
        )
    
    # Обновление known status
    redis_client.hset(REDIS_KEY_KNOWN_ORDERS, str(order_id), status)

Диаграмма процесса

6.3 Task: send_notification

Назначение

Асинхронная отправка уведомления через систему Notifications.

Реализация

@shared_task(
    name="shop.send_notification",
    bind=True,
    max_retries=3,
    default_retry_delay=10
)
def send_notification(
    self,
    event_type: str,
    title: str,
    message: str,
    level: str,
    order_id: int,
    order_data: dict
):
    """
    Отправка уведомления о событии заказа.
    """
    logger.info(f"Sending notification: {event_type} for order #{order_id}")
    
    notification_publisher = NotificationPublisher()
    
    try:
        notification_publisher.publish(
            event_type=event_type,
            title=title,
            message=message,
            source_module="shop",
            source_id=str(order_id),
            level=level,
            data=order_data
        )
        
        logger.info(f"Notification sent: {event_type} for order #{order_id}")
        
    except Exception as e:
        logger.error(f"Failed to send notification: {e}")
        raise self.retry(exc=e)

6.4 MCP Client для Celery

Реализация

# services/mcp_client.py

import httpx
from typing import Optional, List
import logging

logger = logging.getLogger(__name__)

class MCPClient:
    """Клиент для взаимодействия с WooCommerce MCP."""
    
    def __init__(self):
        self.base_url = settings.WOOCOMMERCE_MCP_URL
        self.api_key = f"{settings.WOOCOMMERCE_CONSUMER_KEY}:{settings.WOOCOMMERCE_CONSUMER_SECRET}"
        self.timeout = 30
    
    def _make_request(self, ability: str, params: dict) -> dict:
        """Выполнение MCP запроса."""
        
        headers = {
            "X-MCP-API-Key": self.api_key,
            "Content-Type": "application/json"
        }
        
        # MCP запрос через HTTP
        with httpx.Client(timeout=self.timeout) as client:
            response = client.post(
                f"{self.base_url}/execute",
                headers=headers,
                json={
                    "ability": ability,
                    "params": params
                }
            )
            response.raise_for_status()
            return response.json()
    
    def orders_list(
        self,
        status: Optional[str] = None,
        after: Optional[str] = None,
        before: Optional[str] = None,
        per_page: int = 100,
        orderby: str = "date",
        order: str = "desc"
    ) -> List[dict]:
        """Получение списка заказов."""
        
        params = {
            "per_page": per_page,
            "orderby": orderby,
            "order": order
        }
        
        if status:
            params["status"] = status
        if after:
            params["after"] = after
        if before:
            params["before"] = before
        
        result = self._make_request("woocommerce/orders-list", params)
        return result.get("data", [])
    
    def orders_get(self, order_id: int) -> dict:
        """Получение деталей заказа."""
        
        result = self._make_request(
            "woocommerce/orders-get",
            {"id": order_id}
        )
        return result.get("data", {})

6.5 Redis State Management

Ключи Redis

Ключ	Тип	Описание	TTL
`shop:polling:last_poll_time`	string	Время последнего polling	—
`shop:polling:known_orders`	hash	Известные статусы заказов	7 дней

Операции

# Получение времени последнего polling
last_poll = redis.get("shop:polling:last_poll_time")

# Обновление времени
redis.set("shop:polling:last_poll_time", datetime.utcnow().isoformat())

# Проверка известного статуса заказа
known_status = redis.hget("shop:polling:known_orders", "12345")

# Обновление известного статуса
redis.hset("shop:polling:known_orders", "12345", "processing")

# Очистка старых записей (выполняется периодически)
redis.expire("shop:polling:known_orders", 7 * 24 * 3600)

6.6 Конфигурация

Environment Variables

# Celery
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/0

# Shop Polling
SHOP_POLLING_ENABLED=true
SHOP_POLLING_INTERVAL=60  # секунды

# WooCommerce MCP
WOOCOMMERCE_MCP_URL=https://ohana.market/wp-json/woocommerce/mcp
WOOCOMMERCE_CONSUMER_KEY=ck_xxx
WOOCOMMERCE_CONSUMER_SECRET=cs_xxx

Celery Configuration

# config/celery.py

from celery import Celery
from celery.schedules import crontab

app = Celery('adolf')

app.conf.update(
    broker_url=settings.CELERY_BROKER_URL,
    result_backend=settings.CELERY_RESULT_BACKEND,
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
    timezone='Europe/Moscow',
    enable_utc=True,
    
    # Retry settings
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    
    # Beat schedule
    beat_schedule={
        'shop-poll-orders': {
            'task': 'shop.poll_orders',
            'schedule': 60.0,  # каждые 60 секунд
            'options': {'queue': 'default'}
        },
    }
)

6.7 Обработка ошибок

Retry стратегия

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=30,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=300,
    retry_jitter=True
)
def poll_orders(self):
    ...

Типы ошибок

Ошибка	Действие	Max Retries
ConnectionError	Retry с backoff	3
TimeoutError	Retry с backoff	3
AuthenticationError	Логирование, алерт админу	0
RateLimitError	Retry через 60 сек	3

Логирование ошибок

import logging
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True)
def poll_orders(self):
    try:
        # ... polling logic ...
    except ConnectionError as e:
        logger.warning(f"MCP connection failed: {e}, retry {self.request.retries}")
        raise self.retry(exc=e, countdown=30)
    except Exception as e:
        logger.error(f"poll_orders failed: {e}", exc_info=True)
        raise

6.8 Мониторинг

Метрики

Метрика	Тип	Описание
`shop_poll_duration_seconds`	histogram	Время выполнения polling
`shop_poll_orders_count`	counter	Количество обработанных заказов
`shop_poll_errors_total`	counter	Количество ошибок
`shop_notifications_sent`	counter	Отправленные уведомления

Prometheus метрики

from prometheus_client import Counter, Histogram

POLL_DURATION = Histogram(
    'shop_poll_duration_seconds',
    'Time spent polling WooCommerce'
)

ORDERS_PROCESSED = Counter(
    'shop_poll_orders_count',
    'Number of orders processed',
    ['status']
)

POLL_ERRORS = Counter(
    'shop_poll_errors_total',
    'Number of polling errors',
    ['error_type']
)

@shared_task(bind=True)
def poll_orders(self):
    with POLL_DURATION.time():
        try:
            orders = mcp_client.orders_list(...)
            for order in orders:
                ORDERS_PROCESSED.labels(status=order['status']).inc()
                process_order_event(order, ...)
        except Exception as e:
            POLL_ERRORS.labels(error_type=type(e).__name__).inc()
            raise

Health Check

def check_celery_health():
    """Проверка здоровья Celery."""
    
    # Проверка Beat
    last_poll = redis.get("shop:polling:last_poll_time")
    if last_poll:
        last_poll_time = datetime.fromisoformat(last_poll.decode())
        if datetime.utcnow() - last_poll_time > timedelta(minutes=5):
            return {"status": "unhealthy", "reason": "Polling not running"}
    
    # Проверка Worker
    inspect = celery_app.control.inspect()
    if not inspect.active():
        return {"status": "unhealthy", "reason": "No active workers"}
    
    return {"status": "healthy"}

6.9 Управление задачами

Запуск Celery

# Worker
celery -A app worker --loglevel=info --queues=default

# Beat (scheduler)
celery -A app beat --loglevel=info

# Flower (мониторинг)
celery -A app flower --port=5555

Docker Compose

services:
  celery-worker:
    build: .
    command: celery -A app worker --loglevel=info
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - WOOCOMMERCE_MCP_URL=${WOOCOMMERCE_MCP_URL}
    depends_on:
      - redis
    
  celery-beat:
    build: .
    command: celery -A app beat --loglevel=info
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    depends_on:
      - redis
      - celery-worker

Ручной запуск задачи

# Принудительный polling
from tasks.shop_tasks import poll_orders
poll_orders.delay()

# Проверка статуса
result = poll_orders.delay()
print(result.status)  # PENDING, SUCCESS, FAILURE

Приложение А: Контрольные точки

Критерий	Проверка
Beat работает	Задачи появляются в очереди каждую минуту
Worker работает	Задачи обрабатываются
Polling работает	last_poll_time обновляется
Уведомления	События создают notifications
Retry работает	Ошибки не приводят к потере событий

Приложение Б: Troubleshooting

Polling не запускается

# Проверить Beat
celery -A app beat --loglevel=debug

# Проверить очередь
redis-cli LLEN celery

# Проверить Worker
celery -A app inspect active

Уведомления не приходят

# Проверить логи Worker
grep "send_notification" /var/log/celery/worker.log

# Проверить Redis state
redis-cli GET "shop:polling:last_poll_time"
redis-cli HGETALL "shop:polling:known_orders"

Ошибки MCP

# Тест подключения
curl -H "X-MCP-API-Key: ck_xxx:cs_xxx" \
     https://ohana.market/wp-json/woocommerce/mcp

# Проверить логи
grep "MCP" /var/log/celery/worker.log

Документ подготовлен: Январь 2026
Версия: 1.0
Статус: Черновик

WooCommerce MCP

​6.1 Обзор задач

​Назначение

​Реестр задач

​Архитектура

​6.2 Task: poll_orders

​Назначение

​Конфигурация

​Реализация

​Диаграмма процесса

​6.3 Task: send_notification

​Назначение

​Реализация

​6.4 MCP Client для Celery

​Реализация

​6.5 Redis State Management

​Ключи Redis

​Операции

​6.6 Конфигурация

​Environment Variables

​Celery Configuration

​6.7 Обработка ошибок

​Retry стратегия

​Типы ошибок

​Логирование ошибок

​6.8 Мониторинг

​Метрики

​Prometheus метрики

​Health Check

​6.9 Управление задачами

​Запуск Celery

​Docker Compose

​Ручной запуск задачи

​Приложение А: Контрольные точки

​Приложение Б: Troubleshooting

​Polling не запускается

​Уведомления не приходят

​Ошибки MCP

6.1 Обзор задач

Назначение

Реестр задач

Архитектура

6.2 Task: poll_orders

Назначение

Конфигурация

Реализация

Диаграмма процесса

6.3 Task: send_notification

Назначение

Реализация

6.4 MCP Client для Celery

Реализация

6.5 Redis State Management

Ключи Redis

Операции

6.6 Конфигурация

Environment Variables

Celery Configuration

6.7 Обработка ошибок

Retry стратегия

Типы ошибок

Логирование ошибок

6.8 Мониторинг

Метрики

Prometheus метрики

Health Check

6.9 Управление задачами

Запуск Celery

Docker Compose

Ручной запуск задачи

Приложение А: Контрольные точки

Приложение Б: Troubleshooting

Polling не запускается

Уведомления не приходят

Ошибки MCP