8be44055a2
- reconcile: выручка по документам (точно до копейки), plotseries как справка; чеки сходятся - delta_sync: camelCase ключи job (created/skipped), повторный опрос, audit tolerant — дельта подтверждена (новый чек из МС подтянут: created 1) - НАЙДЕНО: 30000 строк stocks без движений (прежний прямой сид) → пересобрать из движений Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
162 lines
8.4 KiB
Python
162 lines
8.4 KiB
Python
#!/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()
|