food-market/scripts/migration/delta_sync.py

#!/usr/bin/env python3
"""Дельта-досинхронизация МойСклад → ALL MARKET (Part 5). Запускать раз в N минут
(cron / watchdog) до дня перехода. Только ЧТЕНИЕ МойСклада.

Логика:
  1. Берёт lastSync (МСК) из файла состояния (при первом запуске — аргумент --since).
  2. Для каждого типа документов дергает фоновый импорт с filter=updated>=lastSync
     (эндпоинт POST /api/admin/moysklad/import-supplies?type=..&updatedFrom=..),
     последовательно (не параллельно — одна орг, гонка остатков), поллит job до конца.
  3. Удаления: /audit?filter=eventType=delete;moment>=lastSync — логируем (список id;
     ручное удаление в ALL MARKET на дне перехода, т.к. автоудаление документов с
     движениями рискованно и выполняется по регламенту рунбука).
  4. Обновляет lastSync = момент старта прогона (МСК), чтобы следующий прогон взял
     только новые изменения. Перекрытие в 2 мин для надёжности на границе.

Идемпотентность гарантируется мигратором (skip по MoySkladExternalId), поэтому
перекрытие окна не создаёт дублей.
"""
import os, json, time, ssl, sys, urllib.request, urllib.parse, urllib.error
from datetime import datetime, timedelta, timezone

AM = os.environ.get("AM_URL", "https://test.admin.food-market.kz")
STATE = os.environ.get("DELTA_STATE", os.path.expanduser("~/.fm-watchdog/migration-delta-state.json"))
USER = os.environ.get("AM_USER", "foodmarket-migr@food-market.local")
PASS = os.environ.get("AM_PASS", "Migrate12345!")
MSTOK = [l.split("=", 1)[1].strip() for l in open(os.path.expanduser("~/.food-market-secrets/moysklad.env")) if l.startswith("MOYSKLAD_TOKEN=")][0]
MS = "https://api.moysklad.ru/api/remap/1.2"
_ctx = ssl.create_default_context(); _ctx.check_hostname = False; _ctx.verify_mode = ssl.CERT_NONE
MSK = timezone(timedelta(hours=3))

# Документные типы в порядке безопасного переноса (справочники отдельно не трогаем в
# дельте — их меняют редко; при необходимости добавить products/counterparties).
DOC_TYPES = ["supply", "enter", "loss", "purchasereturn", "retaildemand", "retailsalesreturn", "inventory", "money"]
# МойСкладовские entityType для аудита удалений (money раскрывается на под-типы ниже).
AUDIT_ENTITIES = ["supply", "enter", "loss", "purchasereturn", "retaildemand", "retailsalesreturn",
                  "inventory", "paymentout", "paymentin", "cashin", "cashout",
                  "retaildrawercashin", "retaildrawercashout"]


def _get(url, hdr):
    for i in range(4):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=hdr), timeout=130, context=_ctx) as r:
                raw = r.read()
                import gzip
                if r.headers.get("Content-Encoding") == "gzip":
                    raw = gzip.decompress(raw)
                return json.loads(raw.decode("utf-8"))
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(2); continue
            if e.code >= 500:
                time.sleep(2 * (i + 1)); continue
            raise
        except Exception:
            time.sleep(2 * (i + 1))
    return None


def ms(path):
    return _get(MS + path, {"Authorization": f"Bearer {MSTOK}", "Accept-Encoding": "gzip"})


def am_token():
    data = urllib.parse.urlencode({"grant_type": "password", "username": USER, "password": PASS,
                                   "client_id": "food-market-web", "scope": "openid profile email roles api"}).encode()
    req = urllib.request.Request(AM + "/connect/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded"})
    with urllib.request.urlopen(req, timeout=60, context=_ctx) as r:
        return json.loads(r.read())["access_token"]


class DeltaDisabled(Exception):
    """Флаг подгрузки выключен суперадмином (эндпоинт ответил 409)."""


def get_delta_status(tok):
    """Спрашивает у API, включена ли подгрузка. Источник правды — PlatformSettings,
    которым управляет суперадмин. При «Выкл» скрипт выходит, не трогая lastSync."""
    req = urllib.request.Request(AM + "/api/admin/moysklad/delta-sync",
                                 headers={"Authorization": f"Bearer {tok}"})
    with urllib.request.urlopen(req, timeout=60, context=_ctx) as r:
        return json.loads(r.read())


def report_run(tok, started_utc, ok, migrated, error):
    """Пишет итог прогона в PlatformSettings, чтобы суперадмин видел его на
    странице «Настройки платформы». Ошибку самого отчёта глотаем — она не должна
    ронять прогон."""
    try:
        body = json.dumps({
            "startedAt": started_utc.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "finishedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "ok": bool(ok), "migrated": int(migrated),
            "error": (error[:2000] if error else None),
        }).encode("utf-8")
        req = urllib.request.Request(AM + "/api/admin/moysklad/delta-sync/report", data=body,
                                     headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}, method="POST")
        urllib.request.urlopen(req, timeout=60, context=_ctx).read()
    except Exception as e:
        print(f"  [report] не удалось записать итог прогона: {e}")


