Как я забираю чеки из личного кабинета ОФД: Selenium, идемпотентность и разбор JSON

Есть тип задач, который выглядит как «нажми кнопку и скачай файл», а на деле упирается в три вещи: кабинет не отдаёт данные за нужный период целиком, позиции чеков лежат отдельно от шапок, а выгружать надо не разово, а каждую неделю. В статье — как это устроено у меня изнутри, с кодом и с тем, что ломалось.

Что вообще есть в кабинете оператора

Онлайн‑касса отправляет каждый чек оператору фискальных данных. В кабинете ОФД видно то, что оператор получил: номер фискального накопителя, номер фискального документа, фискальный признак, дату и время, номер кассы, сумму, признак расчёта и — в части кабинетов — построчные позиции.

Исторически так, что «простой» отчёт по кнопке отдаёт только шапки. Позиции лежат в теле документа, и до них надо добираться либо отдельным отчётом, либо через API, либо выгрузкой по каждому чеку. Это первое место, где ручная работа вырастает в часы: тысяча чеков — тысяча карточек.

Форматы тоже разные: CSV с точкой с запятой, XLSX, JSON, иногда PDF на каждый чек. У одного оператора экспорт работает сразу, у другого отчёт формируется в фоне и появляется через несколько минут — и это надо уметь ждать кодом, а не глазами.

Почему Selenium, а не requests

Правильный ответ на «как забрать данные из кабинета» — API. У большинства операторов он есть, но условия разные: где‑то заявка в поддержку, где‑то отдельный договор и оплата за объём запросов. Когда клиенту нужна автоматическая выгрузка, а доступа к API нет, остаётся кабинет.

Кабинет — это SPA с формой логина, CSRF‑токеном, иногда капчей и подтверждением по SMS. Пройти это библиотекой запросов можно, но каждый раз, когда меняется схема авторизации, код приходится переписывать. Браузер под управлением Selenium проходит авторизацию так же, как человек, и переживает смену вёрстки логина. Дальше — важный приём: после логина я не парсю HTML, а смотрю, какие XHR‑запросы делает сам интерфейс, и повторяю их из контекста страницы.

# после авторизации в кабинете: дергаем внутренний эндпоинт из страницы
import json

def fetch_receipts(driver, period_from, period_to, page=1):
    js = """
    const done = arguments[arguments.length - 1];
    fetch('/api/v1/documents?dateFrom=' + arguments[0] +
          '&dateTo=' + arguments[1] + '&page=' + arguments[2], {
        credentials: 'include',
        headers: {'Accept': 'application/json'}
    }).then(r => r.text()).then(t => done(t))
      .catch(e => done('ERROR: ' + e));
    """
    driver.set_script_timeout(60)
    raw = driver.execute_async_script(js, period_from, period_to, page)
    if raw.startswith('ERROR: '):
        raise RuntimeError(raw)
    return json.loads(raw)

Такой подход даёт скорость и структуру, но остаётся заложником чужого API: параметры могут поменяться, а лимит на количество запросов — внезапно включиться. Поэтому у меня в коде всегда есть второй путь — скачивание файла через интерфейс, и переключатель между ними.

Схема, которая пережила полгода эксплуатации

Пять шагов, каждый — отдельная функция, чтобы падение одного не тянуло за собой остальные:

1. авторизация в кабинете (с сохранением профиля браузера, чтобы не логиниться каждый запуск);

2. разбивка периода на куски по 7–14 дней — кабинеты не любят длинные интервалы;

3. забор документов (API кабинета или файл через интерфейс);

4. нормализация: единый набор колонок, суммы в рублях, время в одном часовом поясе;

5. запись в базу по ключу, который не даёт задвоить чек при повторном прогоне.

Отдельная таблица‑чекпоинт хранит, до какой даты данные уже забраны. Если запуск упал на середине, следующий продолжит с последней успешной даты, а не начнёт заново.

Скачивание файла: ждать появления, а не спать

Классическая ошибка — time.sleep(30) после клика. На медленном канале этого не хватит, на быстром — потеряется полминуты на каждой кассе. Правильно — смотреть в папку загрузок и ждать, пока исчезнет временный файл .crdownload.

import glob
import os
import time


def download_report(driver, download_dir, click_export, timeout=120):
    """Жмёт «Экспорт» и возвращает путь к готовому файлу."""
    before = set(glob.glob(os.path.join(download_dir, '*')))
    click_export()

    deadline = time.time() + timeout
    while time.time() < deadline:
        fresh = set(glob.glob(os.path.join(download_dir, '*'))) - before
        ready = [f for f in fresh if not f.endswith('.crdownload')]
        if ready:
            return ready[0]
        time.sleep(1)
    raise TimeoutError('экспорт не завершился за %s с' % timeout)

Папку загрузок задаю в опциях драйвера и делаю уникальной на запуск, иначе старые файлы путают логику «чего появилось нового».

Разбор: шапки, позиции и суммы в копейках

Внутри JSON суммы почти всегда в копейках, а дата — либо ISO‑строка с таймзоной, либо «наивное» локальное время. Если этого не учесть, в отчёте появляются суммы в сто раз больше и чеки, уехавшие на несколько часов.

