#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================ УЧЁТ ТОВАРОВ — один файл (Flask + JSON + резервная копия на Hugging Face) ============================================================================ Оптимизированная версия. Что изменилось по сравнению с исходником: 1. База больше не встраивается в HTML целиком. Фотографии (это 46 МБ из 47 МБ файла!) вынесены в отдельный эндпоинт /photo/ с вечным кэшированием браузером и ленивой загрузкой. Страница «худеет» с ~47 МБ до ~200 КБ (в сжатом виде ~50 КБ) и грузится мгновенно. 2. Данные держатся в памяти и не перечитываются с диска на каждый запрос. 3. Запись файла — атомарная (tmp + os.replace) и в фоновом потоке, с «склейкой» частых изменений; выгрузка на HF ограничена по частоте. 4. Клиент: порционная отрисовка каталога, делегированные события, ленивые картинки, debounce поиска — интерфейс не «зависает» на тысячах товаров. 5. Исправлены найденные ошибки (см. отчет): XSS, потеря локальных изменений при рестарте, int('') → ValueError, ID/токены сотрудников, деление дат, удаление «мусорных» коллекций в настройках и др. 6. Все POST-действия валидируются на сервере, у каждого действия есть проверка прав и границ значений. Запуск: python app.py Переменные окружения (все необязательные): HF_TOKEN / HF_TOKEN_READ / HF_TOKEN_WRITE - доступ к датасету на HF HF_REPO_ID="Kgshop/vahid" - репозиторий бэкапа DATA_FILE="data_fabrics.json" - файл базы BACKUP_INTERVAL="300" - минимум секунд между выгрузками на HF PORT="7860" - порт ВАЖНО: состояние хранится в памяти процесса, поэтому запускать нужно ОДНИМ процессом (gunicorn -w 1 --threads 8 app:app), иначе воркеры будут расходиться по данным. ============================================================================ """ from __future__ import annotations import atexit import gzip import json import logging import os import re import secrets import signal import threading import time from base64 import b64decode from datetime import datetime from hashlib import md5 from urllib.parse import quote from zoneinfo import ZoneInfo from flask import (Flask, Response, jsonify, make_response, redirect, render_template_string, request, url_for) from werkzeug.middleware.proxy_fix import ProxyFix # ----------------------------- настройки ------------------------------------ DATA_FILE = os.getenv("DATA_FILE", "data_fabrics.json") REPO_ID = os.getenv("HF_REPO_ID", "Kgshop/vahid") HF_TOKEN_WRITE = os.getenv("HF_TOKEN") or os.getenv("HF_TOKEN_WRITE") or "" HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") or HF_TOKEN_WRITE BACKUP_INTERVAL = max(30, int(os.getenv("BACKUP_INTERVAL", "300") or 300)) PORT = int(os.getenv("PORT", "7860")) TZ = ZoneInfo("Asia/Almaty") HISTORY_LIMIT = 1000 # сколько последних операций храним MAX_QTY = 10_000_000 # защита от абсурдных чисел MAX_PRICE = 1_000_000_000 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S") log = logging.getLogger("warehouse") try: # необязательная зависимость: pip install huggingface_hub from huggingface_hub import HfApi, hf_hub_download _HAS_HF = bool(HF_TOKEN_READ) except Exception: # pragma: no cover HfApi = hf_hub_download = None _HAS_HF = False def almaty_now() -> str: """Текущее время по Алматы в формате дд.мм.гггг чч:мм:сс.""" return datetime.now(TZ).strftime("%d.%m.%Y %H:%M:%S") # ----------------------------- утилиты -------------------------------------- _NUM_RE = re.compile(r"[^0-9,.\-]") def to_int(value, default: int = 0, lo: int | None = None, hi: int | None = None) -> int: """Мягкий парсер чисел: '1 250 ₸', '12,5', '', None → int. В исходном коде было int(req.form[...]) сразу в куче мест — любое пустое поле или «12 500» роняло запрос с ValueError. """ if value is None or isinstance(value, bool): return default if isinstance(value, (int, float)): n = int(value) else: s = _NUM_RE.sub("", str(value).strip()).replace(",", ".") if s in ("", "-", "+", ".", "-."): return default try: n = int(float(s)) except ValueError: return default if lo is not None and n < lo: n = lo if hi is not None and n > hi: n = hi return n def clean_text(value, limit: int) -> str: return str(value or "").strip()[:limit] def new_id(prefix: str = "p") -> str: """Уникальный идентификатор без коллизий (исходный str(time.time()) мог совпасть при двух товарах, добавленных в одну микросекунду).""" return f"{prefix}{int(time.time() * 1000):x}{secrets.token_hex(3)}" def decode_photo(value: str): """'data:image/jpeg;base64,....' → (bytes, mime) либо None.""" if not value: return None mime = "image/jpeg" payload = value if value.startswith("data:"): head, _, payload = value.partition(",") if not _: return None mime = (head[5:].split(";")[0] or "image/jpeg").strip().lower() if mime not in ("image/jpeg", "image/png", "image/webp", "image/gif"): mime = "image/jpeg" try: raw = b64decode(payload, validate=False) except Exception: return None if not raw: return None return raw, mime def empty_state() -> dict: return { "settings": { "warehouses": [{"id": "w_main", "name": "Основной склад"}], "shops": [{"id": "s_main", "name": "Основной магазин"}], "employees": [], }, "products": [], "history": [], } def norm_loc_list(value, fallback_name: str) -> list: out, seen = [], set() if isinstance(value, list): for item in value: if not isinstance(item, dict): continue lid = clean_text(item.get("id"), 64) if not lid or lid in seen: continue seen.add(lid) out.append({"id": lid, "name": clean_text(item.get("name"), 120) or lid}) if not out: out = [{"id": new_id("loc_"), "name": fallback_name}] return out def norm_product(raw: dict) -> dict: """Приводит товар к единому виду (заодно переносит старый формат colors).""" if not isinstance(raw, dict): raw = {} legacy_price = to_int(raw.get("price"), 0, 0, MAX_PRICE) variants = [] for v in raw.get("variants") or []: if not isinstance(v, dict): continue name = clean_text(v.get("name"), 120) if not name: continue stocks = {} if isinstance(v.get("stocks"), dict): for loc, amt in v["stocks"].items(): loc = clean_text(loc, 64) if loc: stocks[loc] = to_int(amt, 0, 0, MAX_QTY) variants.append({"name": name, "price": to_int(v.get("price"), 0, 0, MAX_PRICE), "stocks": stocks}) stocks = {} for c in raw.get("colors") or []: # старый формат if isinstance(c, dict) and isinstance(c.get("stocks"), dict): for loc, amt in c["stocks"].items(): loc = clean_text(loc, 64) if loc: stocks[loc] = stocks.get(loc, 0) + to_int(amt, 0, 0, MAX_QTY) if isinstance(raw.get("stocks"), dict): for loc, amt in raw["stocks"].items(): loc = clean_text(loc, 64) if loc: stocks[loc] = stocks.get(loc, 0) + to_int(amt, 0, 0, MAX_QTY) return { "id": clean_text(raw.get("id"), 64) or new_id(), "name": clean_text(raw.get("name"), 200) or "Без названия", "description": clean_text(raw.get("description"), 2000), "price_retail": to_int(raw.get("price_retail", legacy_price), 0, 0, MAX_PRICE), "price_wholesale": to_int(raw.get("price_wholesale", legacy_price), 0, 0, MAX_PRICE), "variants": variants, "stocks": {} if variants else stocks, "box_qty": max(1, to_int(raw.get("box_qty"), 1, 1, MAX_QTY)), } def norm_history(raw) -> list: out = [] if isinstance(raw, list): for h in raw[:HISTORY_LIMIT]: if not isinstance(h, dict): continue out.append({ "time": clean_text(h.get("time"), 32), "action": clean_text(h.get("action"), 32), "product": clean_text(h.get("product"), 240), "amount": to_int(h.get("amount"), 0, 0, MAX_QTY), "price": to_int(h.get("price"), 0, 0, MAX_PRICE), "from_loc_name": clean_text(h.get("from_loc_name"), 120), "to_loc_name": clean_text(h.get("to_loc_name"), 120), "emp_name": clean_text(h.get("emp_name"), 120), }) return out def normalize_state(raw) -> tuple[dict, dict]: """(state, photos). photos: {product_id: data-url}.""" if not isinstance(raw, dict): return empty_state(), {} state = empty_state() settings = raw.get("settings") if isinstance(raw.get("settings"), dict) else {} state["settings"]["warehouses"] = norm_loc_list(settings.get("warehouses"), "Основной склад") state["settings"]["shops"] = norm_loc_list(settings.get("shops"), "Основной магазин") seen_ids = set() products, photos = [], {} if isinstance(raw.get("products"), list): for item in raw["products"]: p = norm_product(item) if p["id"] in seen_ids: # битые дубли id в базе p["id"] = new_id() seen_ids.add(p["id"]) photos[p["id"]] = str((item or {}).get("photo") or "") if isinstance(item, dict) else "" products.append(p) state["products"] = products employees, seen_tokens = [], set() for e in (settings.get("employees") or []): if not isinstance(e, dict): continue token = clean_text(e.get("token") or e.get("id"), 64) if not token or token in seen_tokens: # дубли токенов = чужой доступ token = secrets.token_urlsafe(18) seen_tokens.add(token) employees.append({ "id": clean_text(e.get("id"), 32) or token[:8], "name": clean_text(e.get("name"), 120) or "Сотрудник", "token": token, "shop_id": clean_text(e.get("shop_id"), 64), }) state["settings"]["employees"] = employees state["history"] = norm_history(raw.get("history")) return state, photos # ------------------------------- склад данных ------------------------------- class Store: """Состояние приложения в памяти + фоновая атомарная запись и бэкап.""" def __init__(self, path: str): self.path = path self.lock = threading.RLock() self.data = empty_state() self.photo_raw: dict[str, str] = {} # id -> исходная data-url (для файла на HF) self.photo_bin: dict[str, tuple[bytes, str]] = {} self.photo_ver: dict[str, str] = {} self.rev = 1 self.hf_enabled = bool(_HAS_HF and HF_TOKEN_WRITE) self._dirty = False self._upload_pending = False self._last_upload = 0.0 self._wake = threading.Event() self._stop = threading.Event() self._thread: threading.Thread | None = None # ---------------- загрузка ---------------- def load(self) -> None: raw = None if os.path.exists(self.path): try: with open(self.path, "r", encoding="utf-8") as fh: raw = json.load(fh) log.info("База загружена из файла %s", self.path) except Exception as exc: log.error("Файл %s повреждён (%s) — работаю с копией HF", self.path, exc) try: os.replace(self.path, self.path + ".broken") except OSError: pass # Скачиваем с HF ТОЛЬКО если локального файла нет: иначе можно # потерять изменения, которые ещё не успели уехать в бэкап (баг старой версии). if raw is None and _HAS_HF: try: p = hf_hub_download(repo_id=REPO_ID, filename=os.path.basename(self.path), repo_type="dataset", token=HF_TOKEN_READ, local_dir=".") with open(p, "r", encoding="utf-8") as fh: raw = json.load(fh) log.info("База скачана с Hugging Face (%s)", REPO_ID) except Exception as exc: log.warning("Не удалось скачать базу с HF: %s", exc) state, photos = normalize_state(raw) self.data = state self.photo_raw = photos self._rebuild_photo_cache() self.rev = int(time.time() * 1000) log.info("Товаров: %d, операций в истории: %d, фото: %d", len(state["products"]), len(state["history"]), sum(1 for v in photos.values() if v)) def _rebuild_photo_cache(self) -> None: self.photo_bin.clear() self.photo_ver.clear() for pid, url in self.photo_raw.items(): if not url: continue dec = decode_photo(url) if dec: self.photo_bin[pid] = dec self.photo_ver[pid] = md5(url.encode("utf-8", "ignore")).hexdigest()[:10] # ---------------- чтение ---------------- def get_photo(self, pid: str): return self.photo_bin.get(pid), self.photo_ver.get(pid, "") def loc_name(self, loc_id: str) -> str: st = self.data["settings"] for group in ("warehouses", "shops"): for loc in st[group]: if loc["id"] == loc_id: return loc["name"] return loc_id or "" def valid_loc(self, loc_id) -> str: loc_id = clean_text(loc_id, 64) st = self.data["settings"] for group in ("warehouses", "shops"): for loc in st[group]: if loc["id"] == loc_id: return loc_id return "" def first_loc(self) -> str: st = self.data["settings"] return (st["warehouses"][0]["id"] if st["warehouses"] else (st["shops"][0]["id"] if st["shops"] else "")) def summary(self, role: str, emp: dict | None, price_type: str) -> tuple[dict, int]: """Компактный JSON для страницы: БЕЗ фото и служебных полей. Возвращает (payload, rev) — rev снят под тем же замком, поэтому ETag страницы всегда соответствует отданным данным. """ with self.lock: return self._build_summary(role, emp, price_type), self.rev def _build_summary(self, role: str, emp: dict | None, price_type: str) -> dict: data = self.data locs = ([{"i": w["id"], "n": w["name"], "k": "w"} for w in data["settings"]["warehouses"]] + [{"i": s["id"], "n": s["name"], "k": "s"} for s in data["settings"]["shops"]]) products = [] for p in data["products"]: products.append({ "q": p["id"], "n": p["name"], "d": p["description"], "r": p["price_retail"], "w": p["price_wholesale"], "b": p["box_qty"], "v": [{"n": v["name"], "p": v["price"], "s": v["stocks"]} for v in p["variants"]], "s": p["stocks"], "ph": self.photo_ver.get(p["id"], ""), }) out = { "role": role, "pt": price_type, "emp": ({"name": emp["name"], "shop_id": emp["shop_id"]} if emp else None), "locs": locs, "p": products, "hf": self.hf_enabled, } if role == "admin": out["h"] = [[h["time"], h["action"], h["product"], h["amount"], h["price"], h["from_loc_name"], h["to_loc_name"], h["emp_name"]] for h in data["history"]] out["emps"] = [{"i": e["id"], "n": e["name"], "t": e["token"], "s": e["shop_id"]} for e in data["settings"]["employees"]] return out # ---------------- изменение ---------------- def touch(self) -> None: """Данные изменились: новый rev, отметка «грязно», будим писателя.""" with self.lock: self.rev += 1 self._dirty = True self._wake.set() def set_photo(self, pid: str, data_url: str) -> None: with self.lock: self.photo_raw[pid] = data_url or "" if data_url: dec = decode_photo(data_url) if dec: self.photo_bin[pid] = dec self.photo_ver[pid] = md5(data_url.encode("utf-8", "ignore")).hexdigest()[:10] else: self.photo_bin.pop(pid, None) self.photo_ver.pop(pid, None) else: self.photo_bin.pop(pid, None) self.photo_ver.pop(pid, None) def drop_photo(self, pid: str) -> None: self.set_photo(pid, "") def add_history(self, action: str, product: str, amount: int = 0, price: int = 0, from_loc: str = "", to_loc: str = "", emp_name: str = "") -> None: hist = self.data["history"] hist.insert(0, { "time": almaty_now(), "action": action, "product": product, "amount": int(amount), "price": int(price), "from_loc_name": from_loc, "to_loc_name": to_loc, "emp_name": emp_name, }) del hist[HISTORY_LIMIT:] # ---------------- запись на диск / бэкап ---------------- def _serialize(self) -> str: """Собирает JSON так, как его ждёт HF (фото внутри товара).""" with self.lock: products = [dict(p, photo=self.photo_raw.get(p["id"], "")) for p in self.data["products"]] payload = {"settings": self.data["settings"], "products": products, "history": self.data["history"]} return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) def flush(self) -> bool: with self.lock: if not self._dirty: return False self._dirty = False try: blob = self._serialize() tmp = f"{self.path}.tmp" with open(tmp, "w", encoding="utf-8") as fh: fh.write(blob) os.replace(tmp, self.path) # атомарная замена self._upload_pending = self.hf_enabled return True except Exception as exc: log.error("Не удалось сохранить базу: %s", exc) with self.lock: self._dirty = True # попробуем ещё раз return False def maybe_upload(self, force: bool = False) -> None: if not self.hf_enabled or not (self._upload_pending or force): return if not force and (time.time() - self._last_upload) < BACKUP_INTERVAL: return try: HfApi().upload_file( path_or_fileobj=self.path, path_in_repo=os.path.basename(self.path), repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Backup {almaty_now()}") self._last_upload = time.time() self._upload_pending = False log.info("Резервная копия выгружена на HF") except Exception as exc: log.warning("Бэкап на HF не удался: %s", exc) def start(self) -> None: if self._thread: return self._thread = threading.Thread(target=self._loop, name="store-writer", daemon=True) self._thread.start() atexit.register(self.shutdown) if threading.current_thread() is threading.main_thread(): try: # при остановке контейнера (SIGTERM) успеваем сбросить данные на диск signal.signal(signal.SIGTERM, self._on_signal) except (ValueError, OSError): # pragma: no cover pass def _on_signal(self, _signum, _frame): # pragma: no cover self.shutdown() raise SystemExit(0) def _loop(self) -> None: while not self._stop.wait(2.0): with self.lock: dirty = self._dirty if dirty: self._stop.wait(1.5) # склеиваем серию правок self.flush() self.maybe_upload() def shutdown(self) -> None: if self._stop.is_set(): return self._stop.set() try: self.flush() self.maybe_upload(force=True) except Exception: pass STORE = Store(DATA_FILE) # ------------------------------ действия (POST) ----------------------------- ADMIN_ACTIONS = {"add", "edit", "delete", "income", "transfer", "add_loc", "del_loc", "add_emp", "del_emp"} def _variant_pairs(form) -> list[tuple[str, int, int]]: """Читает строки вариантов (v_name[] / v_price[] / v_stock[]) с валидацией.""" out = [] for name, price, stock in zip(form.getlist("v_name[]"), form.getlist("v_price[]"), form.getlist("v_stock[]")): name = clean_text(name, 120) if name: out.append((name, to_int(price, 0, 0, MAX_PRICE), to_int(stock, 0, 0, MAX_QTY))) return out def _merge_variants(product: dict, pairs, loc_id: str) -> list: """Пересобирает список вариантов, сохраняя остатки по «старым» складам.""" old = product.get("variants") or [] old_by_name = {v["name"]: v for v in old} result = [] for i, (name, price, stock) in enumerate(pairs): stocks = {} if name in old_by_name: stocks = dict(old_by_name[name]["stocks"]) elif i < len(old): stocks = dict(old[i]["stocks"]) if loc_id: stocks[loc_id] = stock result.append({"name": name, "price": price, "stocks": stocks}) return result def apply_action(form, role: str, emp: dict | None) -> dict: """Единая точка обработки POST. Всегда возвращает dict (ok / error).""" action = clean_text(form.get("action"), 32) emp_name = emp["name"] if emp else "Админ" if action in ADMIN_ACTIONS and role != "admin": return {"ok": False, "error": "Недостаточно прав"} if action == "checkout" and role not in ("admin", "employee"): return {"ok": False, "error": "Недостаточно прав"} data = STORE.data result: dict = {"ok": False, "error": "Неизвестное действие"} changed = False with STORE.lock: products = data["products"] # ----------------------------- добавить товар ------------------------ if action == "add": name = clean_text(form.get("name"), 200) if not name: return {"ok": False, "error": "Укажите название товара"} loc_id = STORE.valid_loc(form.get("loc_id")) or STORE.first_loc() photo = form.get("photo") or "" if len(photo) > 6_000_000: # ~4.5 МБ картинки photo = "" pairs = _variant_pairs(form) pid = new_id() product = { "id": pid, "name": name, "description": clean_text(form.get("description"), 2000), "price_retail": to_int(form.get("price_retail"), 0, 0, MAX_PRICE), "price_wholesale": to_int(form.get("price_wholesale"), 0, 0, MAX_PRICE), "variants": [{"name": n, "price": pr, "stocks": ({loc_id: st} if loc_id else {})} for n, pr, st in pairs], "stocks": {}, "box_qty": max(1, to_int(form.get("box_qty"), 1, 1, MAX_QTY)), } if not product["variants"]: product["stocks"] = {loc_id: to_int(form.get("rolls"), 0, 0, MAX_QTY)} if loc_id else {} products.insert(0, product) if photo: STORE.set_photo(pid, photo) else: STORE.photo_raw[pid] = "" changed = True result = {"ok": True, "id": pid} # ----------------------------- изменить товар ------------------------ elif action == "edit": idx = to_int(form.get("index"), -1) if not (0 <= idx < len(products)): return {"ok": False, "error": "Товар не найден"} p = products[idx] name = clean_text(form.get("name"), 200) if name: p["name"] = name if form.get("description") is not None: p["description"] = clean_text(form.get("description"), 2000) if form.get("price_retail") is not None: p["price_retail"] = to_int(form.get("price_retail"), p["price_retail"], 0, MAX_PRICE) if form.get("price_wholesale") is not None: p["price_wholesale"] = to_int(form.get("price_wholesale"), p["price_wholesale"], 0, MAX_PRICE) if form.get("box_qty") is not None: p["box_qty"] = max(1, to_int(form.get("box_qty"), p["box_qty"], 1, MAX_QTY)) loc_id = STORE.valid_loc(form.get("edit_loc_id")) pairs = _variant_pairs(form) if pairs: p["variants"] = _merge_variants(p, pairs, loc_id) p["stocks"] = {} else: p["variants"] = [] if loc_id and form.get("rolls") is not None: p.setdefault("stocks", {}) p["stocks"][loc_id] = to_int(form.get("rolls"), 0, 0, MAX_QTY) photo = form.get("photo") or "" if photo and len(photo) <= 6_000_000: STORE.set_photo(p["id"], photo) changed = True result = {"ok": True} # ----------------------------- удалить товар ------------------------- elif action == "delete": idx = to_int(form.get("index"), -1) if not (0 <= idx < len(products)): return {"ok": False, "error": "Товар не найден"} removed = products.pop(idx) STORE.drop_photo(removed["id"]) STORE.add_history("Удаление", removed["name"], 0, 0, "", "", emp_name) changed = True result = {"ok": True} # ----------------------------- приход -------------------------------- elif action == "income": idx = to_int(form.get("product_index"), -1) amount = to_int(form.get("rolls"), 0, 1, MAX_QTY) loc_id = STORE.valid_loc(form.get("loc_id")) if not (0 <= idx < len(products)): return {"ok": False, "error": "Товар не найден"} if not loc_id: return {"ok": False, "error": "Выберите склад/магазин"} if amount <= 0: return {"ok": False, "error": "Количество должно быть больше нуля"} p = products[idx] variants = p.get("variants") or [] prod_name = p["name"] if variants: v_idx = to_int(form.get("v_index"), -1) if not (0 <= v_idx < len(variants)): return {"ok": False, "error": "Выберите вариант товара"} v = variants[v_idx] v.setdefault("stocks", {}) v["stocks"][loc_id] = to_int(v["stocks"].get(loc_id), 0, 0, MAX_QTY) + amount prod_name += f" ({v['name']})" else: p.setdefault("stocks", {}) p["stocks"][loc_id] = to_int(p["stocks"].get(loc_id), 0, 0, MAX_QTY) + amount STORE.add_history("Приход", prod_name, amount, p.get("price_wholesale", 0), "", STORE.loc_name(loc_id), emp_name) changed = True result = {"ok": True, "amount": amount} # ----------------------------- перемещение --------------------------- elif action == "transfer": idx = to_int(form.get("pIndex"), -1) v_idx = to_int(form.get("vIndex"), -1) amount = to_int(form.get("amt"), 0, 0, MAX_QTY) from_loc = STORE.valid_loc(form.get("from_loc")) to_loc = STORE.valid_loc(form.get("to_loc")) if not (0 <= idx < len(products)): return {"ok": False, "error": "Товар не найден"} if not from_loc or not to_loc: return {"ok": False, "error": "Выберите склады"} if from_loc == to_loc: return {"ok": False, "error": "Откуда и куда совпадают"} if amount <= 0: return {"ok": False, "error": "Укажите количество"} p = products[idx] variants = p.get("variants") or [] prod_name = p["name"] bucket = None if variants: if not (0 <= v_idx < len(variants)): return {"ok": False, "error": "Выберите вариант товара"} bucket = variants[v_idx].setdefault("stocks", {}) prod_name += f" ({variants[v_idx]['name']})" else: bucket = p.setdefault("stocks", {}) available = to_int(bucket.get(from_loc), 0, 0, MAX_QTY) amount = min(amount, available) if amount <= 0: return {"ok": False, "error": "На складе-источнике нет остатка"} bucket[from_loc] = available - amount bucket[to_loc] = to_int(bucket.get(to_loc), 0, 0, MAX_QTY) + amount STORE.add_history("Перемещение", prod_name, amount, 0, STORE.loc_name(from_loc), STORE.loc_name(to_loc), emp_name) changed = True result = {"ok": True, "amount": amount} # ----------------------------- списание (продажа) -------------------- elif action == "checkout": try: cart = json.loads(form.get("cart") or "[]") except Exception: cart = [] if not isinstance(cart, list): cart = [] sold, errors, total = [], [], 0 for item in cart[:200]: if not isinstance(item, dict): continue idx = to_int(item.get("pIndex"), -1) v_idx = to_int(item.get("vIndex"), -1) amount = to_int(item.get("amt"), 0, 0, MAX_QTY) price = to_int(item.get("price"), 0, 0, MAX_PRICE) loc_id = STORE.valid_loc(item.get("loc")) name = clean_text(item.get("pName"), 240) or "Товар" if amount <= 0: continue if not (0 <= idx < len(products)): errors.append(f"{name}: товар не найден") continue if not loc_id: errors.append(f"{name}: не выбрана локация") continue p = products[idx] variants = p.get("variants") or [] if variants: if not (0 <= v_idx < len(variants)): errors.append(f"{name}: не выбран вариант") continue bucket = variants[v_idx].setdefault("stocks", {}) else: bucket = p.setdefault("stocks", {}) available = to_int(bucket.get(loc_id), 0, 0, MAX_QTY) amount = min(amount, available) if amount <= 0: errors.append(f"{name}: нет остатка") continue bucket[loc_id] = available - amount STORE.add_history("Продажа", name, amount, price, STORE.loc_name(loc_id), "", emp_name) sold.append({"name": name, "amt": amount, "sum": amount * price}) total += amount * price changed = bool(sold) result = {"ok": bool(sold), "sold": sold, "errors": errors, "total": total} if not sold and not errors: result["error"] = "Пустая корзина" # ----------------------------- склады / магазины --------------------- elif action == "add_loc": l_type = clean_text(form.get("l_type"), 16) l_name = clean_text(form.get("l_name"), 120) if l_type not in ("warehouses", "shops") or not l_name: return {"ok": False, "error": "Неверные данные локации"} data["settings"][l_type].append({"id": new_id("loc_"), "name": l_name}) changed = True result = {"ok": True} elif action == "del_loc": l_type = clean_text(form.get("l_type"), 16) l_id = clean_text(form.get("l_id"), 64) if l_type not in ("warehouses", "shops"): return {"ok": False, "error": "Неверный тип локации"} if len(data["settings"][l_type]) <= 1: return {"ok": False, "error": "Нельзя удалить последнюю локацию"} before = len(data["settings"][l_type]) data["settings"][l_type] = [x for x in data["settings"][l_type] if x["id"] != l_id] if len(data["settings"][l_type]) == before: return {"ok": False, "error": "Локация не найдена"} for p in products: # чистим остатки удалённой локации p["stocks"].pop(l_id, None) for v in p.get("variants") or []: v["stocks"].pop(l_id, None) for e in data["settings"]["employees"]: # отвязываем сотрудников if e["shop_id"] == l_id: e["shop_id"] = "" changed = True result = {"ok": True} # ----------------------------- сотрудники ---------------------------- elif action == "add_emp": e_name = clean_text(form.get("e_name"), 120) e_shop = STORE.valid_loc(form.get("e_shop")) if not e_name: return {"ok": False, "error": "Укажите имя сотрудника"} token = secrets.token_urlsafe(18) # криптостойкая ссылка доступа data["settings"]["employees"].append({ "id": new_id("e"), "name": e_name, "token": token, "shop_id": e_shop or "", }) changed = True result = {"ok": True, "token": token} elif action == "del_emp": e_id = clean_text(form.get("e_id"), 64) before = len(data["settings"]["employees"]) data["settings"]["employees"] = [x for x in data["settings"]["employees"] if x["id"] != e_id and x["token"] != e_id] if len(data["settings"]["employees"]) == before: return {"ok": False, "error": "Сотрудник не найден"} changed = True result = {"ok": True} if changed: STORE.touch() return result # ============================== страница ==================================== MAIN_HTML = r''' Учёт товаров

