#!/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()