Интеграция с Platrum Spider

Обзор

Platrum Spider — CLI-инструмент на Python, разработанный Fotofactor для синхронизации данных из Platrum.ru (система управления проектами). Интеграция с SalesFinder позволяет автоматически загружать маркетплейс-аналитику в единую базу данных, запускать мониторинг позиций товаров и генерировать отчёты — всё через командную строку.

Зачем это нужно

Fotofactor создаёт контент для маркетплейсов (фотосъёмка, видео, инфографика). Данные SalesFinder помогают понять, какой контент нужен клиенту, отслеживать эффект от обновлённых карточек и проактивно предлагать оптимизацию.

Архитектура интеграции

Поток данных

┌──────────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│  SalesFinder  │───>│   SQLite     │───>│   Monitors   │───>│   Actions    │
│     API       │    │   Cache      │    │   Engine     │    │   Engine     │
│  (Pro tier)   │    │ (sf_tables)  │    │ (отклонения) │    │ (алерты,     │
└──────────────┘    └──────────────┘    └──────────────┘    │  задачи)     │
                           │                                 └──────────────┘
                           │
                    ┌──────┴───────┐
                    │  CLI команды  │
                    │  salesfinder   │
                    │  analytics/   │
                    │  trends/      │
                    │  competitors  │
                    └──────────────┘

Интеграция с существующими модулями

SalesFinder Syncer ─────┐
                         ├──> Tasks Module: автосоздание задач при падении позиций
                         ├──> Wiki Module: генерация аналитических отчётов в базу знаний
                         ├──> Finance Module: привязка расходов на аналитику к проектам
                         └──> OKR Module: метрики позиций товаров как Key Results

Новые CLI-команды

После интеграции появятся следующие команды:

# Синхронизация данных из SalesFinder
python spider.py sync salesfinder              # Инкрементальная синхронизация
python spider.py sync salesfinder --full       # Полная пересинхронизация

# Аналитика по товарам
python spider.py salesfinder analytics --product-id 12345 --json
python spider.py salesfinder analytics --category "Электроинструменты" --limit 20

# Тренды продаж и цен
python spider.py salesfinder trends --brand "Makita" --days 30
python spider.py salesfinder trends --category-id 456 --metric sales --json

# Конкурентный анализ
python spider.py salesfinder competitors --product-id 12345 --top 10
python spider.py salesfinder competitors --brand "Makita" --vs "Bosch,DeWalt" --json

# Мониторинг позиций
python spider.py salesfinder positions --brand "Makita" --alert-drop 10
python spider.py salesfinder positions --category "Фототехника" --json

# Отчёты
python spider.py salesfinder report --brand "Makita" --period weekly -o report.md
python spider.py salesfinder report --client "ClientName" --format pdf

Принцип CLI-First

В Platrum Spider действует строгая политика: все операции выполняются только через CLI. Никаких отдельных скриптов — если команда не существует, она создаётся в первую очередь.

Структура базы данных

Новые таблицы

Интеграция добавляет 4 основные таблицы в SQLite:

-- Товары из маркетплейсов
CREATE TABLE sf_products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    external_id VARCHAR(50) NOT NULL,          -- ID товара в SalesFinder
    marketplace VARCHAR(20) NOT NULL,           -- 'wb' | 'ozon'
    sku VARCHAR(50),                            -- Артикул
    brand VARCHAR(200),
    name TEXT NOT NULL,
    category_id INTEGER,
    current_price FLOAT,
    current_position INTEGER,
    sales_estimate INTEGER,                     -- Оценка продаж за период
    revenue_estimate FLOAT,                     -- Оценка выручки
    rating FLOAT,
    reviews_count INTEGER,
    images_count INTEGER,
    content_score INTEGER,                      -- Оценка качества карточки (0-100)
    last_updated DATETIME DEFAULT CURRENT_TIMESTAMP,
    client_id VARCHAR(50),                      -- Привязка к клиенту Fotofactor
    UNIQUE(external_id, marketplace)
);

-- Категории маркетплейсов
CREATE TABLE sf_categories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    external_id VARCHAR(50) NOT NULL,
    marketplace VARCHAR(20) NOT NULL,
    name TEXT NOT NULL,
    parent_id INTEGER,
    products_count INTEGER,
    avg_price FLOAT,
    competition_level VARCHAR(20),              -- 'low' | 'medium' | 'high'
    UNIQUE(external_id, marketplace)
);

-- Поисковые запросы и SEO-данные
CREATE TABLE sf_keywords (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    keyword TEXT NOT NULL,
    marketplace VARCHAR(20) NOT NULL,
    frequency INTEGER,                          -- Частотность запроса
    products_count INTEGER,                     -- Кол-во товаров в выдаче
    product_id INTEGER REFERENCES sf_products(id),
    position INTEGER,                           -- Позиция товара по запросу
    collected_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(keyword, marketplace, product_id)
);

