feat(infra): Python FastAPI backend + Docker compose for VPS deploy (GigaChat with Russian root CA)

backend-py/.dockerignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+.env
+.env.local
+credentials.json
+.venv/

backend-py/Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12-slim
+
+# НУЦ Минцифры root CA — для GigaChat SSL.
+# Скачиваем актуальный bundle на этапе сборки и добавляем в системный trust store.
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends ca-certificates curl \
+ && curl -fsSL -o /usr/local/share/ca-certificates/russian_trusted_root_ca.crt \
+        https://gu-st.ru/content/Other/doc/russian_trusted_root_ca.cer \
+ && curl -fsSL -o /usr/local/share/ca-certificates/russian_trusted_sub_ca.crt \
+        https://gu-st.ru/content/Other/doc/russian_trusted_sub_ca.cer \
+ && update-ca-certificates \
+ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app /app/app
+
+# httpx по умолчанию использует certifi → принудительно указываем системный bundle,
+# куда мы добавили НУЦ Минцифры
+ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
+ENV REQUESTS_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--proxy-headers", "--forwarded-allow-ips=*"]

backend-py/app/ai.py

@@ -0,0 +1,152 @@
+"""GigaChat client — OAuth + chat completions."""
+from __future__ import annotations
+import json
+import re
+import threading
+import time
+import uuid
+from typing import Any
+import httpx
+
+from .config import get_config
+
+_AUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
+_CHAT_URL = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions"
+
+_lock = threading.Lock()
+_token: str | None = None
+_token_expires_at: float = 0.0
+
+
+def _get_token() -> str:
+    global _token, _token_expires_at
+    with _lock:
+        # 5-минутный запас перед истечением
+        if _token and time.time() < _token_expires_at - 300:
+            return _token
+
+        cfg = get_config()
+        rq_uid = str(uuid.uuid4())
+        with httpx.Client(timeout=15.0) as client:
+            resp = client.post(
+                _AUTH_URL,
+                headers={
+                    "Authorization": f"Basic {cfg.gigachat_auth_key}",
+                    "RqUID": rq_uid,
+                    "Accept": "application/json",
+                },
+                data={"scope": cfg.gigachat_scope},
+            )
+        resp.raise_for_status()
+        data = resp.json()
+        _token = data.get("access_token") or data.get("tok")
+        if not _token:
+            raise RuntimeError(f"No access_token in GigaChat response: {data}")
+        # expires_at в миллисекундах unix
+        expires_at_ms = data.get("expires_at") or data.get("exp") or 0
+        _token_expires_at = (expires_at_ms / 1000) if expires_at_ms else (time.time() + 1500)
+        return _token
+
+
+SYSTEM_PROMPT_PICKER = (
+    "Ты — эксперт-консультант по подбору кухонной техники для фабрики мебели «ЗОВ».\n"
+    "Помогаешь менеджерам салонов согласовать с клиентом комплект техники.\n\n"
+    "Принципы подбора:\n"
+    "1. Уважай ценовой коридор. У каждой категории `price_ranges.{cat}.from..to` — попадай в него (±5%).\n"
+    "2. Уважай предпочтения по брендам: сначала preferred (★), потом acceptable (✓).\n"
+    "3. Учитывай инфраструктуру: газ исключает индукцию; нет вентиляции = только рециркуляция (угольный фильтр).\n"
+    "4. Учитывай приоритеты выбора (`priorities`): «цена/качество» → балансные модели; «отзывы» → проверенные хиты; «дизайн» → подбирай эстетику; «технологичность» → топовые фичи.\n"
+    "5. Если клиент явно отметил features в `per_cat.{cat}.features` — обязательно ставь модели с этими фичами.\n"
+    "6. ВАЖНО: каждую тех. фичу в highlights ОБЯЗАТЕЛЬНО объясняй простым языком в скобках.\n\n"
+    "Примеры пояснений:\n"
+    "  «NoFrost (не нужно размораживать вручную)»\n"
+    "  «PowerBoost (форсаж — кипятит за минуту)»\n"
+    "  «FlexZone (объединяет зоны под большую сковороду)»\n"
+    "  «4D HotAir (конвекция с 4 сторон — равномерное запекание)»\n"
+    "  «Термощуп (готовит до точной температуры)»\n"
+    "  «AquaStop (защита от протечек)»\n"
+    "  «Инвертор (тише и экономия ~30% электричества)»\n\n"
+    "Формат ответа — валидный JSON без markdown:\n"
+    "{\n"
+    '  "summary": "1-2 предложения общего вывода",\n'
+    '  "items": [{\n'
+    '    "category": "fridge",\n'
+    '    "brand": "Bosch",\n'
+    '    "model": "Serie 4 60см",\n'
+    '    "price_rub": 79990,\n'
+    '    "highlights": ["NoFrost (не нужно размораживать)", "Инвертор (тише и экономия ~30%)"],\n'
+    '    "caveats": "Глубина 660мм — на 60мм больше стандартной ниши",\n'
+    '    "match_score": 0.92,\n'
+    '    "tier_signal": "middle"\n'
+    "  }],\n"
+    '  "total_price_rub": 350000,\n'
+    '  "budget_status": "в_рамках|превышение|значительно_ниже",\n'
+    '  "client_temperature": "premium|middle|budget|mixed",\n'
+    '  "warnings": [],\n'
+    '  "next_steps": []\n'
+    "}\n\n"
+    "Не выдумывай несуществующие артикулы — указывай линейку (Bosch Serie 4 60см)."
+)
+
+
+def call_ai(user_prompt: str, system_prompt: str | None = None,
+            temperature: float = 0.3, max_tokens: int = 4000) -> dict[str, Any]:
+    """Вызов GigaChat. Возвращает {json, text, tokens, model, error?}."""
+    cfg = get_config()
+    try:
+        token = _get_token()
+    except Exception as e:
+        return {"json": None, "text": f"AI auth: {e}", "tokens": 0, "model": cfg.gigachat_model, "error": True}
+
+    payload = {
+        "model": cfg.gigachat_model,
+        "temperature": temperature,
+        "max_tokens": max_tokens,
+        "messages": [
+            {"role": "system", "content": system_prompt or SYSTEM_PROMPT_PICKER},
+            {"role": "user", "content": user_prompt},
+        ],
+    }
+
+    try:
+        with httpx.Client(timeout=60.0) as client:
+            resp = client.post(
+                _CHAT_URL,
+                headers={
+                    "Authorization": f"Bearer {token}",
+                    "Accept": "application/json",
+                    "Content-Type": "application/json",
+                },
+                content=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+            )
+    except Exception as e:
+        return {"json": None, "text": f"AI network: {e}", "tokens": 0, "model": cfg.gigachat_model, "error": True}
+
+    if resp.status_code >= 400:
+        try:
+            j = resp.json()
+            err_msg = j.get("message") or (j.get("error") or {}).get("message") or resp.text[:300]
+        except Exception:
+            err_msg = resp.text[:300]
+        return {"json": None, "text": f"AI HTTP {resp.status_code}: {err_msg}",
+                "tokens": 0, "model": cfg.gigachat_model, "error": True}
+
+    data = resp.json()
+    choice = (data.get("choices") or [{}])[0]
+    response_text = (choice.get("message") or {}).get("content", "")
+    tokens = (data.get("usage") or {}).get("total_tokens", 0)
+    actual_model = data.get("model", cfg.gigachat_model)
+
+    json_obj = None
+    if response_text:
+        try:
+            json_obj = json.loads(response_text)
+        except json.JSONDecodeError:
+            stripped = re.sub(r"^```(?:json)?\s*", "", response_text.strip())
+            stripped = re.sub(r"\s*```\s*$", "", stripped)
+            try:
+                json_obj = json.loads(stripped)
+            except json.JSONDecodeError:
+                pass
+
+    return {"json": json_obj, "text": response_text, "tokens": tokens, "model": actual_model}