import pandas as pd


def receipts_to_frame(payload):
    rows = []
    for doc in payload.get('documents', []):
        head = doc['header']
        rows.append({
            'fn': doc['fn'],
            'fd': int(doc['fiscalDocumentNumber']),
            'fpd': doc['fiscalDocumentSign'],
            'dt': pd.to_datetime(head['dateTime'], utc=True)
                    .tz_convert('Europe/Moscow'),
            'kkt': head['kktRegNumber'],
            'operation': int(head['operationType']),
            'total': int(head['totalSum']) / 100,
            'cash': int(head.get('cashSum', 0)) / 100,
            'card': int(head.get('ecashSum', 0)) / 100,
        })
    return pd.DataFrame(rows)

operationType — то, из‑за чего чаще всего «не сходится с банком»: приход, возврат прихода, расход, возврат расхода. Возвраты нельзя складывать в общую выручку, их считают отдельной строкой отчёта. Если этого не сделать, сверка с эквайрингом не закроется никогда.

Идемпотентность: почему ключ — не «дата + сумма»

Первая версия писала в таблицу всё, что пришло, а дубли вычищала запросом «удали строки с одинаковой датой и суммой». Это работало до первого дня с двумя одинаковыми чеками на 350 рублей в одну минуту — и удаляло настоящий чек.

Правильный естественный ключ — связка номера фискального накопителя, номера фискального документа и фискального признака. Первые два уникальны в пределах накопителя, третий подтверждает, что документ не перепечатан.

create table if not exists ofd_receipt (
    fn          text          not null,
    fd          bigint        not null,
    fpd         text          not null,
    dt          timestamptz   not null,
    kkt         text          not null,
    operation   smallint      not null,
    total       numeric(12,2) not null,
    cash        numeric(12,2) not null default 0,
    card        numeric(12,2) not null default 0,
    loaded_at   timestamptz   not null default now(),
    primary key (fn, fd)
);

Запись — одним insert ... on conflict do update: если чек уже есть, обновляем суммы и время загрузки. Тогда любой прогон можно повторять сколько угодно, а «перезалить период» перестаёт быть опасной операцией.

import psycopg2.extras

SQL = """
insert into ofd_receipt (fn, fd, fpd, dt, kkt, operation, total, cash, card)
values %s
on conflict (fn, fd) do update set
    total = excluded.total,
    cash = excluded.cash,
    card = excluded.card,
    loaded_at = now()
"""


def save(conn, frame):
    rows = [tuple(r) for r in frame[[
        'fn', 'fd', 'fpd', 'dt', 'kkt', 'operation', 'total', 'cash', 'card'
    ]].itertuples(index=False, name=None)]
    with conn.cursor() as cur:
        psycopg2.extras.execute_values(cur, SQL, rows, page_size=500)
    conn.commit()

Ограничения, о которых узнаёшь на второй неделе

Лимит на отчёты. Кабинеты ограничивают число выгрузок и глубину периода: где‑то не даёт больше месяца за раз, где‑то — больше одной выгрузки в несколько минут. Если долбить интерфейс в цикле, прилетит либо капча, либо временная блокировка. Лечится паузами между запросами и разбивкой периода.

Двухфакторка. Включённое подтверждение по SMS ломает любой автоматический вход. Рабочий вариант — отдельный пользователь кабинета с правами только на просмотр, без 2FA, и сохранённый профиль браузера, чтобы сессия жила неделями.

Вёрстка. Селекторы по CSS‑классам живут до ближайшего редизайна. Я держу в коде поиск по видимому тексту кнопки и по data-testid, а не по цепочке div > div > span.

Часовой пояс. Кабинет отдаёт время кассы, а не сервера. Если у точек разные регионы, «вчера» у каждой своё — границы периода надо считать по каждой кассе отдельно, иначе на стыке суток теряются чеки.

Разные ОФД в одной сети. У сети из десяти точек операторы могут быть разными: колонки называются по‑разному, номера касс могут совпадать. Приводить к единой схеме нужно на входе, до записи в базу, иначе склейка превращается в ручную работу.

Что в итоге

Автоматическая выгрузка на одну небольшую сеть — это около 400 строк кода: браузер, два источника данных, нормализация, база и cron. Раз в сутки скрипт проверяет новые чеки, пишет строки в базу и складывает CSV в папку; если что‑то падает — приходит уведомление, и это видно в тот же день, а не в день сдачи отчёта.

Когда этого делать не надо: если выгрузка нужна раз в квартал — хватит ручного экспорта из кабинета, полчаса работы. Если у оператора есть нормальный API и он доступен по заявке — идите в API, браузер тут лишний слой. Selenium оправдан ровно в одном случае: данные нужны регулярно, а доступа, кроме интерфейса, нет.

Отдельно про позиции чеков: если нужна номенклатура, готовьтесь к тому, что запросов будет в разы больше — по одному на документ или пачками с лимитом. Это меняет архитектуру: появляется очередь заданий, и время выполнения считается уже не минутами, а десятками минут.