{{ TITLE }}

Учёт товаров ·

Добавить товар

Варианты товара (не обязательно)
Фото товара (не обязательно)

Изменить товар

Остатки по локациям
Варианты товара
Новое фото (не обязательно)

Приход

Перемещение

''' # ============================== Flask-приложение ============================= APP = Flask(__name__) APP.config.update( MAX_CONTENT_LENGTH=24 * 1024 * 1024, # фото в base64 + поля формы TEMPLATES_AUTO_RELOAD=False, JSON_SORT_KEYS=False, ) try: # Flask >= 2.2 APP.json.ensure_ascii = False except Exception: # pragma: no cover APP.config["JSON_AS_ASCII"] = False APP.wsgi_app = ProxyFix(APP.wsgi_app, x_for=1, x_proto=1, x_host=1) _TPL = None def get_template(): """Шаблон компилируется один раз (render_template_string компилирует каждый раз).""" global _TPL if _TPL is None: _TPL = APP.jinja_env.from_string(MAIN_HTML) return _TPL def wants_json() -> bool: return request.headers.get("X-Requested-With") == "fetch" def render_panel(role: str, emp: dict | None, price_type: str, title: str): summary, rev = STORE.summary(role, emp, price_type) boot = json.dumps(summary, ensure_ascii=False, separators=(",", ":")) # '<' экранируем, чтобы в скрипт нельзя было вставить '' (XSS) boot = boot.replace("<", "\\u003c").replace("\u2028", "\\u2028").replace("\u2029", "\\u2029") html = get_template().render(BOOT=boot, TITLE=title, BACKUP_INTERVAL=BACKUP_INTERVAL) resp = make_response(html) resp.headers["Cache-Control"] = "no-cache" # браузер спрашивает «не изменилось?» resp.set_etag(f"v{rev}-{role}-{price_type}-{emp['id'] if emp else '-'}") return resp.make_conditional(request) # ответит 304, если данные те же def post_response(result: dict, back: str): if wants_json(): return jsonify(result), (200 if result.get("ok") else 400) if not result.get("ok"): err = (result.get("error") or "Не удалось выполнить действие") back = f"{back}{'&' if '?' in back else '?'}err={quote(err)}" return redirect(back, 303) # PRG: перезагрузка без повторной отправки # --------------------------------- роуты ------------------------------------ @APP.get("/") def root(): return redirect(url_for("catalog_retail")) @APP.get("/catalog/retail") def catalog_retail(): return render_panel("public", None, "retail", "Розничный каталог") @APP.get("/catalog/wholesale") def catalog_wholesale(): return render_panel("public", None, "wholesale", "Оптовый каталог") @APP.route("/admin", methods=["GET", "POST"]) def admin_panel(): if request.method == "POST": result = apply_action(request.form, "admin", None) return post_response(result, url_for("admin_panel")) return render_panel("admin", None, "both", "Админ панель") @APP.route("/emp/", methods=["GET", "POST"]) def employee_panel(token: str): emp = next((e for e in STORE.data["settings"]["employees"] if e["token"] == token), None) if not emp: return render_template_string(ERROR_HTML, code=403, msg="Ссылка недействительна или сотрудник удалён"), 403 if request.method == "POST": result = apply_action(request.form, "employee", emp) return post_response(result, url_for("employee_panel", token=token)) return render_panel("employee", emp, "both", "Панель сотрудника") @APP.get("/photo/") def photo(pid: str): """Отдаёт фото товара из памяти. Так 46 МБ картинок не летят в HTML.""" found, ver = STORE.get_photo(pid) if not found: return Response(status=404) raw, mime = found resp = Response(raw, mimetype=mime) resp.headers["Cache-Control"] = "public, max-age=31536000, immutable" resp.set_etag(ver) return resp.make_conditional(request) @APP.get("/healthz") def healthz(): return jsonify(ok=True, products=len(STORE.data["products"]), history=len(STORE.data["history"]), rev=STORE.rev, hf=STORE.hf_enabled, backup_file=os.path.basename(DATA_FILE)) # ------------------------------- сжатие ответов ----------------------------- @APP.after_request def compress_and_secure(resp): resp.headers["X-Content-Type-Options"] = "nosniff" resp.headers["Referrer-Policy"] = "same-origin" if resp.status_code < 300 and not resp.direct_passthrough \ and not resp.headers.get("Content-Encoding") \ and "gzip" in (request.headers.get("Accept-Encoding") or "").lower() \ and resp.mimetype in ("text/html", "application/json", "text/css", "application/javascript", "text/plain"): data = resp.get_data() if len(data) >= 1024: resp.set_data(gzip.compress(data, 5)) resp.headers["Content-Encoding"] = "gzip" resp.headers["Content-Length"] = str(len(resp.get_data())) resp.headers.add("Vary", "Accept-Encoding") return resp # ------------------------------- страницы ошибок ---------------------------- ERROR_HTML = """ {{ code }}

{{ code }}

{{ msg }}

В каталог
""" @APP.errorhandler(404) def on_404(_e): if request.path.startswith("/photo/"): return Response(status=404) return render_template_string(ERROR_HTML, code=404, msg="Страница не найдена"), 404 @APP.errorhandler(403) def on_403(_e): return render_template_string(ERROR_HTML, code=403, msg="Доступ запрещён"), 403 @APP.errorhandler(413) def on_413(_e): return render_template_string(ERROR_HTML, code=413, msg="Слишком большой запрос (фото больше 8 МБ)"), 413 @APP.errorhandler(500) def on_500(err): log.exception("Внутренняя ошибка: %s", err) return render_template_string(ERROR_HTML, code=500, msg="Внутренняя ошибка сервера"), 500 # ---------------------------------- запуск ---------------------------------- STORE.load() STORE.start() def print_banner() -> None: log.info("=" * 62) log.info("Учёт товаров запущен: http://0.0.0.0:%d", PORT) log.info("Админка: /admin Розница: /catalog/retail Опт: /catalog/wholesale") log.info("Товаров: %d | фото: %d | история: %d", len(STORE.data["products"]), sum(1 for v in STORE.photo_ver.values()), len(STORE.data["history"])) if STORE.hf_enabled: log.info("Бэкап на HF включён (%s, не чаще 1 раза в %d c)", REPO_ID, BACKUP_INTERVAL) else: log.warning("Бэкап на HF выключен: не задан HF_TOKEN/HF_TOKEN_READ.") log.info("=" * 62) print_banner() if __name__ == "__main__": APP.run(host="0.0.0.0", port=PORT, debug=False, threaded=True)