1648 lines
69 KiB
Python
1648 lines
69 KiB
Python
#!/usr/bin/env python3
|
||
|
||
import gzip
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
import threading
|
||
|
||
import wesp_runtime_env
|
||
|
||
wesp_runtime_env.apply_kiosk_headless_env()
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Callable, Dict, List, Optional
|
||
|
||
import requests
|
||
|
||
from config import (
|
||
get_config_class,
|
||
read_sync_client_state,
|
||
resolve_effective_sync_role,
|
||
resolve_sync_server_url,
|
||
write_sync_client_state,
|
||
)
|
||
|
||
_CLS = get_config_class()
|
||
CLIENT_VERSION = getattr(_CLS, "SYNC_CLIENT_VERSION", "1.4.0")
|
||
MIN_SERVER_VERSION = getattr(_CLS, "SYNC_CLIENT_MIN_SERVER_VERSION", "1.2.0")
|
||
MAX_SERVER_VERSION = getattr(_CLS, "SYNC_CLIENT_MAX_SERVER_VERSION", "1.4.0")
|
||
|
||
|
||
def _sync_trace_log(msg: str) -> None:
|
||
"""Диагностика sync: WARNING на root — видно при WESP_LOG_LEVEL=WARNING (INFO часто не пишется в файл)."""
|
||
logging.getLogger().warning(msg)
|
||
|
||
|
||
_SYNC_APPLY_ERR_DETAIL_LIMIT = 12
|
||
|
||
|
||
def _truncate_for_sync_log(s: str, max_len: int = 320) -> str:
|
||
t = (s or "").replace("\n", " ").strip()
|
||
if len(t) <= max_len:
|
||
return t
|
||
return t[: max_len - 1] + "…"
|
||
|
||
|
||
def _pull_changes_stats(changes: List[Dict[str, Any]]) -> str:
|
||
if not changes:
|
||
return "{}"
|
||
stats: Dict[str, int] = {}
|
||
for c in changes:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
key = f"{c.get('table_name', '?')}.{c.get('action', '?')}"
|
||
stats[key] = stats.get(key, 0) + 1
|
||
s = str(stats)
|
||
return s[:400] + ("…" if len(s) > 400 else "")
|
||
|
||
|
||
def _preview_pull_task_ids(changes: List[Dict[str, Any]], limit: int = 5) -> str:
|
||
if not changes:
|
||
return "—"
|
||
parts: List[str] = []
|
||
for c in changes[:limit]:
|
||
if isinstance(c, dict):
|
||
tid = str(c.get("task_id") or c.get("id") or "")[:8]
|
||
parts.append(tid or "?")
|
||
suffix = "…" if len(changes) > limit else ""
|
||
return ",".join(parts) + suffix
|
||
|
||
|
||
def _applied_changes_stats(changes: List[Dict[str, Any]], failed_ids: List[str]) -> str:
|
||
"""Сводка по строкам, ушедшим в commit (не в failed_ids)."""
|
||
failed_set = {str(x).strip() for x in failed_ids if x}
|
||
stats: Dict[str, int] = {}
|
||
for ch in changes:
|
||
if not isinstance(ch, dict):
|
||
continue
|
||
tid = str(ch.get("id") or ch.get("task_id") or "").strip()
|
||
if tid in failed_set:
|
||
continue
|
||
key = f"{ch.get('table_name', '?')}.{ch.get('action', '?')}"
|
||
stats[key] = stats.get(key, 0) + 1
|
||
s = str(stats)
|
||
return s[:400] + ("…" if len(s) > 400 else "")
|
||
|
||
|
||
def _preview_failed_task_ids(failed_ids: List[str], limit: int = 8) -> str:
|
||
if not failed_ids:
|
||
return "—"
|
||
return ",".join(str(x)[:8] for x in failed_ids[:limit]) + ("…" if len(failed_ids) > limit else "")
|
||
|
||
|
||
REPORT_PUSH_TABLES = frozenset(
|
||
{
|
||
"loading_report",
|
||
"unloading_report",
|
||
"loading_report_component",
|
||
"component_loading_time",
|
||
"unloading_report_group",
|
||
}
|
||
)
|
||
|
||
|
||
def client_push_tables() -> frozenset:
|
||
"""Таблицы, которые полевой узел (role=client) отправляет на сервер через /api/sync/push."""
|
||
from app.services.sync_manager import SERVER_MASTER_TABLES
|
||
|
||
return frozenset(SERVER_MASTER_TABLES) | REPORT_PUSH_TABLES
|
||
|
||
|
||
def _push_task_sort_key(task: Any) -> tuple:
|
||
"""FK-порядок как в SNAPSHOT_MODELS: component → recipe → ingredient → …"""
|
||
from app.services.sync_manager import SNAPSHOT_TABLE_ORDER
|
||
|
||
created = getattr(task, "created_at", None)
|
||
return (
|
||
SNAPSHOT_TABLE_ORDER.get(getattr(task, "table_name", ""), 999),
|
||
-(int(getattr(task, "priority", 0) or 0)),
|
||
created if created is not None else 0,
|
||
)
|
||
|
||
|
||
class SyncClient:
|
||
"""API-only sync client (transport layer).
|
||
|
||
This client does not import local ORM models and does not read/write
|
||
application databases directly. It only talks to sync HTTP endpoints.
|
||
|
||
Настройки: класс Config (config.py) и переменные WESP_*; изменяемые поля
|
||
(client_id, server_url после API) — в sync_client_state.json (см. config.py).
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.logger = logging.getLogger(__name__)
|
||
|
||
self._cfg = get_config_class()
|
||
state = read_sync_client_state()
|
||
|
||
# Порядок: WESP_SYNC_ROLE → sync_client_state.json → Config (см. resolve_effective_sync_role).
|
||
self.role, self._role_locked_by_env = resolve_effective_sync_role(
|
||
state=state, config_defaults=self._cfg
|
||
)
|
||
self._role_from_env = self._role_locked_by_env
|
||
self._role_from_state = bool(str(state.get("role") or "").strip()) and not self._role_locked_by_env
|
||
|
||
self.server_url, self._sync_server_url_explicit = resolve_sync_server_url(
|
||
state=state, config_defaults=self._cfg
|
||
)
|
||
|
||
client_id = os.getenv("WESP_SYNC_CLIENT_ID", "").strip()
|
||
if not client_id:
|
||
client_id = str(state.get("client_id") or "").strip()
|
||
if not client_id:
|
||
client_id = str(uuid.uuid4())
|
||
write_sync_client_state({"client_id": client_id})
|
||
self.client_id = client_id
|
||
|
||
client_name = os.getenv("WESP_SYNC_CLIENT_NAME", "").strip()
|
||
if not client_name:
|
||
cn = state.get("client_name")
|
||
client_name = str(cn).strip() if cn else ""
|
||
if not client_name:
|
||
client_name = getattr(self._cfg, "SYNC_CLIENT_NAME", "Sync Client") or "Sync Client"
|
||
self.client_name = client_name
|
||
self._flask_app: Optional[Any] = None
|
||
|
||
if (
|
||
str(self.role or "").strip().lower() != "client"
|
||
and not self._role_locked_by_env
|
||
and str(self.server_url or "").strip()
|
||
and str(state.get("client_id") or "").strip()
|
||
):
|
||
self.logger.info(
|
||
"sync: узел с server_url и client_id — работа как client (роль в state: %r)",
|
||
self.role,
|
||
)
|
||
self.role = "client"
|
||
|
||
_steady = int(
|
||
os.getenv(
|
||
"WESP_SYNC_CLIENT_POLL_INTERVAL_SEC",
|
||
str(getattr(self._cfg, "SYNC_CLIENT_POLL_INTERVAL_SEC", 30)),
|
||
)
|
||
)
|
||
self.sync_interval = _steady
|
||
self._poll_interval_steady = _steady
|
||
self._poll_interval_initial = int(
|
||
os.getenv(
|
||
"WESP_SYNC_CLIENT_POLL_INTERVAL_INITIAL_SEC",
|
||
str(getattr(self._cfg, "SYNC_CLIENT_POLL_INTERVAL_INITIAL_SEC", _steady)),
|
||
)
|
||
)
|
||
self.request_timeout = int(
|
||
os.getenv(
|
||
"WESP_SYNC_CLIENT_REQUEST_TIMEOUT_SEC",
|
||
str(getattr(self._cfg, "SYNC_CLIENT_REQUEST_TIMEOUT_SEC", 15)),
|
||
)
|
||
)
|
||
self.is_running = False
|
||
self.sync_thread: Optional[threading.Thread] = None
|
||
|
||
self.config = self._build_config_dict(state)
|
||
|
||
# Колбэк: применить изменения с pull. Может вернуть список id задач, которые не удалось применить
|
||
# (не подтверждаются на сервере — повторная выдача).
|
||
self.on_remote_changes: Optional[
|
||
Callable[[List[Dict[str, Any]]], Optional[List[str]]]
|
||
] = None
|
||
# Колбэк: обработать результат локального push (например, обновить local sync_queue).
|
||
self.on_local_push_result: Optional[
|
||
Callable[[List[Dict[str, Any]], Dict[str, Any]], None]
|
||
] = None
|
||
|
||
raw_auto = os.getenv("WESP_CLIENT_LOG_AUTO_UPLOAD", "").strip()
|
||
if raw_auto:
|
||
self.log_auto_upload = raw_auto.lower() not in ("0", "false", "no", "off")
|
||
else:
|
||
self.log_auto_upload = bool(getattr(self._cfg, "WESP_CLIENT_LOG_AUTO_UPLOAD", True))
|
||
self.log_auto_interval_sec = max(
|
||
3600,
|
||
int(
|
||
os.getenv(
|
||
"WESP_CLIENT_LOG_AUTO_INTERVAL_SEC",
|
||
str(getattr(self._cfg, "WESP_CLIENT_LOG_AUTO_INTERVAL_SEC", 86400)),
|
||
)
|
||
),
|
||
)
|
||
self.log_auto_timeout_sec = max(
|
||
15,
|
||
int(
|
||
os.getenv(
|
||
"WESP_CLIENT_LOG_AUTO_UPLOAD_TIMEOUT_SEC",
|
||
str(getattr(self._cfg, "WESP_CLIENT_LOG_AUTO_UPLOAD_TIMEOUT_SEC", 120)),
|
||
)
|
||
),
|
||
)
|
||
self._log_upload_lock = threading.Lock()
|
||
self._cycle_lock = threading.Lock()
|
||
self._last_log_upload_at = str(state.get("last_client_log_upload_at") or "").strip()
|
||
self._pull_busy_retries = int(
|
||
getattr(self._cfg, "SYNC_CLIENT_PULL_BUSY_RETRIES", 12)
|
||
)
|
||
self._pull_busy_delay = float(
|
||
getattr(self._cfg, "SYNC_CLIENT_PULL_BUSY_DELAY_SEC", 0.05)
|
||
)
|
||
self._pull_network_retries = int(
|
||
getattr(self._cfg, "SYNC_CLIENT_PULL_NETWORK_RETRIES", 4)
|
||
)
|
||
self._pull_network_base_delay = float(
|
||
getattr(self._cfg, "SYNC_CLIENT_PULL_NETWORK_BASE_DELAY_SEC", 0.8)
|
||
)
|
||
self._pull_network_max_delay = float(
|
||
getattr(self._cfg, "SYNC_CLIENT_PULL_NETWORK_MAX_DELAY_SEC", 8.0)
|
||
)
|
||
_fb = state.get("first_bootstrap_done")
|
||
if _fb is None:
|
||
_fb = getattr(self._cfg, "SYNC_FIRST_BOOTSTRAP_DONE", False)
|
||
self._initial_sync_active = not bool(_fb)
|
||
self._defer_first_background_cycle = False
|
||
self._last_pull_meta: Dict[str, Any] = {}
|
||
self._last_pull_integrity_ok = True
|
||
self._last_cycle_apply_failed = False
|
||
self._progress_lock = threading.Lock()
|
||
self._initial_sync_applied_tasks = 0
|
||
self._initial_sync_started_monotonic: Optional[float] = None
|
||
self._last_logged_percent_bucket = -1
|
||
|
||
def _current_poll_interval_sec(self) -> int:
|
||
if self._initial_sync_active:
|
||
return max(1, int(self._poll_interval_initial))
|
||
return max(1, int(self._poll_interval_steady))
|
||
|
||
def _effective_pull_limit(self) -> int:
|
||
if self._initial_sync_active:
|
||
return max(1, int(getattr(self._cfg, "SYNC_CLIENT_PULL_LIMIT_INITIAL", 50)))
|
||
return max(1, int(getattr(self._cfg, "SYNC_CLIENT_DEFAULT_PULL_LIMIT", 100)))
|
||
|
||
def _apply_remote_changes_batched(self, pulled: List[Dict[str, Any]]) -> set[str]:
|
||
"""Возвращает id задач (sync_queue), которые не нужно подтверждать: ошибка apply или commit."""
|
||
failed: set[str] = set()
|
||
if not pulled or self.on_remote_changes is None:
|
||
return failed
|
||
|
||
def _task_ids(batch: List[Dict[str, Any]]) -> set[str]:
|
||
ids: set[str] = set()
|
||
for change in batch:
|
||
tid = change.get("id") or change.get("task_id")
|
||
if tid is not None and str(tid).strip():
|
||
ids.add(str(tid).strip())
|
||
return ids
|
||
|
||
def _merge(ret: Any) -> None:
|
||
if ret is None:
|
||
return
|
||
if isinstance(ret, (list, tuple, set)):
|
||
for x in ret:
|
||
if x is not None and str(x).strip():
|
||
failed.add(str(x).strip())
|
||
|
||
chunk = int(getattr(self._cfg, "SYNC_CLIENT_APPLY_CHUNK_SIZE", 0) or 0)
|
||
if chunk <= 0:
|
||
try:
|
||
_merge(self.on_remote_changes(pulled))
|
||
except Exception as exc:
|
||
self.logger.error(
|
||
"sync: apply remote changes failed: %s", exc, exc_info=True
|
||
)
|
||
failed.update(_task_ids(pulled))
|
||
return failed
|
||
for i in range(0, len(pulled), chunk):
|
||
batch = pulled[i : i + chunk]
|
||
try:
|
||
_merge(self.on_remote_changes(batch))
|
||
except Exception as exc:
|
||
self.logger.error(
|
||
"sync: apply remote changes failed: %s", exc, exc_info=True
|
||
)
|
||
failed.update(_task_ids(batch))
|
||
if i + chunk < len(pulled):
|
||
time.sleep(0)
|
||
return failed
|
||
|
||
def _build_config_dict(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
||
cfg = self._cfg
|
||
pinned = os.getenv("WESP_SYNC_PINNED_SERVER_ID", "").strip()
|
||
if not pinned:
|
||
pinned = str(state.get("pinned_server_id") or "").strip()
|
||
if not pinned:
|
||
pinned = getattr(cfg, "SYNC_PINNED_SERVER_ID", "") or ""
|
||
|
||
fb = state.get("first_bootstrap_done")
|
||
if fb is None:
|
||
fb = getattr(cfg, "SYNC_FIRST_BOOTSTRAP_DONE", False)
|
||
|
||
return {
|
||
"role": self.role,
|
||
"server_url": self.server_url,
|
||
"sync_server_url_explicit": bool(self._sync_server_url_explicit),
|
||
"client_id": self.client_id,
|
||
"client_name": self.client_name,
|
||
"version": getattr(cfg, "SYNC_CLIENT_VERSION", CLIENT_VERSION),
|
||
"sync_interval": self.sync_interval,
|
||
"request_timeout": self.request_timeout,
|
||
"version_constraints": {
|
||
"min_server_version": getattr(cfg, "SYNC_CLIENT_MIN_SERVER_VERSION", MIN_SERVER_VERSION),
|
||
"max_server_version": getattr(cfg, "SYNC_CLIENT_MAX_SERVER_VERSION", MAX_SERVER_VERSION),
|
||
},
|
||
"sync_compression": {
|
||
"enabled": getattr(cfg, "SYNC_CLIENT_COMPRESSION_ENABLED", True),
|
||
"algorithm": getattr(cfg, "SYNC_CLIENT_COMPRESSION_ALGORITHM", "gzip"),
|
||
"level": int(getattr(cfg, "SYNC_CLIENT_COMPRESSION_LEVEL", 6)),
|
||
"min_size": int(getattr(cfg, "SYNC_CLIENT_COMPRESSION_MIN_SIZE", 0)),
|
||
"min_ratio": float(getattr(cfg, "SYNC_CLIENT_COMPRESSION_MIN_RATIO", 0)),
|
||
"max_size": int(getattr(cfg, "SYNC_CLIENT_COMPRESSION_MAX_SIZE", 10 * 1024 * 1024)),
|
||
"force_compression": getattr(cfg, "SYNC_CLIENT_COMPRESSION_FORCE", True),
|
||
},
|
||
"first_bootstrap_done": bool(fb),
|
||
"pinned_server_id": pinned,
|
||
"updated_at": datetime.now().isoformat(),
|
||
}
|
||
|
||
def _effective_pull_timeout(self) -> int:
|
||
"""До завершения первой синхронизации — длиннее (снапшот + ожидание lock на сервере)."""
|
||
steady = max(
|
||
self.request_timeout,
|
||
int(getattr(self._cfg, "SYNC_CLIENT_PULL_HTTP_TIMEOUT_SEC", 300)),
|
||
)
|
||
if not self._initial_sync_active:
|
||
return steady
|
||
initial = max(
|
||
self.request_timeout,
|
||
int(getattr(self._cfg, "SYNC_CLIENT_INITIAL_PULL_HTTP_TIMEOUT_SEC", 600)),
|
||
)
|
||
return max(steady, initial)
|
||
|
||
def _merge_pull_progress_meta(self, body: Any) -> None:
|
||
if not isinstance(body, dict):
|
||
return
|
||
with self._progress_lock:
|
||
if "remaining_hint" in body:
|
||
self._last_pull_meta["remaining_hint"] = body.get("remaining_hint")
|
||
if "initial_sync_active" in body:
|
||
self._last_pull_meta["initial_sync_active_server"] = bool(
|
||
body.get("initial_sync_active")
|
||
)
|
||
|
||
def _reset_initial_sync_progress_counters(self) -> None:
|
||
with self._progress_lock:
|
||
self._initial_sync_applied_tasks = 0
|
||
self._initial_sync_started_monotonic = None
|
||
self._last_logged_percent_bucket = -1
|
||
|
||
def _bump_initial_sync_progress(self, locally_applied_ok_count: int) -> None:
|
||
"""Учитывает только успешно применённые к локальной БД задачи (не размер пачки pull)."""
|
||
if not self._initial_sync_active or locally_applied_ok_count <= 0:
|
||
return
|
||
with self._progress_lock:
|
||
if self._initial_sync_started_monotonic is None:
|
||
self._initial_sync_started_monotonic = time.monotonic()
|
||
self._initial_sync_applied_tasks += locally_applied_ok_count
|
||
rem = self._last_pull_meta.get("remaining_hint")
|
||
applied = self._initial_sync_applied_tasks
|
||
total = None
|
||
pct = None
|
||
if isinstance(rem, int) and rem >= 0:
|
||
total = applied + rem
|
||
if total > 0:
|
||
pct = min(100.0, 100.0 * float(applied) / float(total))
|
||
elapsed = (
|
||
time.monotonic() - self._initial_sync_started_monotonic
|
||
if self._initial_sync_started_monotonic is not None
|
||
else 0.0
|
||
)
|
||
tps = applied / elapsed if elapsed > 0.5 else None
|
||
bucket = int((pct or 0) // 10) if pct is not None else -1
|
||
if pct is not None and bucket > self._last_logged_percent_bucket:
|
||
self._last_logged_percent_bucket = bucket
|
||
eta_txt = "—"
|
||
if isinstance(rem, int) and rem >= 0 and tps and tps > 0:
|
||
eta_sec = rem / tps
|
||
eta_txt = f"{int(eta_sec)} с" if eta_sec < 120 else f"{eta_sec / 60:.1f} мин"
|
||
self.logger.info(
|
||
"sync: первая синхронизация ~%.1f%% (%s/%s применено локально, оценка по серверу), ~%.1f задач/с, осталось ~%s",
|
||
pct,
|
||
applied,
|
||
total if total is not None else "?",
|
||
(tps or 0.0),
|
||
eta_txt,
|
||
)
|
||
|
||
def snapshot_initial_sync_progress(self) -> Dict[str, Any]:
|
||
"""Снимок для админки: %, скорость, ETA. tasks_applied — успешно применённые к локальной БД, не размер pull."""
|
||
if str(self.role or "").strip().lower() != "client":
|
||
return {"active": False}
|
||
with self._progress_lock:
|
||
if not self._initial_sync_active:
|
||
return {"active": False}
|
||
applied = self._initial_sync_applied_tasks
|
||
rem = self._last_pull_meta.get("remaining_hint")
|
||
rem_i = int(rem) if isinstance(rem, int) else None
|
||
total = (applied + rem_i) if rem_i is not None else None
|
||
pct = None
|
||
if total is not None and total > 0:
|
||
pct = min(100.0, 100.0 * float(applied) / float(total))
|
||
elapsed = 0.0
|
||
if self._initial_sync_started_monotonic is not None:
|
||
elapsed = max(0.0, time.monotonic() - self._initial_sync_started_monotonic)
|
||
tps = (applied / elapsed) if elapsed > 0.5 else None
|
||
eta_sec = None
|
||
if rem_i is not None and rem_i >= 0 and tps and tps > 0:
|
||
eta_sec = float(rem_i) / float(tps)
|
||
return {
|
||
"active": True,
|
||
"tasks_applied": applied,
|
||
"tasks_remaining_hint": rem_i,
|
||
"tasks_total_estimate": total,
|
||
"percent": round(pct, 2) if pct is not None else None,
|
||
"tasks_per_sec": round(tps, 2) if tps is not None else None,
|
||
"elapsed_sec": round(elapsed, 1),
|
||
"eta_sec": round(eta_sec, 1) if eta_sec is not None else None,
|
||
}
|
||
|
||
def _store_server_now(self, body: Any) -> None:
|
||
if not isinstance(body, dict):
|
||
return
|
||
server_now = str(body.get("server_now") or body.get("timestamp") or "").strip()
|
||
if server_now:
|
||
write_sync_client_state({"last_server_now": server_now})
|
||
|
||
def _local_initial_sync_complete(self) -> bool:
|
||
"""Клиент завершает первую синхронизацию только когда сервер и локальный pull согласны."""
|
||
server_done = not bool(self._last_pull_meta.get("initial_sync_active_server", True))
|
||
has_more = bool(self._last_pull_meta.get("has_more"))
|
||
if not server_done or has_more:
|
||
return False
|
||
if self._last_cycle_apply_failed:
|
||
return False
|
||
return True
|
||
|
||
def _apply_initial_sync_from_body(self, body: Any) -> None:
|
||
"""Сервер отдаёт initial_sync_active; при false — фиксируем first_bootstrap_done в состоянии узла."""
|
||
if isinstance(body, dict):
|
||
self._store_server_now(body)
|
||
if not isinstance(body, dict) or "initial_sync_active" not in body:
|
||
return
|
||
if bool(body.get("initial_sync_active")):
|
||
return
|
||
if not self._initial_sync_active:
|
||
return
|
||
if not self._local_initial_sync_complete():
|
||
self.logger.info(
|
||
"sync: сервер сообщил initial_sync_active=false, но локально ещё не завершено "
|
||
"(has_more=%s apply_failed=%s)",
|
||
self._last_pull_meta.get("has_more"),
|
||
self._last_cycle_apply_failed,
|
||
)
|
||
return
|
||
self._initial_sync_active = False
|
||
self._reset_initial_sync_progress_counters()
|
||
now = datetime.now().isoformat()
|
||
write_sync_client_state({"first_bootstrap_done": True, "updated_at": now})
|
||
self.config["first_bootstrap_done"] = True
|
||
self.config["updated_at"] = now
|
||
self.logger.info(
|
||
"sync: первая синхронизация завершена (initial_sync_active=false от сервера)"
|
||
)
|
||
|
||
def _with_retry(self, fn, max_attempts: int = 3, base_delay: float = 1.0):
|
||
last_exc = None
|
||
for attempt in range(1, max_attempts + 1):
|
||
try:
|
||
return fn()
|
||
except Exception as e: # pragma: no cover - network/runtime path
|
||
last_exc = e
|
||
if attempt < max_attempts:
|
||
time.sleep(base_delay * (2 ** (attempt - 1)))
|
||
raise last_exc
|
||
|
||
def _request_pull_with_busy_retry(
|
||
self, payload: Dict[str, Any]
|
||
) -> Dict[str, Any]:
|
||
"""POST /api/sync/pull with separate busy and network retry policies."""
|
||
retries = max(1, self._pull_busy_retries)
|
||
delay = max(0.0, self._pull_busy_delay)
|
||
extra = {"X-WESP-Initial-Sync": "1"} if self._initial_sync_active else None
|
||
last: Dict[str, Any] = {}
|
||
for attempt in range(retries):
|
||
last = self._request_json(
|
||
"POST",
|
||
"/api/sync/pull",
|
||
payload,
|
||
timeout_sec=self._effective_pull_timeout(),
|
||
extra_headers=extra,
|
||
retry_profile="pull_network",
|
||
)
|
||
code = int(last.get("status_code") or 0)
|
||
if code == 202:
|
||
body = last.get("body") or {}
|
||
retry_after = int(body.get("retry_after_sec") or 2)
|
||
if attempt < retries - 1:
|
||
time.sleep(max(1, retry_after))
|
||
continue
|
||
return last
|
||
if code in (409, 429) and attempt < retries - 1:
|
||
time.sleep(delay * (1.0 + attempt * 0.2))
|
||
continue
|
||
return last
|
||
return last
|
||
|
||
def _request_json(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
payload: Optional[Dict[str, Any]] = None,
|
||
allow_gzip_outgoing: bool = False,
|
||
*,
|
||
timeout_sec: Optional[int] = None,
|
||
extra_headers: Optional[Dict[str, str]] = None,
|
||
retry_profile: str = "default",
|
||
) -> Dict[str, Any]:
|
||
url = f"{self.server_url}{path}"
|
||
method_upper = method.upper()
|
||
to = int(timeout_sec) if timeout_sec is not None else self.request_timeout
|
||
|
||
def do_request():
|
||
hdrs: Dict[str, str] = dict(extra_headers or {})
|
||
if method_upper == "GET":
|
||
response = requests.get(url, headers=hdrs, timeout=to)
|
||
else:
|
||
request_kwargs: Dict[str, Any] = {
|
||
"timeout": to,
|
||
"headers": hdrs,
|
||
}
|
||
if payload is not None:
|
||
if allow_gzip_outgoing and self._compression_enabled():
|
||
raw = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
compressed = gzip.compress(raw, compresslevel=self._compression_level())
|
||
hdrs["Content-Encoding"] = "gzip"
|
||
request_kwargs["data"] = compressed
|
||
else:
|
||
request_kwargs["json"] = payload
|
||
response = requests.post(url, **request_kwargs)
|
||
|
||
try:
|
||
body = response.json() if response.content else {}
|
||
except Exception:
|
||
body = {}
|
||
code = int(response.status_code or 0)
|
||
if retry_profile == "pull_network" and code in (502, 503, 504):
|
||
raise requests.RequestException(f"transient HTTP {code}")
|
||
return code, body
|
||
|
||
if retry_profile == "pull_network":
|
||
status_code, body = self._with_network_retry(do_request)
|
||
else:
|
||
status_code, body = self._with_retry(do_request)
|
||
return {"status_code": status_code, "body": body}
|
||
|
||
def _with_network_retry(self, fn):
|
||
last_exc = None
|
||
attempts = max(1, int(self._pull_network_retries))
|
||
base = max(0.05, float(self._pull_network_base_delay))
|
||
max_delay = max(base, float(self._pull_network_max_delay))
|
||
for attempt in range(1, attempts + 1):
|
||
try:
|
||
return fn()
|
||
except (requests.Timeout, requests.ConnectionError, requests.RequestException) as e:
|
||
last_exc = e
|
||
if attempt >= attempts:
|
||
break
|
||
backoff = min(max_delay, base * (2 ** (attempt - 1)))
|
||
jitter = backoff * 0.25 * random.random()
|
||
time.sleep(backoff + jitter)
|
||
raise last_exc
|
||
|
||
def _log_sync_http_fail(self, op: str, result: Dict[str, Any]) -> None:
|
||
"""Пишем тело ответа сервера — иначе при 500 в логе только «HTTP 500»."""
|
||
code = int(result.get("status_code") or 0)
|
||
raw = result.get("body")
|
||
detail = ""
|
||
if isinstance(raw, dict):
|
||
detail = str(
|
||
raw.get("message")
|
||
or raw.get("error")
|
||
or raw.get("detail")
|
||
or ""
|
||
).strip()
|
||
if not detail and raw is not None:
|
||
detail = str(raw)[:1200]
|
||
if code in (409, 429):
|
||
log_fn = self.logger.info
|
||
elif code == 503:
|
||
log_fn = self.logger.warning
|
||
elif code >= 500:
|
||
log_fn = self.logger.error
|
||
else:
|
||
log_fn = self.logger.warning
|
||
log_fn(
|
||
"sync: %s → HTTP %s %s",
|
||
op,
|
||
code,
|
||
(detail[:1500] if detail else "—"),
|
||
)
|
||
|
||
def _log_pull_received(self, code: int, body: Dict[str, Any]) -> None:
|
||
"""Сводка тела ответа pull — сопоставлять с [SYNC-PULL-OUT] на сервере."""
|
||
try:
|
||
try:
|
||
approx = len(json.dumps(body, ensure_ascii=False))
|
||
except Exception:
|
||
approx = 0
|
||
ch = body.get("changes")
|
||
n = len(ch) if isinstance(ch, list) else 0
|
||
stats = _pull_changes_stats(ch) if isinstance(ch, list) else "{}"
|
||
bp = body.get("bootstrap_progress")
|
||
bp_s = "—"
|
||
if isinstance(bp, dict):
|
||
bp_s = f"{bp.get('phase')}:{bp.get('cursor')}/{bp.get('total_models')}"
|
||
cid = (self.client_id or "")[:12]
|
||
msg = (
|
||
"[SYNC-PULL-IN] client=%s… http=%s approx_json_bytes=%s batch_id=%s total=%s "
|
||
"has_more=%s remaining=%s initial_sync=%s retry_after=%s bootstrap=%s task_ids=%s stats=%s"
|
||
% (
|
||
cid,
|
||
code,
|
||
approx,
|
||
str(body.get("batch_id") or "")[:32],
|
||
body.get("total", n),
|
||
body.get("has_more"),
|
||
body.get("remaining_hint"),
|
||
body.get("initial_sync_active"),
|
||
body.get("retry_after_sec"),
|
||
bp_s,
|
||
_preview_pull_task_ids(ch) if isinstance(ch, list) else "—",
|
||
stats,
|
||
)
|
||
)
|
||
_sync_trace_log(msg)
|
||
except Exception as exc:
|
||
logging.getLogger().warning(
|
||
"[SYNC-PULL-IN] client=%s… (сводка не построена: %s)",
|
||
(self.client_id or "")[:12],
|
||
exc,
|
||
)
|
||
|
||
def _compression_enabled(self) -> bool:
|
||
return bool(getattr(self._cfg, "SYNC_CLIENT_COMPRESSION_ENABLED", True))
|
||
|
||
def _compression_level(self) -> int:
|
||
level = int(getattr(self._cfg, "SYNC_CLIENT_COMPRESSION_LEVEL", 6))
|
||
if level < 1:
|
||
return 1
|
||
if level > 9:
|
||
return 9
|
||
return level
|
||
|
||
def _sync_transport_configured(self) -> bool:
|
||
return bool((self.server_url or "").strip())
|
||
|
||
def save_client_id(self) -> None:
|
||
self.config["client_id"] = self.client_id
|
||
write_sync_client_state({"client_id": self.client_id})
|
||
|
||
def update_server_url(self, new_url: str) -> None:
|
||
normalized = new_url.rstrip("/")
|
||
self.server_url = normalized
|
||
self._sync_server_url_explicit = True
|
||
self.config["server_url"] = normalized
|
||
self.config["sync_server_url_explicit"] = True
|
||
self.config["updated_at"] = datetime.now().isoformat()
|
||
write_sync_client_state(
|
||
{
|
||
"server_url": normalized.rstrip("/") + "/",
|
||
"updated_at": self.config["updated_at"],
|
||
}
|
||
)
|
||
|
||
def _pull_payload(self, limit: int) -> Dict[str, Any]:
|
||
p: Dict[str, Any] = {"client_id": self.client_id, "limit": limit}
|
||
cn = (self.client_name or "").strip()
|
||
if cn:
|
||
p["client_name"] = cn[:100]
|
||
return p
|
||
|
||
def register(self) -> bool:
|
||
"""Lightweight bootstrap using sync API only.
|
||
|
||
Server side auto-registers client during /api/sync/pull processing.
|
||
"""
|
||
if not self._sync_transport_configured():
|
||
return False
|
||
payload = self._pull_payload(1)
|
||
try:
|
||
result = self._request_pull_with_busy_retry(payload)
|
||
except Exception as exc: # pragma: no cover - network path
|
||
from app.services.sync_error_display import log_sync_transport_problem
|
||
|
||
log_sync_transport_problem(
|
||
self.logger,
|
||
"bootstrap/pull",
|
||
exc,
|
||
server_url=self.server_url,
|
||
)
|
||
return False
|
||
code = int(result.get("status_code") or 0)
|
||
body = result.get("body")
|
||
if isinstance(body, dict):
|
||
self._apply_initial_sync_from_body(body)
|
||
if code not in (200, 202):
|
||
self._log_sync_http_fail("bootstrap/pull", result)
|
||
return code in (200, 202)
|
||
|
||
def pull_changes(self, limit: int = 100) -> List[Dict[str, Any]]:
|
||
if not self._sync_transport_configured():
|
||
return []
|
||
payload = self._pull_payload(limit)
|
||
try:
|
||
result = self._request_pull_with_busy_retry(payload)
|
||
except Exception as exc: # pragma: no cover - network path
|
||
from app.services.sync_error_display import log_sync_transport_problem
|
||
|
||
log_sync_transport_problem(
|
||
self.logger,
|
||
"pull",
|
||
exc,
|
||
server_url=self.server_url,
|
||
)
|
||
self._last_pull_integrity_ok = False
|
||
return []
|
||
code = int(result.get("status_code") or 0)
|
||
raw_body = result.get("body")
|
||
body: Dict[str, Any] = raw_body if isinstance(raw_body, dict) else {}
|
||
self._last_pull_meta = {
|
||
"status_code": code,
|
||
"has_more": bool(body.get("has_more")),
|
||
"batch_id": str(body.get("batch_id") or ""),
|
||
"retry_after_sec": int(body.get("retry_after_sec") or 0),
|
||
"server_now": str(body.get("server_now") or body.get("timestamp") or ""),
|
||
"initial_sync_active_server": bool(body.get("initial_sync_active"))
|
||
if "initial_sync_active" in body
|
||
else self._last_pull_meta.get("initial_sync_active_server"),
|
||
}
|
||
if code == 202:
|
||
self._last_pull_integrity_ok = True
|
||
self._merge_pull_progress_meta(body)
|
||
self._store_server_now(body)
|
||
self._log_pull_received(code, body)
|
||
return []
|
||
if code != 200:
|
||
self._log_sync_http_fail("pull", result)
|
||
self._last_pull_integrity_ok = False
|
||
return []
|
||
self._apply_initial_sync_from_body(body)
|
||
self._merge_pull_progress_meta(body)
|
||
self._store_server_now(body)
|
||
self._log_pull_received(code, body)
|
||
changes = body.get("changes", [])
|
||
if not isinstance(changes, list):
|
||
self._last_pull_integrity_ok = False
|
||
return []
|
||
declared_total = body.get("total")
|
||
if isinstance(declared_total, int) and declared_total != len(changes):
|
||
self._last_pull_integrity_ok = False
|
||
self.logger.warning(
|
||
"sync: pull batch integrity mismatch total=%s len(changes)=%s batch_id=%s",
|
||
declared_total,
|
||
len(changes),
|
||
self._last_pull_meta.get("batch_id") or "—",
|
||
)
|
||
return []
|
||
self._last_pull_integrity_ok = True
|
||
return changes
|
||
|
||
def confirm_tasks(self, task_ids: List[str]) -> bool:
|
||
if not task_ids:
|
||
return True
|
||
if not self._sync_transport_configured():
|
||
return False
|
||
payload = {"client_id": self.client_id, "task_ids": task_ids}
|
||
attempts = max(1, int(getattr(self._cfg, "SYNC_CLIENT_CONFIRM_NETWORK_RETRIES", 5)))
|
||
base = max(0.0, float(self._pull_network_base_delay))
|
||
max_delay = max(base, float(self._pull_network_max_delay))
|
||
last: Dict[str, Any] = {}
|
||
for attempt in range(attempts):
|
||
last = self._request_json(
|
||
"POST",
|
||
"/api/sync/confirm",
|
||
payload,
|
||
retry_profile="pull_network",
|
||
)
|
||
cbody = last.get("body")
|
||
if isinstance(cbody, dict):
|
||
self._store_server_now(cbody)
|
||
if "initial_sync_active" in cbody:
|
||
with self._progress_lock:
|
||
server_initial = bool(cbody.get("initial_sync_active"))
|
||
self._last_pull_meta["initial_sync_active_server"] = server_initial
|
||
if not server_initial:
|
||
self._last_pull_meta["has_more"] = False
|
||
self._apply_initial_sync_from_body(cbody)
|
||
code = int(last.get("status_code") or 0)
|
||
if code == 200:
|
||
return True
|
||
if attempt < attempts - 1 and code in (500, 502, 503, 504):
|
||
backoff = min(max_delay, base * (2**attempt))
|
||
jitter = backoff * 0.25 * random.random()
|
||
time.sleep(backoff + jitter)
|
||
continue
|
||
break
|
||
self._log_sync_http_fail("confirm", last)
|
||
return False
|
||
|
||
def push_changes(self, changes: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
if not self._sync_transport_configured():
|
||
return {
|
||
"success": False,
|
||
"error": True,
|
||
"message": "server_url не задан — синхронизация отключена",
|
||
}
|
||
payload = {"client_id": self.client_id, "changes": changes or []}
|
||
result = self._request_json(
|
||
"POST",
|
||
"/api/sync/push",
|
||
payload,
|
||
allow_gzip_outgoing=True,
|
||
)
|
||
status_code = result["status_code"]
|
||
body = result["body"] or {}
|
||
if status_code in (200, 201):
|
||
return body if isinstance(body, dict) else {"success": True}
|
||
self._log_sync_http_fail("push", result)
|
||
if status_code == 409:
|
||
if isinstance(body, dict):
|
||
body.setdefault("success", False)
|
||
return body
|
||
return {"success": False, "error": True, "message": "Sync conflict"}
|
||
return {
|
||
"success": False,
|
||
"error": True,
|
||
"message": (body.get("message") if isinstance(body, dict) else "Sync push failed"),
|
||
"status_code": status_code,
|
||
}
|
||
|
||
def default_local_log_path(self) -> str:
|
||
"""Путь к локальному лог-файлу на узле (env WESP_CLIENT_LOCAL_LOG_PATH или WESP_ADMIN_LOG_PATH)."""
|
||
p = os.getenv("WESP_CLIENT_LOCAL_LOG_PATH", "").strip()
|
||
if p:
|
||
return os.path.abspath(os.path.expanduser(p))
|
||
cfg_path = getattr(self._cfg, "WESP_ADMIN_LOG_PATH", "") or ""
|
||
return os.path.abspath(os.path.expanduser(str(cfg_path))) if cfg_path else ""
|
||
|
||
def upload_client_log_bytes(
|
||
self,
|
||
data: bytes,
|
||
*,
|
||
remote_name: str = "wesp.log",
|
||
timeout: Optional[int] = None,
|
||
) -> Dict[str, Any]:
|
||
"""Отправить сырые байты лога на сервер (gzip), отдельно от sync БД."""
|
||
if not self._sync_transport_configured():
|
||
return {"success": False, "status_code": 0, "body": {"message": "server_url не задан"}}
|
||
url = f"{self.server_url}/api/sync/client-log"
|
||
compressed = gzip.compress(data)
|
||
headers: Dict[str, str] = {
|
||
"X-WESP-Client-Id": self.client_id,
|
||
"X-WESP-Log-Name": remote_name,
|
||
"Content-Encoding": "gzip",
|
||
"Content-Type": "application/octet-stream",
|
||
}
|
||
secret = os.getenv("WESP_CLIENT_LOG_UPLOAD_SECRET", "").strip()
|
||
if secret:
|
||
headers["X-WESP-Client-Log-Secret"] = secret
|
||
|
||
to = int(timeout) if timeout is not None else self.request_timeout
|
||
|
||
def do_request():
|
||
response = requests.post(url, data=compressed, headers=headers, timeout=to)
|
||
try:
|
||
body = response.json() if response.content else {}
|
||
except Exception:
|
||
body = {}
|
||
return response.status_code, body
|
||
|
||
status_code, body = self._with_retry(do_request)
|
||
ok = status_code == 200 and isinstance(body, dict) and body.get("success")
|
||
return {"success": ok, "status_code": status_code, "body": body}
|
||
|
||
def upload_client_log_file(
|
||
self,
|
||
log_path: str,
|
||
*,
|
||
remote_name: Optional[str] = None,
|
||
timeout: Optional[int] = None,
|
||
) -> Dict[str, Any]:
|
||
"""Отправить файл лога с диска на сервер."""
|
||
path = os.path.abspath(os.path.expanduser(log_path))
|
||
with open(path, "rb") as f:
|
||
data = f.read()
|
||
name = remote_name or os.path.basename(path)
|
||
return self.upload_client_log_bytes(data, remote_name=name, timeout=timeout)
|
||
|
||
def _is_client_log_auto_upload_due(self) -> bool:
|
||
if not self.log_auto_upload or self.role != "client":
|
||
return False
|
||
if not self._sync_transport_configured():
|
||
return False
|
||
path = self.default_local_log_path()
|
||
if not path or not os.path.isfile(path):
|
||
return False
|
||
if not self._last_log_upload_at:
|
||
return True
|
||
try:
|
||
last = datetime.fromisoformat(self._last_log_upload_at.replace("Z", "+00:00"))
|
||
if last.tzinfo is None:
|
||
last = last.replace(tzinfo=timezone.utc)
|
||
last = last.astimezone(timezone.utc)
|
||
except Exception:
|
||
return True
|
||
elapsed = (datetime.now(timezone.utc) - last).total_seconds()
|
||
return elapsed >= self.log_auto_interval_sec
|
||
|
||
def _run_auto_client_log_upload_locked(self) -> None:
|
||
path = self.default_local_log_path()
|
||
if not path or not os.path.isfile(path):
|
||
return
|
||
try:
|
||
result = self.upload_client_log_file(
|
||
path,
|
||
remote_name=os.path.basename(path),
|
||
timeout=self.log_auto_timeout_sec,
|
||
)
|
||
if result.get("success"):
|
||
now_iso = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||
write_sync_client_state({"last_client_log_upload_at": now_iso})
|
||
self._last_log_upload_at = now_iso
|
||
self.logger.info("Автовыгрузка лога на сервер выполнена (%s)", os.path.basename(path))
|
||
else:
|
||
self.logger.warning(
|
||
"Автовыгрузка лога не удалась: %s",
|
||
(result.get("body") or {}).get("message") or result.get("status_code"),
|
||
)
|
||
except OSError as exc:
|
||
self.logger.warning("Автовыгрузка лога: файл недоступен: %s", exc)
|
||
except Exception as exc: # pragma: no cover - сеть
|
||
self.logger.warning("Автовыгрузка лога: ошибка: %s", exc, exc_info=True)
|
||
|
||
def _maybe_schedule_auto_log_upload(self) -> None:
|
||
"""Раз в сутки (интервал из конфига), в отдельном потоке — не блокирует pull/push/confirm."""
|
||
if not self._is_client_log_auto_upload_due():
|
||
return
|
||
|
||
def worker() -> None:
|
||
if not self._log_upload_lock.acquire(blocking=False):
|
||
return
|
||
try:
|
||
if not self._is_client_log_auto_upload_due():
|
||
return
|
||
self._run_auto_client_log_upload_locked()
|
||
finally:
|
||
self._log_upload_lock.release()
|
||
|
||
threading.Thread(target=worker, daemon=True, name="wesp-client-log-auto-upload").start()
|
||
|
||
def collect_local_changes(self) -> List[Dict[str, Any]]:
|
||
"""Transport-only client does not read local DB.
|
||
|
||
Integrators can monkey-patch this method or assign a bound method that
|
||
returns local changes payload for /api/sync/push.
|
||
"""
|
||
return []
|
||
|
||
def apply_change(self, change: Dict[str, Any]) -> bool:
|
||
"""Transport-only client does not apply DB changes directly.
|
||
|
||
Integrators can process pulled changes via `on_remote_changes`.
|
||
"""
|
||
_ = change
|
||
return True
|
||
|
||
def sync_cycle(self) -> Dict[str, Any]:
|
||
result: Dict[str, Any] = {
|
||
"success": True,
|
||
"pushed": 0,
|
||
"pulled": 0,
|
||
"confirmed": 0,
|
||
"conflicts": 0,
|
||
}
|
||
|
||
with self._cycle_lock:
|
||
flask_app = getattr(self, "_flask_app", None)
|
||
if flask_app is not None:
|
||
with flask_app.app_context():
|
||
return self._sync_cycle_impl(result)
|
||
return self._sync_cycle_impl(result)
|
||
|
||
def _sync_cycle_impl(self, result: Dict[str, Any]) -> Dict[str, Any]:
|
||
self._last_cycle_apply_failed = False
|
||
local_changes = self.collect_local_changes()
|
||
if local_changes:
|
||
push_result = self.push_changes(local_changes)
|
||
if self.on_local_push_result is not None:
|
||
try:
|
||
self.on_local_push_result(local_changes, push_result)
|
||
except Exception as exc:
|
||
self.logger.error("sync: finalize local push failed: %s", exc, exc_info=True)
|
||
result["success"] = False
|
||
result["pushed"] = len(local_changes) if push_result.get("success") else 0
|
||
result["conflicts"] = int(push_result.get("total_conflicts", 0) or 0)
|
||
if push_result.get("error") and not push_result.get("success"):
|
||
result["success"] = False
|
||
|
||
max_rounds = max(1, int(getattr(self._cfg, "SYNC_CLIENT_PULL_BUSY_RETRIES", 12)))
|
||
total_pulled = 0
|
||
total_confirmed = 0
|
||
pull_limit = self._effective_pull_limit()
|
||
for _ in range(max_rounds):
|
||
pulled = self.pull_changes(pull_limit)
|
||
total_pulled += len(pulled)
|
||
apply_failed_ids: set[str] = set()
|
||
if pulled:
|
||
apply_failed_ids = self._apply_remote_changes_batched(pulled)
|
||
if apply_failed_ids:
|
||
self._last_cycle_apply_failed = True
|
||
n_ok = len(pulled) - len(apply_failed_ids)
|
||
if n_ok > 0:
|
||
self._bump_initial_sync_progress(n_ok)
|
||
if apply_failed_ids and self._initial_sync_active:
|
||
self.logger.warning(
|
||
"sync: первая синхронизация: %s из %s задач не применены к локальной БД (см. ERROR выше по каждой задаче)",
|
||
len(apply_failed_ids),
|
||
len(pulled),
|
||
)
|
||
|
||
task_ids = [
|
||
c.get("id") or c.get("task_id")
|
||
for c in pulled
|
||
if c.get("id") or c.get("task_id")
|
||
]
|
||
to_confirm = [
|
||
tid
|
||
for tid in task_ids
|
||
if tid and str(tid) not in apply_failed_ids
|
||
]
|
||
if len(task_ids) > len(to_confirm):
|
||
self.logger.error(
|
||
"sync: не подтверждено %s задач (ошибка записи в локальную БД) — сервер выдаст их снова",
|
||
len(task_ids) - len(to_confirm),
|
||
)
|
||
if to_confirm and self._last_pull_integrity_ok and self.confirm_tasks(to_confirm):
|
||
total_confirmed += len(to_confirm)
|
||
elif task_ids and not self._last_pull_integrity_ok:
|
||
self.logger.warning(
|
||
"sync: batch skipped (no confirm) due to integrity check failure, batch_id=%s",
|
||
self._last_pull_meta.get("batch_id") or "—",
|
||
)
|
||
has_more = bool(self._last_pull_meta.get("has_more"))
|
||
if not has_more and self._initial_sync_active:
|
||
self._apply_initial_sync_from_body(
|
||
{
|
||
"initial_sync_active": self._last_pull_meta.get(
|
||
"initial_sync_active_server", True
|
||
),
|
||
}
|
||
)
|
||
if not has_more:
|
||
break
|
||
time.sleep(max(0.05, self._pull_busy_delay))
|
||
result["pulled"] = total_pulled
|
||
result["confirmed"] = total_confirmed
|
||
|
||
self._maybe_schedule_auto_log_upload()
|
||
self.logger.debug(
|
||
"sync: цикл pushed=%s pulled=%s confirmed=%s conflicts=%s ok=%s",
|
||
result.get("pushed"),
|
||
result.get("pulled"),
|
||
result.get("confirmed"),
|
||
result.get("conflicts"),
|
||
result.get("success"),
|
||
)
|
||
return result
|
||
|
||
def _sync_loop(self, defer_first_cycle: bool = False) -> None:
|
||
"""defer_first_cycle: не дублировать pull сразу после register()+sync_cycle() в main."""
|
||
if defer_first_cycle:
|
||
time.sleep(self._current_poll_interval_sec())
|
||
while self.is_running:
|
||
try:
|
||
self.sync_cycle()
|
||
except Exception as e: # pragma: no cover - background loop protection
|
||
from app.services.sync_error_display import log_sync_transport_problem
|
||
|
||
log_sync_transport_problem(
|
||
self.logger,
|
||
"цикл",
|
||
e,
|
||
server_url=self.server_url,
|
||
)
|
||
time.sleep(self._current_poll_interval_sec())
|
||
|
||
def start_sync(self) -> None:
|
||
if self.is_running:
|
||
return
|
||
if not self._sync_transport_configured():
|
||
self.logger.warning(
|
||
"sync: фоновый цикл не запущен — server_url не задан (WESP_SYNC_SERVER_URL или sync_client_state.json)"
|
||
)
|
||
return
|
||
self.is_running = True
|
||
defer = self._defer_first_background_cycle
|
||
self._defer_first_background_cycle = False
|
||
self.sync_thread = threading.Thread(
|
||
target=self._sync_loop,
|
||
args=(defer,),
|
||
daemon=True,
|
||
name="wesp-sync-pull-push",
|
||
)
|
||
self.sync_thread.start()
|
||
self.logger.info(
|
||
"sync: фоновый цикл → %s интервал %ss (initial %ss пока первая синхронизация)",
|
||
self.server_url,
|
||
self._poll_interval_steady,
|
||
self._poll_interval_initial,
|
||
)
|
||
|
||
def start_sync_with_retry(self) -> bool:
|
||
"""Старт фонового цикла pull/push после успешного register (подключение к серверу)."""
|
||
try:
|
||
if self.config.get("role") != "client":
|
||
self.logger.debug("sync: role=%s — фоновый клиент не запускается", self.role)
|
||
return False
|
||
if not self._sync_transport_configured():
|
||
self.logger.warning(
|
||
"sync: role=client, но server_url не задан — фоновый sync не запускается. "
|
||
"Укажите WESP_SYNC_SERVER_URL или server_url в sync_client_state.json (например http://192.168.0.10 или http://komton_srv_1.local)."
|
||
)
|
||
return False
|
||
if not self.register():
|
||
self.logger.warning(
|
||
"sync: register не удался — к серверу не подключились, фоновый цикл не запускается"
|
||
)
|
||
return False
|
||
try:
|
||
# Первая полная итерация в потоке вызывающего — до фона, без гонки с циклом.
|
||
self.sync_cycle()
|
||
except Exception as cycle_exc: # pragma: no cover - startup path
|
||
from app.services.sync_error_display import log_sync_transport_problem
|
||
|
||
log_sync_transport_problem(
|
||
self.logger,
|
||
"стартовый цикл",
|
||
cycle_exc,
|
||
server_url=self.server_url,
|
||
)
|
||
# Фоновый поток не делает второй pull в ту же секунду (меньше конкуренции на сервере/SQLite).
|
||
self._defer_first_background_cycle = True
|
||
self.start_sync()
|
||
return True
|
||
except Exception as e: # pragma: no cover - startup path
|
||
from app.services.sync_error_display import log_sync_transport_problem
|
||
|
||
log_sync_transport_problem(
|
||
self.logger,
|
||
"старт",
|
||
e,
|
||
server_url=self.server_url,
|
||
)
|
||
return False
|
||
|
||
def stop_sync(self):
|
||
self.is_running = False
|
||
if self.sync_thread and self.sync_thread.is_alive():
|
||
self.sync_thread.join(timeout=5)
|
||
self.sync_thread = None
|
||
|
||
|
||
sync_client: Optional[SyncClient] = None
|
||
|
||
|
||
def _resolve_flask_app(app: Any) -> Any:
|
||
"""Реальный Flask app вместо current_app LocalProxy (для фоновых потоков sync)."""
|
||
get_obj = getattr(app, "_get_current_object", None)
|
||
if callable(get_obj):
|
||
return get_obj()
|
||
return app
|
||
|
||
|
||
def _attach_local_db_apply(client: SyncClient, app: Any) -> None:
|
||
"""Записывает выданные сервером задачи в локальные SQLite (роль client, процесс с Flask)."""
|
||
from app import db
|
||
from app.models import WESP_SUPPRESS_SYNC_ENQUEUE
|
||
from app.services.sync_manager import apply_sync_change
|
||
|
||
def apply_batch(changes: List[Dict[str, Any]]) -> List[str]:
|
||
"""Возвращает id задач sync_queue, которые не удалось применить (их не подтверждаем — сервер отдаст снова)."""
|
||
if not changes:
|
||
return []
|
||
all_ids = [
|
||
str(c.get("id") or c.get("task_id")).strip()
|
||
for c in changes
|
||
if c.get("id") or c.get("task_id")
|
||
]
|
||
with app.app_context():
|
||
failed_ids: List[str] = []
|
||
apply_err_details: List[tuple[str, str, str, str, str]] = []
|
||
db.session.info[WESP_SUPPRESS_SYNC_ENQUEUE] = True
|
||
try:
|
||
for ch in changes:
|
||
tid = ch.get("id") or ch.get("task_id")
|
||
stid = str(tid).strip() if tid else ""
|
||
tn = ch.get("table_name")
|
||
rid = ch.get("record_id")
|
||
action = str(ch.get("action") or "update").strip()
|
||
raw = ch.get("data")
|
||
data = raw if isinstance(raw, dict) else {}
|
||
if not tn or not rid:
|
||
err = "нет table_name или record_id в строке pull"
|
||
client.logger.error(
|
||
"sync: строка pull без table_name/record_id, пропуск: %s",
|
||
ch,
|
||
)
|
||
if stid:
|
||
failed_ids.append(stid)
|
||
apply_err_details.append(
|
||
(stid[:8], str(tn or "?"), action, str(rid or "")[:48], err)
|
||
)
|
||
continue
|
||
if action in ("create", "update") and not data:
|
||
err = (
|
||
"пустой data — на сервере нет строки или get_record_data_for_sync вернул None"
|
||
)
|
||
client.logger.error(
|
||
"sync: пустой data для %s.%s — на сервере нет строки или get_record_data_for_sync вернул None",
|
||
tn,
|
||
(rid or "")[:48],
|
||
)
|
||
if stid:
|
||
failed_ids.append(stid)
|
||
apply_err_details.append(
|
||
(stid[:8], str(tn), action, str(rid)[:48], err)
|
||
)
|
||
continue
|
||
res = apply_sync_change(str(tn), str(rid), action, data)
|
||
if not res.get("success"):
|
||
err = str(res.get("error") or "unknown")
|
||
if stid:
|
||
failed_ids.append(stid)
|
||
apply_err_details.append(
|
||
(stid[:8], str(tn), action, str(rid)[:48], err)
|
||
)
|
||
client.logger.error(
|
||
"sync: локальное применение не удалось %s %s: %s",
|
||
tn,
|
||
(rid or "")[:36],
|
||
res.get("error"),
|
||
)
|
||
try:
|
||
db.session.commit()
|
||
except Exception as exc:
|
||
db.session.rollback()
|
||
client.logger.error("sync: commit после применения pull: %s", exc, exc_info=True)
|
||
# Confirm только после успешного commit — иначе все task_ids failed.
|
||
return list(all_ids)
|
||
cid = (client.client_id or "")[:12]
|
||
ok_stats = _applied_changes_stats(changes, failed_ids)
|
||
if not failed_ids:
|
||
amsg = "[SYNC-APPLY-DB] client=%s… commit_ok задач=%s stats=%s" % (
|
||
cid,
|
||
len(changes),
|
||
ok_stats,
|
||
)
|
||
_sync_trace_log(amsg)
|
||
else:
|
||
amsg = (
|
||
"[SYNC-APPLY-DB] client=%s… commit_ok задач=%s ошибок_apply=%s stats_ok=%s failed_task_ids=%s"
|
||
% (
|
||
cid,
|
||
len(changes) - len(failed_ids),
|
||
len(failed_ids),
|
||
ok_stats,
|
||
_preview_failed_task_ids(failed_ids),
|
||
)
|
||
)
|
||
_sync_trace_log(amsg)
|
||
if apply_err_details:
|
||
shown = min(len(apply_err_details), _SYNC_APPLY_ERR_DETAIL_LIMIT)
|
||
for t8, tab, act, rec, emsg in apply_err_details[:shown]:
|
||
_sync_trace_log(
|
||
"[SYNC-APPLY-ERR] task=%s… table=%s action=%s record=%s err=%s"
|
||
% (
|
||
t8,
|
||
tab,
|
||
act,
|
||
rec,
|
||
_truncate_for_sync_log(emsg),
|
||
)
|
||
)
|
||
rest = len(apply_err_details) - shown
|
||
if rest > 0:
|
||
_sync_trace_log(
|
||
"[SYNC-APPLY-ERR] … ещё %s ошибок (всего с деталями в батче: %s)"
|
||
% (rest, len(apply_err_details))
|
||
)
|
||
return failed_ids
|
||
finally:
|
||
db.session.info.pop(WESP_SUPPRESS_SYNC_ENQUEUE, None)
|
||
db.session.remove()
|
||
|
||
client.on_remote_changes = apply_batch
|
||
|
||
|
||
def _attach_local_db_push(client: SyncClient, app: Any) -> None:
|
||
"""Собирает локальные задачи sync_queue (справочники + отчёты) и завершает их после push."""
|
||
from sqlalchemy import select
|
||
|
||
from app import db
|
||
from app.models import SyncQueue
|
||
from app.services.sync_manager import requeue_stuck_processing
|
||
from app.services.sync_record_data import get_record_data_for_sync
|
||
from app.timeutil import utc_now_naive
|
||
|
||
push_tables = client_push_tables()
|
||
|
||
def collect_batch() -> List[Dict[str, Any]]:
|
||
with app.app_context():
|
||
try:
|
||
requeue_stuck_processing(
|
||
timeout_minutes=int(getattr(client._cfg, "SYNC_REQUEUE_TIMEOUT_MINUTES", 15))
|
||
)
|
||
except Exception:
|
||
client.logger.exception("sync: requeue stuck local push tasks failed")
|
||
limit = max(
|
||
1,
|
||
int(
|
||
getattr(
|
||
client._cfg,
|
||
"SYNC_CLIENT_PUSH_LIMIT",
|
||
getattr(client._cfg, "SYNC_CLIENT_DEFAULT_PULL_LIMIT", 100),
|
||
)
|
||
),
|
||
)
|
||
now = utc_now_naive()
|
||
changes: List[Dict[str, Any]] = []
|
||
try:
|
||
tasks = (
|
||
db.session.execute(
|
||
select(SyncQueue)
|
||
.where(SyncQueue.status == "pending")
|
||
.where(SyncQueue.is_deleted.is_(False))
|
||
.where(SyncQueue.table_name.in_(tuple(sorted(push_tables))))
|
||
.order_by(SyncQueue.priority.desc(), SyncQueue.created_at.asc())
|
||
.limit(limit)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
tasks = sorted(tasks, key=_push_task_sort_key)
|
||
for task in tasks:
|
||
action = str(task.action or "update").strip()
|
||
record_data = get_record_data_for_sync(task.table_name, task.record_id)
|
||
if not record_data:
|
||
if action == "delete":
|
||
record_data = {}
|
||
else:
|
||
client.logger.warning(
|
||
"sync: local push skipped, no record data for %s.%s action=%s",
|
||
task.table_name,
|
||
(task.record_id or "")[:48],
|
||
action,
|
||
)
|
||
continue
|
||
task.status = "processing"
|
||
task.processed_at = now
|
||
task.error_message = None
|
||
changes.append(
|
||
{
|
||
"id": task.id,
|
||
"task_id": task.id,
|
||
"table_name": task.table_name,
|
||
"record_id": task.record_id,
|
||
"action": action,
|
||
"data": record_data,
|
||
}
|
||
)
|
||
if changes:
|
||
db.session.commit()
|
||
else:
|
||
db.session.rollback()
|
||
return changes
|
||
except Exception:
|
||
db.session.rollback()
|
||
client.logger.exception("sync: collect local push changes failed")
|
||
return []
|
||
finally:
|
||
db.session.remove()
|
||
|
||
def finalize_batch(changes: List[Dict[str, Any]], push_result: Dict[str, Any]) -> None:
|
||
task_ids = [
|
||
str(ch.get("task_id") or ch.get("id")).strip()
|
||
for ch in changes
|
||
if ch.get("task_id") or ch.get("id")
|
||
]
|
||
if not task_ids:
|
||
return
|
||
with app.app_context():
|
||
try:
|
||
rows = (
|
||
db.session.execute(select(SyncQueue).where(SyncQueue.id.in_(task_ids)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
now = utc_now_naive()
|
||
success = bool(push_result.get("success"))
|
||
msg = str(push_result.get("message") or "").strip()
|
||
for task in rows:
|
||
if success:
|
||
task.status = "completed"
|
||
task.completed_at = now
|
||
task.error_message = None
|
||
else:
|
||
task.status = "pending"
|
||
task.processed_at = None
|
||
task.completed_at = None
|
||
task.retry_count = min(
|
||
int(task.retry_count or 0) + 1,
|
||
int(task.max_retries or 3),
|
||
)
|
||
task.error_message = (msg[:500] if msg else None)
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
finally:
|
||
db.session.remove()
|
||
|
||
client.collect_local_changes = collect_batch
|
||
client.on_local_push_result = finalize_batch
|
||
|
||
|
||
def init_sync_client(app: Optional[Any] = None) -> None:
|
||
"""Если передан Flask app и role=client — pull-пачки пишутся в локальные recipes/reports.db."""
|
||
global sync_client
|
||
flask_app = _resolve_flask_app(app) if app is not None else None
|
||
if flask_app is not None:
|
||
with flask_app.app_context():
|
||
sync_client = SyncClient()
|
||
sync_client._flask_app = flask_app
|
||
else:
|
||
sync_client = SyncClient()
|
||
log = sync_client.logger
|
||
role = sync_client.config.get("role")
|
||
if role == "client" and flask_app is None:
|
||
log.error(
|
||
"sync: role=client, но init_sync_client() без Flask app — запись pull в локальную БД отключена "
|
||
"(нужен init_sync_client(app) из run.py/wsgi.py)."
|
||
)
|
||
if flask_app is not None and role == "client":
|
||
_attach_local_db_apply(sync_client, flask_app)
|
||
_attach_local_db_push(sync_client, flask_app)
|
||
uri = str(flask_app.config.get("SQLALCHEMY_DATABASE_URI") or "")
|
||
log.info(
|
||
"sync: применение pull в локальную SQLite включено (role=client). Основная БД: %s",
|
||
uri[:120] + ("…" if len(uri) > 120 else ""),
|
||
)
|
||
elif flask_app is not None and role != "client":
|
||
log.warning(
|
||
"sync: применение pull в локальную БД отключено: role=%r (ожидается role=client для полевой точки). "
|
||
"Порядок: WESP_SYNC_ROLE → data/sync_client_state.json → Config. "
|
||
"Сейчас env=%s state=%s.",
|
||
role,
|
||
getattr(sync_client, "_role_from_env", False),
|
||
getattr(sync_client, "_role_from_state", False),
|
||
)
|
||
if str(sync_client.role or "").strip().lower() == "client":
|
||
started = sync_client.start_sync_with_retry()
|
||
if flask_app is not None and not started and sync_client._sync_transport_configured():
|
||
_schedule_sync_client_connect_retry(flask_app)
|
||
|
||
|
||
def _should_schedule_sync_autostart(app) -> bool:
|
||
"""Не запускать sync в родительском процессе Flask reloader (он не слушает порт)."""
|
||
if bool(app.config.get("DEBUG", False)):
|
||
return os.environ.get("WERKZEUG_RUN_MAIN") == "true"
|
||
return True
|
||
|
||
|
||
def _role_for_sync_autostart(app) -> str:
|
||
"""Роль для autostart: client из state/env или полевая точка с server_url+client_id."""
|
||
role, _ = resolve_effective_sync_role(config_defaults=app.config)
|
||
role_s = str(role or "").strip().lower() or "server"
|
||
if role_s == "client":
|
||
return "client"
|
||
state = read_sync_client_state()
|
||
if str(state.get("server_url") or "").strip() and str(state.get("client_id") or "").strip():
|
||
return "client"
|
||
return role_s
|
||
|
||
|
||
def schedule_sync_client_autostart(app, delay_sec: float = 0.6) -> None:
|
||
"""Отложенный autostart: HTTP-сервер успевает открыть порт до register()/sync_cycle.
|
||
|
||
Синхронный init_sync_client до app.run() блокировал главный поток и давал запросы
|
||
на 127.0.0.1:5000 до bind — «подвисание» UI и лишние отказы.
|
||
"""
|
||
if app.config.get("TESTING") or not app.config.get("SYNC_CLIENT_AUTOSTART", True):
|
||
return
|
||
if not _should_schedule_sync_autostart(app):
|
||
return
|
||
role = _role_for_sync_autostart(app)
|
||
if role != "client":
|
||
logging.getLogger(__name__).info(
|
||
"sync autostart пропущен: роль %r (фоновый pull только для client)",
|
||
role,
|
||
)
|
||
return
|
||
|
||
def _run() -> None:
|
||
time.sleep(max(0.0, float(delay_sec)))
|
||
init_sync_client(app)
|
||
|
||
threading.Thread(target=_run, daemon=True, name="wesp-sync-autostart").start()
|
||
|
||
|
||
def _connect_retry_thread_running() -> bool:
|
||
return any(
|
||
t.name == "wesp-sync-connect-retry" and t.is_alive() for t in threading.enumerate()
|
||
)
|
||
|
||
|
||
def _schedule_sync_client_connect_retry(app, *, delay_sec: float = 30.0) -> None:
|
||
"""Повтор register/start, пока не подключимся к серверу (503, рестарт сервера и т.п.)."""
|
||
|
||
if _connect_retry_thread_running():
|
||
return
|
||
|
||
def _worker() -> None:
|
||
interval = max(5.0, float(delay_sec))
|
||
attempt = 0
|
||
log = logging.getLogger(__name__)
|
||
while True:
|
||
time.sleep(interval)
|
||
global sync_client
|
||
if sync_client is not None and sync_client.is_running:
|
||
return
|
||
with app.app_context():
|
||
if sync_client is None:
|
||
init_sync_client(app)
|
||
if sync_client is not None and sync_client.is_running:
|
||
return
|
||
continue
|
||
if str(sync_client.role or "").strip().lower() != "client":
|
||
return
|
||
if not sync_client._sync_transport_configured():
|
||
return
|
||
attempt += 1
|
||
log.info(
|
||
"sync: повторное подключение к %s (попытка %s)",
|
||
sync_client.server_url,
|
||
attempt,
|
||
)
|
||
if sync_client.start_sync_with_retry():
|
||
return
|
||
|
||
threading.Thread(target=_worker, daemon=True, name="wesp-sync-connect-retry").start()
|
||
|
||
|
||
def stop_sync_client():
|
||
global sync_client
|
||
if sync_client:
|
||
sync_client.stop_sync()
|
||
sync_client = None
|
||
|
||
|
||
def apply_sync_client_runtime(app: Any) -> bool:
|
||
"""После сохранения настроек в sync_client_state.json — перечитать файл и перезапустить фоновый sync в этом процессе.
|
||
|
||
Не вызывает перезапуск всего процесса (run.py). Возвращает False при TESTING, AUTOSTART=off или родителе Flask-reloader.
|
||
"""
|
||
if app.config.get("TESTING"):
|
||
return False
|
||
if not app.config.get("SYNC_CLIENT_AUTOSTART", True):
|
||
return False
|
||
if not _should_schedule_sync_autostart(app):
|
||
return False
|
||
stop_sync_client()
|
||
init_sync_client(app)
|
||
return True
|
||
|
||
|
||
def get_initial_sync_progress_snapshot() -> Dict[str, Any]:
|
||
"""Для /api/admin/summary на узле с ролью client."""
|
||
global sync_client
|
||
if sync_client is None:
|
||
return {"active": False}
|
||
return sync_client.snapshot_initial_sync_progress()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
client = SyncClient()
|
||
if client.config.get("role") == "client":
|
||
client.start_sync_with_retry()
|
||
try:
|
||
time.sleep(30)
|
||
finally:
|
||
client.stop_sync()
|