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"]


def trigger(tok, doc_type, updated_from):
    q = urllib.parse.urlencode({"type": doc_type, "updatedFrom": updated_from})
    req = urllib.request.Request(AM + "/api/admin/moysklad/import-supplies?" + q, data=b"{}",
                                 headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}, method="POST")
    with urllib.request.urlopen(req, timeout=60, context=_ctx) as r:
        return json.loads(r.read())["jobId"]


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()
    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
    run_start = datetime.now(MSK).strftime("%Y-%m-%d %H:%M:%S")
    print(f"=== Дельта-синхронизация: изменения с {last} (МСК), старт {run_start} ===")
    tok = am_token()

    total_created = 0
    for dt in DOC_TYPES:
        jid = trigger(tok, dt, last)
        res = poll(tok, jid)
        created = res.get("Created", 0) or 0
        total_created += created
        print(f"  [{dt}] {res.get('status')}: создано {created}, пропущено {res.get('Skipped', 0)}. {res.get('Message','')[:120]}")

    # удаления (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)
    print(f"=== Готово. Перенесено изменений: {total_created}. lastSync → {run_start}. Состояние: {STATE} ===")


if __name__ == "__main__":
    main()