-- История цен для мониторинга РРЦ
CREATE TABLE sf_price_history (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    product_id INTEGER NOT NULL REFERENCES sf_products(id),
    price FLOAT NOT NULL,
    old_price FLOAT,                            -- Зачёркнутая цена
    discount_percent FLOAT,
    collected_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- Индексы для быстрых запросов
CREATE INDEX idx_sf_products_brand ON sf_products(brand);
CREATE INDEX idx_sf_products_marketplace ON sf_products(marketplace);
CREATE INDEX idx_sf_products_client ON sf_products(client_id);
CREATE INDEX idx_sf_keywords_keyword ON sf_keywords(keyword);
CREATE INDEX idx_sf_price_history_product ON sf_price_history(product_id);
CREATE INDEX idx_sf_price_history_date ON sf_price_history(collected_at);

Реализация SalesFinder Syncer

Паттерн синхронизатора

Syncer следует оптимизированному паттерну Platrum Spider с bulk upsert и инкрементальной синхронизацией:

"""SalesFinder Syncer — синхронизация маркетплейс-аналитики."""

from datetime import datetime, timedelta
from typing import Optional

from ..api.salesfinder_client import SalesFinderClient
from ..sync.base import BaseSyncer
from ..sync.bulk_operations import BulkUpsertMixin, PerformanceTracker
from ..database.models import SFProduct, SFCategory, SFKeyword, SFPriceHistory
from ..database.session import get_session


class SalesFinderSyncer(BaseSyncer, BulkUpsertMixin):
    """Синхронизатор данных из SalesFinder API.

    Поддерживает:
    - Инкрементальная синхронизация (только изменения с последнего запуска)
    - Bulk upsert операции (500 записей за batch)
    - Отслеживание производительности
    - Rate limiting (согласно лимитам SalesFinder API)
    """

    def __init__(self, api_client: SalesFinderClient):
        super().__init__(name="salesfinder")
        self.api = api_client
        self.tracker = PerformanceTracker()

    async def _do_sync(self, full_sync: bool = False) -> int:
        """Основной метод синхронизации.

        Args:
            full_sync: True — полная пересинхронизация,
                       False — только изменения с последнего запуска.

        Returns:
            Количество синхронизированных записей.
        """
        total_synced = 0

        # Фаза 1: Синхронизация категорий
        self.tracker.start_phase("categories")
        categories = await self.api.get_categories()
        with get_session() as session:
            total_synced += self.bulk_upsert(
                session=session,
                model=SFCategory,
                records=categories,
                unique_fields=["external_id", "marketplace"],
                batch_size=500
            )
        self.tracker.end_phase("categories")

        # Фаза 2: Синхронизация товаров
        self.tracker.start_phase("products")
        since_date = None if full_sync else self._get_last_sync_date()
        products = await self.api.get_products(updated_since=since_date)
        with get_session() as session:
            total_synced += self.bulk_upsert(
                session=session,
                model=SFProduct,
                records=products,
                unique_fields=["external_id", "marketplace"],
                batch_size=500
            )
        self.tracker.end_phase("products")

        # Фаза 3: Синхронизация ценовой истории
        self.tracker.start_phase("price_history")
        for product_batch in self._chunked(products, 100):
            prices = await self.api.get_price_history(
                product_ids=[p["external_id"] for p in product_batch],
                since=since_date
            )
            with get_session() as session:
                total_synced += self.bulk_upsert(
                    session=session,
                    model=SFPriceHistory,
                    records=prices,
                    unique_fields=["product_id", "collected_at"],
                    batch_size=500
                )
        self.tracker.end_phase("price_history")

        # Фаза 4: Синхронизация ключевых слов
        self.tracker.start_phase("keywords")
        keywords = await self.api.get_keywords(updated_since=since_date)
        with get_session() as session:
            total_synced += self.bulk_upsert(
                session=session,
                model=SFKeyword,
                records=keywords,
                unique_fields=["keyword", "marketplace", "product_id"],
                batch_size=500
            )
        self.tracker.end_phase("keywords")

        self.tracker.print_summary()
        return total_synced

    def _get_last_sync_date(self) -> Optional[datetime]:
        """Получает дату последней синхронизации из sync_status."""
        with get_session() as session:
            status = session.query(SyncStatus).filter_by(
                module="salesfinder"
            ).first()
            if status and status.last_sync:
                return status.last_sync
        return datetime.utcnow() - timedelta(days=7)  # Фолбэк: неделя

API-клиент SalesFinder

"""HTTP-клиент для SalesFinder API с rate limiting."""

import httpx
from ..api.rate_limiter import RateLimiter


class SalesFinderClient:
    """Клиент для работы с SalesFinder API.

    Требует подписку Pro для доступа к API.
    Rate limit: 10 req/sec (настраивается).
    """

    BASE_URL = "https://api.salesfinder.ru/v1"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.limiter = RateLimiter(max_per_second=10, max_per_minute=200)
        self.client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=120.0
        )

    async def get_products(self, updated_since=None, **kwargs):
        """Получить список товаров с фильтрацией."""
        await self.limiter.acquire()
        params = {"updated_since": updated_since, **kwargs}
        response = await self.client.get("/products", params=params)
        response.raise_for_status()
        return response.json()["data"]

    async def get_categories(self, marketplace="wb"):
        """Получить дерево категорий маркетплейса."""
        await self.limiter.acquire()
        response = await self.client.get(
            "/categories", params={"marketplace": marketplace}
        )
        response.raise_for_status()
        return response.json()["data"]

    async def get_price_history(self, product_ids, since=None):
        """Получить историю цен для списка товаров."""
        await self.limiter.acquire()
        response = await self.client.post(
            "/prices/history",
            json={"product_ids": product_ids, "since": since}
        )
        response.raise_for_status()
        return response.json()["data"]

    async def get_keywords(self, updated_since=None):
        """Получить поисковые запросы и позиции."""
        await self.limiter.acquire()
        response = await self.client.get(
            "/keywords", params={"updated_since": updated_since}
        )
        response.raise_for_status()
        return response.json()["data"]