backend-py/app/auth.py

@@ -0,0 +1,52 @@
+"""Telegram WebApp initData verification (HMAC-SHA-256)."""
+from __future__ import annotations
+import hmac
+import hashlib
+import json
+import time
+from typing import Any
+from urllib.parse import parse_qsl
+
+
+def verify_init_data(init_data: str, bot_token: str, max_age_sec: int = 86400) -> dict[str, Any] | None:
+    """
+    Проверяет подпись initData от Telegram WebApp.
+    Возвращает распарсенные данные с ключом 'user' (dict) или None при невалидной/просроченной подписи.
+
+    Спецификация: https://core.telegram.org/bots/webapps#validating-data-received-via-the-mini-app
+    """
+    if not init_data:
+        return None
+    parsed = dict(parse_qsl(init_data, keep_blank_values=True))
+    received_hash = parsed.pop("hash", None)
+    if not received_hash:
+        return None
+
+    # data_check_string: ключ=значение, отсортированы алфавитно, разделитель \n
+    data_check_string = "\n".join(f"{k}={parsed[k]}" for k in sorted(parsed))
+
+    # secret_key = HMAC-SHA-256(key="WebAppData", data=BOT_TOKEN)
+    secret_key = hmac.new(b"WebAppData", bot_token.encode(), hashlib.sha256).digest()
+    expected_hash = hmac.new(secret_key, data_check_string.encode(), hashlib.sha256).hexdigest()
+
+    if not hmac.compare_digest(expected_hash, received_hash):
+        return None
+
+    # Свежесть подписи (24 часа по умолчанию)
+    auth_date = int(parsed.get("auth_date", "0"))
+    if time.time() - auth_date > max_age_sec:
+        return None
+
+    user = None
+    if "user" in parsed:
+        try:
+            user = json.loads(parsed["user"])
+        except json.JSONDecodeError:
+            user = None
+
+    return {
+        "user": user,
+        "auth_date": auth_date,
+        "start_param": parsed.get("start_param"),
+        "chat_instance": parsed.get("chat_instance"),
+    }

