#!/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) # эндпоинт сериализует ключи в 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) print(f"=== Готово. Перенесено изменений: {total_created}. lastSync → {run_start}. Состояние: {STATE} ===") if __name__ == "__main__": main()