def trigger(tok, doc_type, updated_from):
    # delta=true помечает запрос как дельта-подгрузку: при выключенном флаге
    # сервер ответит 409 (защита даже при ручном запуске скрипта).
    q = urllib.parse.urlencode({"type": doc_type, "updatedFrom": updated_from, "delta": "true"})
    req = urllib.request.Request(AM + "/api/admin/moysklad/import-supplies?" + q, data=b"{}",
                                 headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=60, context=_ctx) as r:
            return json.loads(r.read())["jobId"]
    except urllib.error.HTTPError as e:
        if e.code == 409:
            raise DeltaDisabled()
        raise


def poll(tok, job_id, timeout_s=1800):
    t0 = time.time()
    while time.time() - t0 < timeout_s:
        d = _get(AM + f"/api/admin/moysklad/jobs/{job_id}", {"Authorization": f"Bearer {tok}"})
        if d and d.get("status") in ("Succeeded", "Failed"):
            # повторное чтение: финальный flush счётчиков (Created/Skipped) мог отстать
            # на один тик от статуса — читаем ещё раз, чтобы цифры были точными.
            time.sleep(2)
            d2 = _get(AM + f"/api/admin/moysklad/jobs/{job_id}", {"Authorization": f"Bearer {tok}"})
            return d2 or d
        time.sleep(5)
    return {"status": "Timeout"}


def load_state():
    try:
        return json.load(open(STATE))
    except Exception:
        return {}


def save_state(st):
    os.makedirs(os.path.dirname(STATE), exist_ok=True)
    json.dump(st, open(STATE, "w"), ensure_ascii=False, indent=1)


def main():
    since = None
    if "--since" in sys.argv:
        since = sys.argv[sys.argv.index("--since") + 1]
    st = load_state()
    tok = am_token()

    # Главный флаг: источник правды — PlatformSettings (суперадмин). При «Выкл» —
    # одна строка в логе и выход; lastSync НЕ трогаем, чтобы при повторном включении
    # первый прогон догнал пропущенное.
    try:
        status = get_delta_status(tok)
    except Exception as e:
        print(f"Не удалось получить флаг подгрузки у API: {e}")
        sys.exit(1)
    if not status.get("enabled", False):
        print("Подгрузка из МойСклад выключена (флаг Выкл) — выходим, lastSync не трогаем.")
        return

    last = since or st.get("lastSync")
    if not last:
        print("Нет lastSync. Укажи --since 'YYYY-MM-DD HH:MM:SS' (МСК) при первом запуске.")
        sys.exit(2)
    # перекрытие 2 минуты на границе окна (идемпотентность защищает от дублей)
    try:
        last_dt = datetime.strptime(last, "%Y-%m-%d %H:%M:%S") - timedelta(minutes=2)
        last = last_dt.strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        pass
    started_utc = datetime.now(timezone.utc)
    run_start = datetime.now(MSK).strftime("%Y-%m-%d %H:%M:%S")
    print(f"=== Дельта-синхронизация: изменения с {last} (МСК), старт {run_start} ===")

    total_created = 0
    ok = False
    error_text = None
    try:
        for dt in DOC_TYPES:
            jid = trigger(tok, dt, last)
            res = poll(tok, jid)
            # эндпоинт сериализует ключи в camelCase (created/skipped/total/message)
            created = res.get("created", 0) or 0
            skipped = res.get("skipped", 0) or 0
            total_created += created
            print(f"  [{dt}] {res.get('status')}: создано {created}, пропущено {skipped}. {res.get('message','')[:140]}")

        # удаления (audit). filter целиком экранируем; при 4xx/недоступности — мягко пропускаем.
        deleted = []
        try:
            flt = urllib.parse.quote(f"eventType=delete;moment>={last}")
            audit = ms(f"/audit?limit=100&filter={flt}")
        except Exception:
            audit = None
        if audit is None:
            print("  [deletions] /audit недоступен (нет прав токена) — проверить удаления вручную на дне перехода.")
        else:
            for row in audit.get("rows", []):
                if row.get("entityType") in AUDIT_ENTITIES:
                    deleted.append({"entityType": row.get("entityType"), "moment": row.get("moment"), "info": row.get("info")})
            print(f"  [deletions] событий удаления документов с {last}: {len(deleted)}")
            for d in deleted[:20]:
                print(f"     - {d['entityType']} @ {d['moment']}: {d['info']}")

        st["lastSync"] = run_start
        st["lastRun"] = datetime.now(MSK).strftime("%Y-%m-%d %H:%M:%S")
        st["lastCreated"] = total_created
        st["lastDeletions"] = deleted
        save_state(st)
        ok = True
        print(f"=== Готово. Перенесено изменений: {total_created}. lastSync → {run_start}. Состояние: {STATE} ===")
    except DeltaDisabled:
        # Флаг выключили посреди прогона — выходим чисто, lastSync не трогаем.
        print("Подгрузка выключена во время прогона (409) — выходим, lastSync не трогаем.")
        return
    except Exception as e:
        error_text = str(e)
        print(f"ОШИБКА прогона: {e}")
    finally:
        report_run(tok, started_utc, ok, total_created, error_text)
    if not ok:
        sys.exit(1)


if __name__ == "__main__":
    main()