backend-py/app/config.py

@@ -0,0 +1,42 @@
+"""Конфиг бэкенда — читается из переменных окружения."""
+from __future__ import annotations
+import os
+from dataclasses import dataclass
+from functools import lru_cache
+
+
+@dataclass(frozen=True)
+class Config:
+    bot_token: str
+    admin_tg_id: int
+    sheet_id: str
+    google_credentials_path: str
+
+    gigachat_auth_key: str
+    gigachat_model: str
+    gigachat_scope: str
+
+    active_period_days: int
+    grace_period_days: int
+
+
+def _required(name: str) -> str:
+    val = os.getenv(name)
+    if not val:
+        raise RuntimeError(f"Missing required env var: {name}")
+    return val
+
+
+@lru_cache(maxsize=1)
+def get_config() -> Config:
+    return Config(
+        bot_token=_required("BOT_TOKEN"),
+        admin_tg_id=int(os.getenv("ADMIN_TG_ID", "0")),
+        sheet_id=_required("SHEET_ID"),
+        google_credentials_path=os.getenv("GOOGLE_CREDENTIALS_PATH", "/app/credentials.json"),
+        gigachat_auth_key=_required("GIGACHAT_AUTH_KEY"),
+        gigachat_model=os.getenv("GIGACHAT_MODEL", "GigaChat-Pro"),
+        gigachat_scope=os.getenv("GIGACHAT_SCOPE", "GIGACHAT_API_PERS"),
+        active_period_days=int(os.getenv("ACTIVE_PERIOD_DAYS", "90")),
+        grace_period_days=int(os.getenv("GRACE_PERIOD_DAYS", "14")),
+    )

backend-py/app/main.py

