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