Сценарии использования

1. Автоматический мониторинг позиций товаров

Система отслеживает позиции товаров клиентов и создаёт алерты при значительном падении.

# Настроить мониторинг для клиента
python spider.py salesfinder positions --client "MakitaRU" --alert-drop 10

# Пример вывода при срабатывании:
# ⚠️  ALERT: Makita DHP453 — позиция упала с #3 на #15 (запрос "шуруповёрт аккумуляторный")
# 📋 Создана задача #3012 в Platrum: "Оптимизировать карточку Makita DHP453"

Как работает монитор:

class PositionDropMonitor(BaseMonitor):
    """Мониторит падение позиций товаров клиентов."""

    async def _do_check(self):
        issues = []
        with get_session() as session:
            # Сравниваем позиции за последние 7 дней
            drops = session.execute(text("""
                SELECT p.name, p.brand, k.keyword,
                       k.position as current_pos,
                       prev.position as prev_pos,
                       (k.position - prev.position) as drop_amount
                FROM sf_keywords k
                JOIN sf_products p ON k.product_id = p.id
                JOIN sf_keywords prev ON prev.keyword = k.keyword
                    AND prev.product_id = k.product_id
                    AND prev.collected_at < k.collected_at
                WHERE k.position > prev.position + :threshold
                    AND k.collected_at > :since
                ORDER BY drop_amount DESC
            """), {"threshold": 10, "since": week_ago})

            for row in drops:
                issues.append({
                    "product_name": row.name,
                    "brand": row.brand,
                    "keyword": row.keyword,
                    "severity": "critical" if row.drop_amount > 20 else "high",
                    "message": f"Позиция упала: {row.prev_pos} → {row.current_pos}"
                })
        return issues

Автоматическое создание задач

При обнаружении падения позиции более чем на 20 пунктов система автоматически создаёт задачу в Platrum с назначением ответственного менеджера. Задача содержит рекомендации по оптимизации карточки на основе данных SalesFinder.

2. Мониторинг цен конкурентов

# Отслеживать цены конкурентов
python spider.py salesfinder competitors --brand "Makita" --vs "Bosch,DeWalt" --json

# Пример JSON-ответа:
{
  "brand": "Makita",
  "category": "Шуруповёрты",
  "competitors": [
    {
      "brand": "Bosch",
      "avg_price": 12500,
      "price_diff_percent": -8.2,
      "products_count": 45,
      "avg_rating": 4.6
    },
    {
      "brand": "DeWalt",
      "avg_price": 14200,
      "price_diff_percent": +4.1,
      "products_count": 38,
      "avg_rating": 4.5
    }
  ]
}

3. Еженедельные аналитические отчёты

# Генерация недельного отчёта
python spider.py salesfinder report --brand "Makita" --period weekly -o report.md

# Отчёт автоматически сохраняется в Wiki
python spider.py wiki articles-save \
  --title "Аналитика Makita: неделя 7/2026" \
  --space-id 5 \
  --content-file report.md \
  --real

4. SEO-данные для контент-воркфлоу

# Топ ключевых слов для категории
python spider.py salesfinder keywords --category "Фототехника" --top 50 --json

# Результат: список ключевых слов с частотностью для создания контента
# Используется AI Content Factory для генерации описаний карточек

Интеграция с существующими модулями

Связь с Tasks Module

При критических событиях (падение позиций, нарушение РРЦ) автоматически создаются задачи:

# Автосоздание задачи при падении позиции
async def create_position_alert_task(product, drop_info):
    """Создаёт задачу в Platrum при падении позиции товара."""
    task_data = {
        "name": f"[SF] Оптимизировать карточку: {product.name}",
        "description": (
            f"Позиция товара {product.name} ({product.sku}) "
            f"упала с #{drop_info.prev_pos} на #{drop_info.current_pos} "
            f"по запросу \"{drop_info.keyword}\".\n\n"
            f"Рекомендации:\n"
            f"1. Обновить главное фото\n"
            f"2. Оптимизировать заголовок под запрос\n"
            f"3. Добавить инфографику с УТП"
        ),
        "board_panel_id": CONTENT_PANEL_ID,
        "responsible_user_ids": [CONTENT_MANAGER_ID],
        "is_important": drop_info.drop_amount > 20,
        "deadline": (datetime.now() + timedelta(days=3)).isoformat()
    }
    return await write_client.create_task(task_data)

Связь с Wiki Module

Еженедельные отчёты автоматически публикуются в базу знаний:

# Генерация и публикация отчёта
python spider.py salesfinder report --period weekly --publish-wiki --space-id 5

Связь с OKR Module

Метрики SalesFinder привязываются к целям компании:

OKR	Метрика SalesFinder	Тип
Увеличить выручку клиентов на 25%	revenue_estimate по брендам клиентов	money
Вывести 50 товаров в топ-10	position <= 10 count	task_count
Повысить content score до 85+	Средний content_score клиентов	percentage

Технические требования

Подписка SalesFinder

Параметр	Требование
Тариф	Pro (необходим доступ к API)
API-лимиты	10 req/sec, 200 req/min
Данные	WB + Ozon (оба маркетплейса)
История	Минимум 90 дней ценовой истории

Стек технологий

Компонент	Технология
HTTP-клиент	httpx (async, как в основном проекте)
Rate limiting	Token bucket (9 req/sec — аналогично Platrum API)
База данных	SQLite + SQLAlchemy 2.0
CLI	Typer + Rich (как в основном проекте)
Валидация	Pydantic v2

Переменные окружения

# Добавить в .env
SALESFINDER_API_KEY=sf_xxxxxxxxxxxxx    # API-ключ SalesFinder
SALESFINDER_TIER=pro                     # Тариф подписки
SALESFINDER_RATE_LIMIT=10               # Запросов в секунду

План реализации

Фаза 1: Базовая синхронизация (1 неделя)

• Создать SalesFinderClient в src/api/salesfinder_client.py
• Создать модели данных в src/database/models.py
• Написать миграцию для новых таблиц
• Реализовать SalesFinderSyncer в src/sync/salesfinder_syncer.py
• Добавить команду sync salesfinder в CLI

Фаза 2: Аналитика и CLI (1 неделя)

• Команда salesfinder analytics — просмотр данных по товарам
• Команда salesfinder trends — графики трендов
• Команда salesfinder competitors — конкурентный анализ
• Команда salesfinder keywords — SEO-аналитика
• Поддержка --json для всех команд

Фаза 3: Мониторинг и алерты (1 неделя)

• PositionDropMonitor — мониторинг падения позиций
• PriceViolationMonitor — мониторинг нарушений РРЦ
• ContentScoreMonitor — мониторинг качества карточек
• Интеграция с Tasks Module (автосоздание задач)
• Интеграция с Wiki Module (автопубликация отчётов)

Фаза 4: Отчёты и дашборды (1 неделя)

• Еженедельные автоотчёты в Platrum Wiki
• Экспорт в CSV/JSON/Excel
• Streamlit-дашборд для визуализации
• Telegram-уведомления о критических событиях

Оценка трудозатрат

Полная интеграция занимает 4 недели разработки. Основная сложность — в корректной обработке данных SalesFinder API и настройке мониторов с правильными порогами срабатывания.

Пример полного рабочего процесса

# 1. Первичная синхронизация
python spider.py sync salesfinder --full

# 2. Проверка статуса
python spider.py status
# → SalesFinder: 2,450 products, 18,300 keywords, last sync: 5 min ago

# 3. Анализ конкретного бренда
python spider.py salesfinder analytics --brand "Makita" --json

# 4. Запуск мониторов
python spider.py monitor --show-issues
# → [SF] Позиция Makita DHP453 упала на 12 пунктов (critical)
# → [SF] Нарушение РРЦ: Bosch GSR 18V — цена ниже на 15% (high)

# 5. Выполнение действий (создание задач, комментариев)
python spider.py action --real

# 6. Генерация отчёта
python spider.py salesfinder report --period weekly -o weekly_sf.md

# 7. Настройка автосинхронизации (каждый час)
python spider.py scheduler-cmd start --duration 86400