@@ -0,0 +1,401 @@
+"""ЗОВ Backend — FastAPI app. Полный порт Apps Script Code.gs."""
+from __future__ import annotations
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+from fastapi import FastAPI, Request, Response
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from .config import get_config
+from .auth import verify_init_data
+from . import sheets, ai, telegram as tg
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+log = logging.getLogger("zov.backend")
+
+app = FastAPI(title="ZOV Backend", version="2.0")
+
+# CORS — MiniApp хостится на github.io, бэкенд на api.wasrusgen1.pro.
+# Простые запросы (text/plain или без Content-Type) не триггерят preflight.
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["GET", "POST", "OPTIONS"],
+    allow_headers=["*"],
+)
+
+
+# =================================================================
+# Health & ping
+# =================================================================
+
+@app.get("/healthz")
+async def healthz():
+    return {"ok": True, "service": "zov-tech-backend", "time": _now_iso()}
+
+
+@app.get("/")
+async def root():
+    return {
+        "status": "ok",
+        "service": "zov-tech-backend",
+        "version": "2.0",
+        "available_paths": ["me", "measurement", "podbor", "ping", "test_ai", "test_telegram", "seed_admin"],
+    }
+
+
+# =================================================================
+# Compatibility layer with Apps Script (?path=X)
+# =================================================================
+
+@app.post("/")
+async def root_post(request: Request):
+    return await _dispatch_post(request)
+
+
+@app.api_route("/exec", methods=["GET", "POST"])
+async def apps_script_compat(request: Request):
+    """Эмулируем поведение `/exec?path=X` чтобы старый MiniApp-код тоже работал."""
+    if request.method == "GET":
+        path = request.query_params.get("path", "")
+        if path == "ping":
+            return {"pong": True, "time": _now_iso()}
+        if path == "seed_admin":
+            return JSONResponse(_handle_seed_admin())
+        if path == "test_ai" or path == "test_claude":
+            return JSONResponse(_handle_test_ai())
+        if path == "test_telegram":
+            return JSONResponse(_handle_test_telegram())
+        return {"status": "ok", "service": "zov-tech-backend"}
+    return await _dispatch_post(request)
+
+
+async def _dispatch_post(request: Request):
+    path = request.query_params.get("path", "")
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+
+    handlers = {
+        "me":            _handle_me,
+        "measurement":   _handle_measurement,
+        "podbor":        _handle_podbor,
+        "ping":          lambda b: {"pong": True, "time": _now_iso()},
+        "seed_admin":    lambda b: _handle_seed_admin(),
+        "test_ai":       lambda b: _handle_test_ai(),
+        "test_claude":   lambda b: _handle_test_ai(),
+        "test_telegram": lambda b: _handle_test_telegram(),
+    }
+    fn = handlers.get(path)
+    if not fn:
+        return JSONResponse({"error": "unknown_path", "path": path}, status_code=404)
+
+    try:
+        return JSONResponse(fn(body))
+    except Exception as e:
+        log.exception("api error on path=%s", path)
+        sheets.log_event("api_error", None, {"path": path, "error": str(e)})
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+# =================================================================
+# Native /api/* routes (preferred for new MiniApp)
+# =================================================================
+
+@app.post("/api/me")
+async def api_me(request: Request):
+    body = await _safe_json(request)
+    return _handle_me(body)
+
+
+@app.post("/api/measurement")
+async def api_measurement(request: Request):
+    body = await _safe_json(request)
+    return _handle_measurement(body)
+
+
+@app.post("/api/podbor")
+async def api_podbor(request: Request):
+    body = await _safe_json(request)
+    return _handle_podbor(body)
+
+
+@app.get("/api/test_ai")
+async def api_test_ai():
+    return _handle_test_ai()
+
+
+@app.get("/api/test_telegram")
+async def api_test_telegram():
+    return _handle_test_telegram()
+
+
+@app.get("/api/seed_admin")
+async def api_seed_admin():
+    return _handle_seed_admin()
+
+
+# =================================================================
+# Handlers
+# =================================================================
+
+def _handle_me(body: dict[str, Any]) -> dict[str, Any]:
+    cfg = get_config()
+    init_data = body.get("initData") or ""
+    auth = verify_init_data(init_data, cfg.bot_token)
+    if not auth or not auth.get("user"):
+        return {"error": "invalid_init_data"}
+
+    tg_user = auth["user"]
+    tg_id = tg_user["id"]
+    start_param = body.get("startParam") or auth.get("start_param")
+    explicit_role = body.get("role") if body.get("role") in ("manager", "client") else None
+    user = sheets.get_or_create_user(tg_user, start_param, explicit_role)
+
+    if user.get("role") == "manager":
+        m = sheets.get_manager_profile(tg_id) or {
+            "full_name": user.get("full_name", ""), "salon": "",
+            "is_zov_employee": False, "status": "lapsed", "active_until": None,
+        }
+        return {
+            "role": "manager",
+            "user": {
+                "tg_id": tg_id,
+                "full_name": m.get("full_name") or user.get("full_name", ""),
+                "salon": m.get("salon", ""),
+                "avatar_initial": _initial(m.get("full_name") or tg_user.get("first_name", "")),
+            },
+            "status": m.get("status", "lapsed"),
+            "status_until": _format_date(m.get("active_until")),
+        }
+
+    # client
+    c = sheets.get_client_profile(tg_id) or {}
+    manager = None
+    mgr_id = c.get("manager_tg_id")
+    if mgr_id:
+        try:
+            mgr_id_int = int(mgr_id)
+            mp = sheets.get_manager_profile(mgr_id_int)
+            if mp:
+                manager = {"full_name": mp.get("full_name"), "salon": mp.get("salon")}
+        except (TypeError, ValueError):
+            pass
+
+    full_name = c.get("full_name") or user.get("full_name", "")
+    return {
+        "role": "client",
+        "user": {
+            "tg_id": tg_id,
+            "full_name": full_name,
+            "avatar_initial": _initial(full_name or tg_user.get("first_name", "")),
+        },
+        "manager": manager,
+    }
+
+
+def _handle_measurement(body: dict[str, Any]) -> dict[str, Any]:
+    cfg = get_config()
+    auth = verify_init_data(body.get("initData") or "", cfg.bot_token)
+    if not auth or not auth.get("user"):
+        return {"error": "invalid_init_data"}
+    tg_id = auth["user"]["id"]
+    user = sheets.find_user(tg_id)
+    if not user:
+        return {"error": "user_not_found"}
+
+    m = body.get("measurement") or {}
+    measurement_id = _short_id()
+    filled_by = "manager_for_client" if user.get("role") == "manager" else "client_self"
+
+    client_tg_id = m.get("client_tg_id") if user.get("role") == "manager" else tg_id
+    manager_tg_id = tg_id if user.get("role") == "manager" else (
+        sheets.find_row("Clients", "tg_id", tg_id) or {}
+    ).get("manager_tg_id", "")
+
+    sheets.append_row("Measurements", [
+        measurement_id, _now_iso(), client_tg_id or "", manager_tg_id or "",
+        filled_by,
+        m.get("layout", ""), m.get("area_m2", ""), m.get("ceiling_mm", ""),
+        json.dumps(m.get("walls") or {}, ensure_ascii=False),
+        json.dumps(m.get("openings") or {}, ensure_ascii=False),
+        json.dumps(m.get("infra") or {}, ensure_ascii=False),
+        json.dumps(m.get("niches") or {}, ensure_ascii=False),
+        ",".join(m.get("photos") or []),
+        m.get("notes", ""),
+        "submitted",
+    ])
+
+    if client_tg_id:
+        sheets.update_cell_by_key("Clients", "tg_id", client_tg_id, "last_measurement_id", measurement_id)
+
+    if filled_by == "client_self" and manager_tg_id:
+        tg.send_message(
+            manager_tg_id,
+            f"📐 Новый замер от клиента <b>{user.get('full_name') or tg_id}</b>.\n"
+            f"Площадь: {m.get('area_m2', '?')} м², форма: {m.get('layout', '?')}.\n"
+            f"Открыть в кабинете для просмотра."
+        )
+
+    sheets.log_event("measurement_submitted", tg_id, {"id": measurement_id, "filled_by": filled_by})
+    return {"ok": True, "id": measurement_id}
+
+
+def _handle_podbor(body: dict[str, Any]) -> dict[str, Any]:
+    cfg = get_config()
+    auth = verify_init_data(body.get("initData") or "", cfg.bot_token)
+    if not auth or not auth.get("user"):
+        return {"error": "invalid_init_data"}
+    tg_id = auth["user"]["id"]
+    user = sheets.find_user(tg_id)
+    if not user:
+        return {"error": "user_not_found"}
+    if user.get("role") != "manager":
+        return {"error": "only_manager_can_request_podbor"}
+
+    checklist = body.get("checklist") or {}
+    client_name = body.get("client_name", "")
+    client_tg_id = body.get("client_tg_id", "")
+    measurement_id = body.get("measurement_id", "")
+    lead_id = _short_id()
+
+    sheets.append_row("Leads", [
+        lead_id, _now_iso(), tg_id, client_tg_id, client_name, measurement_id,
+        json.dumps(checklist, ensure_ascii=False),
+        "", "", 0, False, "new", 0,
+    ])
+
+    user_prompt = (
+        f"Подбери технику для следующего клиента:\n\n"
+        f"{json.dumps({'client': {'name': client_name}, 'checklist': checklist}, ensure_ascii=False, indent=2)}"
+    )
+    ai_result = ai.call_ai(user_prompt)
+
+    # Update lead row with AI response
+    sheets.update_cell_by_key("Leads", "id", lead_id, "ai_response",
+                              json.dumps(ai_result.get("json") or ai_result.get("text", ""), ensure_ascii=False))
+    sheets.update_cell_by_key("Leads", "id", lead_id, "ai_model", ai_result.get("model", ""))
+    sheets.update_cell_by_key("Leads", "id", lead_id, "ai_tokens_used", ai_result.get("tokens", 0))
+    sheets.update_cell_by_key("Leads", "id", lead_id, "sent_to_tg", True)
+
+    summary_text = _format_podbor_for_telegram(ai_result, client_name)
+    tg.send_message(tg_id, summary_text)
+
+    sheets.log_event("podbor_completed", tg_id, {"id": lead_id, "tokens": ai_result.get("tokens", 0)})
+    return {"ok": True, "id": lead_id, "summary": summary_text}
+
+
+def _handle_test_ai() -> dict[str, Any]:
+    cfg = get_config()
+    res = ai.call_ai("Скажи одной фразой: что за фабрика ЗОВ?",
+                     system_prompt="Ты — кратко и по делу отвечаешь. Без markdown.")
+    return {
+        "ok": not res.get("error"),
+        "provider": "GigaChat",
+        "model": res.get("model", cfg.gigachat_model),
+        "response_text": (res.get("text") or "")[:500],
+        "tokens": res.get("tokens", 0),
+    }
+
+
+def _handle_test_telegram() -> dict[str, Any]:
+    cfg = get_config()
+    ok = tg.send_message(
+        cfg.admin_tg_id,
+        "🟢 Привет из Python-бэкенда на VPS! Связка backend↔бот работает.",
+    )
+    return {"ok": ok, "sent_to": cfg.admin_tg_id}
+
+
+def _handle_seed_admin() -> dict[str, Any]:
+    cfg = get_config()
+    admin_id = cfg.admin_tg_id
+    if sheets.find_row("Managers", "tg_id", admin_id):
+        return {"ok": True, "status": "already_seeded", "admin_id": admin_id}
+    sheets.append_row("Managers", [
+        admin_id, "Руслан Васильев", "vasrusgen@gmail.com", "",
+        "ЗОВ — куратор сети", "Санкт-Петербург",
+        True, "active", "", "", 0, 0, 0, "MGR_ADMIN",
+    ])
+    if not sheets.find_user(admin_id):
+        sheets.append_row("Users", [
+            admin_id, "VASRUSGEN", "Руслан", "Васильев", "manager",
+            _now_iso(), _now_iso(), "",
+        ])
+    return {"ok": True, "status": "seeded", "admin_id": admin_id, "full_name": "Руслан Васильев"}
+
+
+# =================================================================
+# Helpers
+# =================================================================
+
+def _format_podbor_for_telegram(ai_result: dict[str, Any], client_name: str) -> str:
+    if ai_result.get("error"):
+        return f"❌ Не удалось получить подбор от AI.\n{ai_result.get('text', '')}"
+    j = ai_result.get("json")
+    if not j:
+        return "<b>Подбор готов</b>\n\n" + (ai_result.get("text") or "")[:3500]
+
+    lines = ["✅ <b>Подбор готов</b>"]
+    if client_name:
+        lines.append(f"Клиент: <b>{client_name}</b>")
+    lines.append("")
+    if j.get("summary"):
+        lines.append(j["summary"])
+        lines.append("")
+
+    for item in (j.get("items") or []):
+        lines.append(f"<b>{item.get('brand', '')} {item.get('model', '')}</b>")
+        if item.get("price_rub"):
+            lines.append(f"💰 {_format_price(item['price_rub'])} ₽")
+        if item.get("highlights"):
+            lines.append("✓ " + ", ".join(item["highlights"]))
+        if item.get("caveats"):
+            lines.append(f"⚠️ {item['caveats']}")
+        lines.append("")
+
+    if j.get("total_price_rub"):
+        lines.append(f"<b>ИТОГО: {_format_price(j['total_price_rub'])} ₽</b> · {j.get('budget_status', '')}")
+    if j.get("warnings"):
+        lines.append("\n⚠️ " + "; ".join(j["warnings"]))
+    return "\n".join(lines)
+
+
+def _format_price(n: int | float) -> str:
+    if n is None:
+        return "—"
+    s = str(int(round(float(n))))
+    # Разделители тысяч пробелом
+    return " ".join([s[max(0, len(s) - 3 * (i + 1)):len(s) - 3 * i] for i in range((len(s) + 2) // 3)][::-1]).strip()
+
+
+def _initial(name: str) -> str:
+    return ((name or "").strip()[:1] or "?").upper()
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _format_date(d) -> str | None:
+    if not d:
+        return None
+    if isinstance(d, datetime):
+        return d.strftime("%d.%m.%Y")
+    return str(d)
+
+
+def _short_id() -> str:
+    return uuid.uuid4().hex[:13]
+
+
+async def _safe_json(request: Request) -> dict[str, Any]:
+    try:
+        return await request.json()
+    except Exception:
+        return {}

backend-py/app/sheets.py

@@ -0,0 +1,214 @@
+"""Тонкая обёртка над Google Sheets через gspread + service account."""
+from __future__ import annotations
+import threading
+from datetime import datetime, timedelta, timezone
+from typing import Any
+import gspread
+from google.oauth2.service_account import Credentials
+
+from .config import get_config
+
+_SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
+_lock = threading.Lock()
+_client: gspread.Client | None = None
+_book: gspread.Spreadsheet | None = None
+
+
+def _client_book() -> tuple[gspread.Client, gspread.Spreadsheet]:
+    global _client, _book
+    with _lock:
+        if _client is None:
+            cfg = get_config()
+            creds = Credentials.from_service_account_file(cfg.google_credentials_path, scopes=_SCOPES)
+            _client = gspread.authorize(creds)
+            _book = _client.open_by_key(cfg.sheet_id)
+        return _client, _book  # type: ignore
+
+
+def sheet(name: str) -> gspread.Worksheet:
+    _, book = _client_book()
+    return book.worksheet(name)
+
+
+def append_row(name: str, row: list[Any]) -> None:
+    sheet(name).append_row(row, value_input_option="USER_ENTERED")
+
+
+def find_row(sheet_name: str, key_col: str, key_val: Any) -> dict[str, Any] | None:
+    """Линейный поиск по колонке-ключу. Возвращает строку как dict или None."""
+    s = sheet(sheet_name)
+    rows = s.get_all_values()
+    if not rows:
+        return None
+    headers = rows[0]
+    if key_col not in headers:
+        return None
+    idx = headers.index(key_col)
+    for r in rows[1:]:
+        if len(r) > idx and str(r[idx]).strip() == str(key_val).strip():
+            return dict(zip(headers, r + [""] * (len(headers) - len(r))))
+    return None
+
+
+def update_cell_by_key(sheet_name: str, key_col: str, key_val: Any, target_col: str, new_val: Any) -> bool:
+    s = sheet(sheet_name)
+    rows = s.get_all_values()
+    if not rows:
+        return False
+    headers = rows[0]
+    if key_col not in headers or target_col not in headers:
+        return False
+    key_idx = headers.index(key_col)
+    target_idx = headers.index(target_col)
+    for i, r in enumerate(rows[1:], start=2):
+        if len(r) > key_idx and str(r[key_idx]).strip() == str(key_val).strip():
+            s.update_cell(i, target_idx + 1, new_val)
+            return True
+    return False
+
+
+def get_setting(key: str) -> str | None:
+    row = find_row("Settings", "key", key)
+    return (row or {}).get("value")
+
+
+# === Доменные хелперы ===
+
+def find_user(tg_id: int) -> dict[str, Any] | None:
+    if not tg_id:
+        return None
+    row = find_row("Users", "tg_id", tg_id)
+    if not row:
+        return None
+    full_name = (f"{row.get('first_name', '')} {row.get('last_name', '')}".strip()
+                 or row.get("tg_username", ""))
+    return {**row, "full_name": full_name}
+
+
+def get_or_create_user(tg_user: dict[str, Any], start_param: str | None,
+                       explicit_role: str | None = None) -> dict[str, Any]:
+    cfg = get_config()
+    tg_id = tg_user["id"]
+    admin_id = cfg.admin_tg_id
+
+    existing = find_user(tg_id)
+    now = _now()
+
+    if existing:
+        update_cell_by_key("Users", "tg_id", tg_id, "last_seen_at", now)
+        # Админ всегда manager
+        if tg_id == admin_id and existing.get("role") != "manager":
+            update_cell_by_key("Users", "tg_id", tg_id, "role", "manager")
+            ensure_admin_manager(tg_user)
+            existing["role"] = "manager"
+        elif explicit_role and tg_id != admin_id and existing.get("role") != explicit_role:
+            update_cell_by_key("Users", "tg_id", tg_id, "role", explicit_role)
+            existing["role"] = explicit_role
+        return existing
+
+    # Новый пользователь
+    role = "client"
+    invite_code = ""
+    if tg_id == admin_id:
+        role = "manager"
+    elif explicit_role in ("manager", "client"):
+        role = explicit_role
+    elif start_param and start_param.startswith("client_inv_"):
+        role = "client"
+        invite_code = start_param
+
+    append_row("Users", [
+        tg_id,
+        tg_user.get("username", ""),
+        tg_user.get("first_name", ""),
+        tg_user.get("last_name", ""),
+        role,
+        now,
+        now,
+        invite_code,
+    ])
+    if tg_id == admin_id:
+        ensure_admin_manager(tg_user)
+    return find_user(tg_id) or {}
+
+
+def ensure_admin_manager(tg_user: dict[str, Any]) -> None:
+    tg_id = tg_user["id"]
+    if find_row("Managers", "tg_id", tg_id):
+        return
+    full_name = (f"{tg_user.get('first_name', '')} {tg_user.get('last_name', '')}".strip()
+                 or tg_user.get("username", "") or str(tg_id))
+    append_row("Managers", [
+        tg_id, full_name, "vasrusgen@gmail.com", "",
+        "ЗОВ — куратор сети", "Санкт-Петербург",
+        True, "active", "", "", 0, 0, 0, "MGR_ADMIN",
+    ])
+
+
+def get_manager_profile(tg_id: int) -> dict[str, Any] | None:
+    cfg = get_config()
+    row = find_row("Managers", "tg_id", tg_id)
+    if not row:
+        return None
+    is_zov = str(row.get("is_zov_employee", "")).lower() in ("true", "1", "да", "yes")
+
+    last_order = _parse_date(row.get("last_order_date"))
+    active_period = int(get_setting("ACTIVE_PERIOD_DAYS") or cfg.active_period_days)
+    grace_period = int(get_setting("GRACE_PERIOD_DAYS") or cfg.grace_period_days)
+
+    active_until = None
+    status = "lapsed"
+    if is_zov:
+        status = "active"
+    elif last_order:
+        active_until = last_order + timedelta(days=active_period)
+        grace_until = active_until + timedelta(days=grace_period)
+        now = _now()
+        if now <= active_until:
+            status = "active"
+        elif now <= grace_until:
+            status = "grace"
+        else:
+            status = "lapsed"
+
+    return {
+        **row,
+        "is_zov_employee": is_zov,
+        "active_until": active_until,
+        "status": status,
+    }
+
+
+def get_client_profile(tg_id: int) -> dict[str, Any] | None:
+    return find_row("Clients", "tg_id", tg_id)
+
+
+def log_event(event: str, tg_id: int | None, payload: dict[str, Any] | None = None) -> None:
+    import json
+    try:
+        append_row("Logs", [
+            _now(), event, tg_id or "",
+            json.dumps(payload, ensure_ascii=False) if payload else "",
+        ])
+    except Exception:
+        pass
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc).astimezone()
+
+
+def _parse_date(v: Any) -> datetime | None:
+    if not v:
+        return None
+    if isinstance(v, datetime):
+        return v
+    s = str(v).strip()
+    if not s:
+        return None
+    for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%Y-%m-%dT%H:%M:%S", "%d.%m.%Y %H:%M:%S"):
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            continue
+    return None

backend-py/app/telegram.py

@@ -0,0 +1,28 @@
+"""Тонкая обёртка над Telegram Bot API для отправки уведомлений из backend."""
+from __future__ import annotations
+import httpx
+
+from .config import get_config
+
+
+def send_message(chat_id: int | str, text: str, **kwargs) -> bool:
+    """Отправляет сообщение пользователю/чату. Возвращает True при успехе."""
+    cfg = get_config()
+    if not cfg.bot_token or not chat_id:
+        return False
+    payload = {
+        "chat_id": chat_id,
+        "text": text,
+        "parse_mode": "HTML",
+        "disable_web_page_preview": True,
+    }
+    payload.update(kwargs)
+    try:
+        with httpx.Client(timeout=15.0) as client:
+            resp = client.post(
+                f"https://api.telegram.org/bot{cfg.bot_token}/sendMessage",
+                json=payload,
+            )
+        return 200 <= resp.status_code < 300
+    except Exception:
+        return False

backend-py/requirements.txt

@@ -0,0 +1,7 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.32.0
+pydantic>=2.9
+httpx>=0.27.0
+gspread>=6.0.0
+google-auth>=2.30.0
+python-dotenv>=1.0.0

bot/.dockerignore

@@ -0,0 +1,6 @@
+.venv/
+__pycache__/
+*.pyc
+.env
+.env.local
+credentials.json

bot/Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.12-slim
+
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends ca-certificates \
+ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . /app
+
+ENV PYTHONIOENCODING=utf-8
+ENV PYTHONUNBUFFERED=1
+
+CMD ["python", "main.py"]

deploy/.env.example

@@ -0,0 +1,27 @@
+# ============================================================
+# Скопируйте в /opt/zov-tech/deploy/.env на сервере и заполните.
+# Не коммитить!
+# ============================================================
+
+# Telegram bot
+BOT_TOKEN=8281503057:AAEXmOepY8quH8E3RqOjFbgn7owV1ngnbGA
+ADMIN_TG_ID=5937498515
+
+# GigaChat (от Сбера)
+GIGACHAT_AUTH_KEY=MDE5ZTExY2ItNDgzZi03ZWY4LTk2YjctZjAxNzQ4ZWEwNmVkOmQ1Mzc0OWVlLWUyYjItNDg2Zi04NTk1LWRmNmNlYzQ5M2JjMw==
+GIGACHAT_MODEL=GigaChat-Pro
+GIGACHAT_SCOPE=GIGACHAT_API_PERS
+
+# Google Sheet (БД)
+SHEET_ID=1vAB3u4iOz45awVLp5Pc1X-y5NzPbMugTlagplVdSiR8
+GOOGLE_CREDENTIALS_PATH=/app/credentials.json
+
+# MiniApp (используется ботом)
+MINIAPP_URL=https://wasrusgen.github.io/zov-tech/
+
+# Бэкенд URL для бота (внутри Docker — имя сервиса)
+BACKEND_URL=http://backend:8000
+
+# Бизнес-правила
+ACTIVE_PERIOD_DAYS=90
+GRACE_PERIOD_DAYS=14

deploy/Caddyfile.snippet

@@ -0,0 +1,20 @@
+# Snippet добавляется в /opt/furniture/deploy/Caddyfile.
+# Caddy автоматически получает SSL для api.wasrusgen1.pro через Let's Encrypt.
+
+api.wasrusgen1.pro {
+    reverse_proxy zov-backend:8000
+
+    encode zstd gzip
+    header {
+        Strict-Transport-Security "max-age=31536000; includeSubDomains"
+        X-Content-Type-Options "nosniff"
+        Referrer-Policy "strict-origin-when-cross-origin"
+        -Server
+    }
+    log {
+        output file /data/zov-access.log {
+            roll_size 10mb
+            roll_keep 5
+        }
+    }
+}

deploy/docker-compose.yml

@@ -0,0 +1,46 @@
+services:
+  backend:
+    build:
+      context: ../backend-py
+      dockerfile: Dockerfile
+    image: zov-tech-backend:latest
+    container_name: zov-backend
+    restart: unless-stopped
+    env_file:
+      - .env
+    volumes:
+      - ./credentials.json:/app/credentials.json:ro
+    networks:
+      - web   # внешняя сеть от deploy-стека (Caddy там)
+      - internal
+    expose:
+      - "8000"
+    healthcheck:
+      test: ["CMD", "python", "-c", "import urllib.request,sys; r=urllib.request.urlopen('http://127.0.0.1:8000/healthz', timeout=3); sys.exit(0 if r.status==200 else 1)"]
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 15s
+
+  bot:
+    build:
+      context: ../bot
+      dockerfile: Dockerfile
+    image: zov-tech-bot:latest
+    container_name: zov-bot
+    restart: unless-stopped
+    env_file:
+      - .env
+    networks:
+      - internal
+    depends_on:
+      backend:
+        condition: service_healthy
+
+networks:
+  # Использует уже существующую сеть от furniture-deploy stack — там Caddy
+  web:
+    name: deploy_web
+    external: true
+  internal:
+    driver: bridge