byrsapty
71210b6bdb
Previously the badge always showed next to the price for any per_100g product and never next to the weight at all. Adds two per-product checkboxes (show_estimate_badge_price, default on; show_estimate_badge_weight, default off) so the admin can turn either one on or off independently — covers a shashlik-style product where the weight itself is also just an estimate, not only the price. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1059 lines
44 KiB
Python
1059 lines
44 KiB
Python
import uuid
|
||
from pathlib import Path
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from flask import flash, jsonify, redirect, render_template, request, url_for
|
||
from flask_login import current_user, login_required, login_user, logout_user
|
||
from werkzeug.utils import secure_filename
|
||
|
||
from bober_bbq.admin import admin_bp
|
||
from bober_bbq.admin.authz import admin_required
|
||
from bober_bbq.config import config
|
||
from bober_bbq.extensions import db
|
||
from bober_bbq.models import (
|
||
ALLERGENS,
|
||
DEFAULT_SETTINGS,
|
||
ORDER_STATUS_LABELS,
|
||
ORDER_STATUS_NEXT,
|
||
ORDER_STATUSES,
|
||
ORDER_TYPE_LABELS,
|
||
ORDER_TYPES,
|
||
PAYMENT_METHODS,
|
||
AdminUser,
|
||
Category,
|
||
Ingredient,
|
||
Order,
|
||
OrderItem,
|
||
Product,
|
||
PromoCode,
|
||
Setting,
|
||
StockMovement,
|
||
)
|
||
from bober_bbq.utils import checkbox_prro, login_throttle
|
||
from bober_bbq.utils.audit import log_action
|
||
from bober_bbq.utils.inventory import deduct_for_order, inventory_enabled, restore_for_order
|
||
from bober_bbq.utils.notify_customer import notify_customer_status_change
|
||
from bober_bbq.utils.pricing import calc_delivery_fee, calc_pickup_discount
|
||
from bober_bbq.utils.theme import derive_palette
|
||
from bober_bbq.utils.workhours import is_open as _is_open
|
||
from bot.services import order_service
|
||
|
||
ALLOWED_IMAGE_EXT = {"jpg", "jpeg", "png", "webp", "svg"}
|
||
|
||
|
||
@admin_bp.context_processor
|
||
def inject_admin_globals():
|
||
# logo/cafe_name are public info (already exposed via /api/settings), so
|
||
# show them on the login page too, before the user is authenticated.
|
||
globals_ = {
|
||
"logo_url": Setting.get("logo_url", ""),
|
||
"cafe_name": Setting.get("cafe_name", "Bober BBQ"),
|
||
"admin_theme": derive_palette(Setting.get("admin_theme_accent", "#ff7a3d")),
|
||
}
|
||
if not current_user.is_authenticated:
|
||
return globals_
|
||
inv_enabled = Setting.get_bool("inventory_enabled", False)
|
||
suppliers_manager_access = Setting.get_bool("suppliers_manager_access", False)
|
||
globals_.update(
|
||
{
|
||
"new_orders_count": Order.query.filter_by(status="new").count(),
|
||
"is_open": _is_open(),
|
||
"work_hours_from": Setting.get("work_hours_from", "11:00"),
|
||
"work_hours_to": Setting.get("work_hours_to", "21:00"),
|
||
"order_status_labels": ORDER_STATUS_LABELS,
|
||
"inventory_enabled": inv_enabled,
|
||
"low_stock_count": (
|
||
Ingredient.query.filter(Ingredient.current_stock <= Ingredient.reorder_threshold).count()
|
||
if inv_enabled
|
||
else 0
|
||
),
|
||
"suppliers_manager_access": suppliers_manager_access,
|
||
"suppliers_nav_visible": current_user.role == "admin" or suppliers_manager_access,
|
||
"sidebar_collapsible": Setting.get_bool("sidebar_collapsible_menu", True),
|
||
}
|
||
)
|
||
return globals_
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Auth
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if request.method == "POST":
|
||
ip = request.remote_addr or "unknown"
|
||
locked_for = login_throttle.seconds_locked(ip)
|
||
if locked_for is not None:
|
||
flash(f"Забагато невдалих спроб входу. Спробуйте ще раз через {locked_for // 60 + 1} хв.", "error")
|
||
return render_template("admin/login.html")
|
||
|
||
username = request.form.get("username", "")
|
||
password = request.form.get("password", "")
|
||
user = AdminUser.query.filter_by(username=username).first()
|
||
if user and user.check_password(password):
|
||
login_throttle.record_success(ip)
|
||
user.last_login_at = datetime.now(timezone.utc)
|
||
db.session.commit()
|
||
login_user(user)
|
||
return redirect(url_for("admin.dashboard"))
|
||
login_throttle.record_failure(ip)
|
||
flash("Невірний логін або пароль", "error")
|
||
return render_template("admin/login.html")
|
||
|
||
|
||
@admin_bp.route("/api/new-orders-count")
|
||
@login_required
|
||
def api_new_orders_count():
|
||
return jsonify({"count": Order.query.filter_by(status="new").count()})
|
||
|
||
|
||
@admin_bp.route("/logout")
|
||
@login_required
|
||
def logout():
|
||
logout_user()
|
||
return redirect(url_for("admin.login"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dashboard
|
||
# ---------------------------------------------------------------------------
|
||
def _pct_change(today: float, yesterday: float) -> float | None:
|
||
if yesterday <= 0:
|
||
return None
|
||
return round((today - yesterday) / yesterday * 100)
|
||
|
||
|
||
STATUS_COLORS = {
|
||
"new": "#ff7a3d",
|
||
"confirmed": "#eab308",
|
||
"cooking": "#fb923c",
|
||
"ready": "#34d399",
|
||
"courier": "#a78bfa",
|
||
"completed": "#22c55e",
|
||
"cancelled": "#f87171",
|
||
}
|
||
|
||
|
||
def _uk_plural(n: int, one: str, few: str, many: str) -> str:
|
||
n_mod_100 = n % 100
|
||
if 11 <= n_mod_100 <= 14:
|
||
return many
|
||
n_mod_10 = n % 10
|
||
if n_mod_10 == 1:
|
||
return one
|
||
if 2 <= n_mod_10 <= 4:
|
||
return few
|
||
return many
|
||
|
||
|
||
def _rating_summary() -> dict:
|
||
counts = {n: 0 for n in range(1, 6)}
|
||
for (r,) in db.session.query(Order.rating).filter(Order.rating.isnot(None)):
|
||
if r in counts:
|
||
counts[r] += 1
|
||
total = sum(counts.values())
|
||
average = round(sum(n * c for n, c in counts.items()) / total, 1) if total else 0
|
||
max_bucket = max(counts.values()) if total else 0
|
||
return {
|
||
"count": total,
|
||
"count_label": _uk_plural(total, "оцінка", "оцінки", "оцінок"),
|
||
"average": average,
|
||
"average_rounded": round(average),
|
||
"distribution": counts,
|
||
"distribution_pct": {n: (round(c / max_bucket * 100) if max_bucket else 0) for n, c in counts.items()},
|
||
}
|
||
|
||
|
||
def _status_donut(status_counts: dict) -> dict:
|
||
total = sum(status_counts.values())
|
||
legend = []
|
||
stops = []
|
||
angle = 0.0
|
||
for status in ORDER_STATUSES:
|
||
count = status_counts.get(status, 0)
|
||
pct = (count / total * 100) if total else 0
|
||
color = STATUS_COLORS[status]
|
||
legend.append({"status": status, "label": ORDER_STATUS_LABELS[status], "count": count, "pct": round(pct)})
|
||
if count:
|
||
start = angle
|
||
end = angle + pct
|
||
stops.append(f"{color} {start:.2f}% {end:.2f}%")
|
||
angle = end
|
||
gradient = f"conic-gradient({', '.join(stops)})" if stops else "conic-gradient(var(--panel-2) 0 100%)"
|
||
return {"total": total, "legend": legend, "gradient": gradient}
|
||
|
||
|
||
@admin_bp.route("/")
|
||
@login_required
|
||
def dashboard():
|
||
tz = ZoneInfo(config.TIMEZONE)
|
||
now_local = datetime.now(tz)
|
||
today_start_local = now_local.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
yesterday_start_local = today_start_local - timedelta(days=1)
|
||
|
||
today_start = today_start_local.astimezone(timezone.utc).replace(tzinfo=None)
|
||
yesterday_start = yesterday_start_local.astimezone(timezone.utc).replace(tzinfo=None)
|
||
|
||
today_orders = Order.query.filter(Order.created_at >= today_start).all()
|
||
yesterday_orders = Order.query.filter(
|
||
Order.created_at >= yesterday_start, Order.created_at < today_start
|
||
).all()
|
||
|
||
# Cancelled orders never earned anything — kept out of every
|
||
# revenue/count KPI below, but still counted in the status donut
|
||
# further down (that chart's whole point is showing cancellations too).
|
||
today_active = [o for o in today_orders if o.status != "cancelled"]
|
||
yesterday_active = [o for o in yesterday_orders if o.status != "cancelled"]
|
||
|
||
def summarize(orders):
|
||
delivery = [o for o in orders if o.order_type == "delivery"]
|
||
pickup = [o for o in orders if o.order_type == "pickup"]
|
||
dine_in = [o for o in orders if o.order_type == "dine_in"]
|
||
revenue = sum(o.total for o in orders)
|
||
return {
|
||
"count": len(orders),
|
||
"delivery": len(delivery),
|
||
"pickup": len(pickup),
|
||
"dine_in": len(dine_in),
|
||
"revenue": revenue,
|
||
"avg_check": round(revenue / len(orders)) if orders else 0,
|
||
}
|
||
|
||
today_stats = summarize(today_active)
|
||
yesterday_stats = summarize(yesterday_active)
|
||
|
||
status_counts = {s: 0 for s in ORDER_STATUSES}
|
||
for o in today_orders:
|
||
status_counts[o.status] = status_counts.get(o.status, 0) + 1
|
||
|
||
recent_orders = Order.query.order_by(Order.created_at.desc()).limit(8).all()
|
||
donut = _status_donut(status_counts)
|
||
|
||
top_products: dict[str, int] = {}
|
||
for order in today_active:
|
||
for item in order.items:
|
||
top_products[item.product_name] = top_products.get(item.product_name, 0) + item.quantity
|
||
top_products_today = sorted(top_products.items(), key=lambda kv: kv[1], reverse=True)[:5]
|
||
top_products_max = max((qty for _, qty in top_products_today), default=1)
|
||
|
||
finance_today = {
|
||
"revenue": today_stats["revenue"],
|
||
"card": sum(o.total for o in today_active if o.payment_method == "card"),
|
||
"cash": sum(o.total for o in today_active if o.payment_method == "cash"),
|
||
"avg_check": today_stats["avg_check"],
|
||
"count": today_stats["count"],
|
||
}
|
||
|
||
stats = {
|
||
"today": today_stats,
|
||
"trend": {
|
||
"count": _pct_change(today_stats["count"], yesterday_stats["count"]),
|
||
"delivery": _pct_change(today_stats["delivery"], yesterday_stats["delivery"]),
|
||
"pickup": _pct_change(today_stats["pickup"], yesterday_stats["pickup"]),
|
||
"dine_in": _pct_change(today_stats["dine_in"], yesterday_stats["dine_in"]),
|
||
"revenue": _pct_change(today_stats["revenue"], yesterday_stats["revenue"]),
|
||
"avg_check": _pct_change(today_stats["avg_check"], yesterday_stats["avg_check"]),
|
||
},
|
||
"yesterday": yesterday_stats,
|
||
"new_orders": Order.query.filter_by(status="new").count(),
|
||
"products": Product.query.count(),
|
||
"categories": Category.query.count(),
|
||
"total_orders": Order.query.count(),
|
||
}
|
||
|
||
return render_template(
|
||
"admin/dashboard.html",
|
||
orders=recent_orders,
|
||
stats=stats,
|
||
donut=donut,
|
||
top_products_today=top_products_today,
|
||
top_products_max=top_products_max,
|
||
finance_today=finance_today,
|
||
delivery_chat_configured=bool(Setting.get("delivery_chat_id")),
|
||
pickup_chat_configured=bool(Setting.get("pickup_chat_id")),
|
||
dine_in_chat_configured=bool(Setting.get("dine_in_chat_id")),
|
||
ratings=_rating_summary(),
|
||
recent_reviews=Order.query.filter(Order.rating.isnot(None)).order_by(Order.rated_at.desc()).limit(5).all(),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Reviews (customer ratings)
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/reviews")
|
||
@login_required
|
||
def reviews_list():
|
||
score = request.args.get("score", type=int)
|
||
query = Order.query.filter(Order.rating.isnot(None))
|
||
if score in range(1, 6):
|
||
query = query.filter(Order.rating == score)
|
||
|
||
page = request.args.get("page", 1, type=int)
|
||
pagination = db.paginate(
|
||
query.order_by(Order.rated_at.desc()),
|
||
page=page,
|
||
per_page=50,
|
||
error_out=False,
|
||
)
|
||
return render_template(
|
||
"admin/reviews.html",
|
||
reviews=pagination.items,
|
||
pagination=pagination,
|
||
score=score,
|
||
ratings=_rating_summary(),
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Categories
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/categories")
|
||
@login_required
|
||
def categories_list():
|
||
categories = Category.query.order_by(Category.sort_order).all()
|
||
return render_template("admin/categories.html", categories=categories)
|
||
|
||
|
||
@admin_bp.route("/categories/new", methods=["POST"])
|
||
@admin_required
|
||
def categories_new():
|
||
name = request.form.get("name", "").strip()
|
||
slug = request.form.get("slug", "").strip() or _slugify(name)
|
||
if not name:
|
||
flash("Назва категорії обовʼязкова", "error")
|
||
return redirect(url_for("admin.categories_list"))
|
||
name_en = request.form.get("name_en", "").strip() or None
|
||
max_order = db.session.query(db.func.max(Category.sort_order)).scalar() or 0
|
||
db.session.add(Category(name=name, name_en=name_en, slug=slug, sort_order=max_order + 1))
|
||
db.session.commit()
|
||
log_action("category_create", f"Додано категорію «{name}»")
|
||
flash("Категорію додано", "success")
|
||
return redirect(url_for("admin.categories_list"))
|
||
|
||
|
||
@admin_bp.route("/categories/<int:category_id>/edit", methods=["POST"])
|
||
@admin_required
|
||
def categories_edit(category_id):
|
||
category = Category.query.get_or_404(category_id)
|
||
category.name = request.form.get("name", category.name).strip()
|
||
category.name_en = request.form.get("name_en", "").strip() or None
|
||
category.is_active = bool(request.form.get("is_active"))
|
||
db.session.commit()
|
||
log_action("category_edit", f"Оновлено категорію «{category.name}»")
|
||
flash("Категорію оновлено", "success")
|
||
return redirect(url_for("admin.categories_list"))
|
||
|
||
|
||
@admin_bp.route("/categories/<int:category_id>/delete", methods=["POST"])
|
||
@admin_required
|
||
def categories_delete(category_id):
|
||
category = Category.query.get_or_404(category_id)
|
||
name = category.name
|
||
db.session.delete(category)
|
||
db.session.commit()
|
||
log_action("category_delete", f"Видалено категорію «{name}»")
|
||
flash("Категорію видалено", "success")
|
||
return redirect(url_for("admin.categories_list"))
|
||
|
||
|
||
@admin_bp.route("/categories/<int:category_id>/move/<direction>", methods=["POST"])
|
||
@admin_required
|
||
def categories_move(category_id, direction):
|
||
categories = Category.query.order_by(Category.sort_order).all()
|
||
idx = next((i for i, c in enumerate(categories) if c.id == category_id), None)
|
||
if idx is not None:
|
||
swap_idx = idx - 1 if direction == "up" else idx + 1
|
||
if 0 <= swap_idx < len(categories):
|
||
categories[idx].sort_order, categories[swap_idx].sort_order = (
|
||
categories[swap_idx].sort_order,
|
||
categories[idx].sort_order,
|
||
)
|
||
db.session.commit()
|
||
return redirect(url_for("admin.categories_list"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Promo codes
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/promo-codes")
|
||
@admin_required
|
||
def promo_codes_list():
|
||
promos = PromoCode.query.order_by(PromoCode.created_at.desc()).all()
|
||
return render_template("admin/promo_codes.html", promos=promos)
|
||
|
||
|
||
@admin_bp.route("/promo-codes/new", methods=["POST"])
|
||
@admin_required
|
||
def promo_codes_new():
|
||
code = request.form.get("code", "").strip().upper()
|
||
if not code:
|
||
flash("Введіть код промокоду", "error")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
if PromoCode.query.filter(db.func.upper(PromoCode.code) == code).first():
|
||
flash(f"Промокод «{code}» вже існує", "error")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
discount_percent = request.form.get("discount_percent", type=int) or 0
|
||
if not (1 <= discount_percent <= 100):
|
||
flash("Знижка має бути від 1 до 100%", "error")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
expires_at = None
|
||
expires_raw = request.form.get("expires_at", "").strip()
|
||
if expires_raw:
|
||
try:
|
||
expires_at = datetime.strptime(expires_raw, "%Y-%m-%d").replace(hour=23, minute=59, second=59)
|
||
except ValueError:
|
||
flash("Невірний формат дати", "error")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
db.session.add(
|
||
PromoCode(
|
||
code=code,
|
||
discount_percent=discount_percent,
|
||
first_order_only=bool(request.form.get("first_order_only")),
|
||
min_order_amount=request.form.get("min_order_amount", type=int) or 0,
|
||
max_uses=request.form.get("max_uses", type=int),
|
||
expires_at=expires_at,
|
||
)
|
||
)
|
||
db.session.commit()
|
||
log_action("promo_code_create", f"Створено промокод «{code}» (−{discount_percent}%)")
|
||
flash("Промокод створено", "success")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
|
||
@admin_bp.route("/promo-codes/<int:promo_id>/toggle", methods=["POST"])
|
||
@admin_required
|
||
def promo_codes_toggle(promo_id):
|
||
promo = PromoCode.query.get_or_404(promo_id)
|
||
promo.is_active = not promo.is_active
|
||
db.session.commit()
|
||
log_action("promo_code_toggle", f"{'Увімкнено' if promo.is_active else 'Вимкнено'} промокод «{promo.code}»")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
|
||
@admin_bp.route("/promo-codes/<int:promo_id>/delete", methods=["POST"])
|
||
@admin_required
|
||
def promo_codes_delete(promo_id):
|
||
promo = PromoCode.query.get_or_404(promo_id)
|
||
code = promo.code
|
||
db.session.delete(promo)
|
||
db.session.commit()
|
||
log_action("promo_code_delete", f"Видалено промокод «{code}»")
|
||
flash("Промокод видалено", "success")
|
||
return redirect(url_for("admin.promo_codes_list"))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Products
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/products")
|
||
@login_required
|
||
def products_list():
|
||
category_id = request.args.get("category_id", type=int)
|
||
query = Product.query
|
||
if category_id:
|
||
query = query.filter_by(category_id=category_id)
|
||
products = query.order_by(Product.category_id, Product.sort_order).all()
|
||
categories = Category.query.order_by(Category.sort_order).all()
|
||
return render_template(
|
||
"admin/products.html", products=products, categories=categories, selected_category=category_id
|
||
)
|
||
|
||
|
||
@admin_bp.route("/products/new", methods=["GET", "POST"])
|
||
@admin_required
|
||
def products_new():
|
||
categories = Category.query.order_by(Category.sort_order).all()
|
||
if request.method == "POST":
|
||
product = Product(category_id=request.form.get("category_id", type=int))
|
||
_apply_product_form(product)
|
||
if product.pricing_mode == "per_100g" and not product.reference_weight_g:
|
||
flash("Вкажіть орієнтовну вагу порції для товару з ціною за 100 г", "error")
|
||
return redirect(url_for("admin.products_new"))
|
||
db.session.add(product)
|
||
db.session.commit()
|
||
log_action("product_create", f"Додано товар «{product.name}» ({product.effective_unit_price} грн)")
|
||
flash("Товар додано", "success")
|
||
return redirect(url_for("admin.products_list"))
|
||
return render_template("admin/product_form.html", product=None, categories=categories, allergens=ALLERGENS)
|
||
|
||
|
||
@admin_bp.route("/products/<int:product_id>/edit", methods=["GET", "POST"])
|
||
@admin_required
|
||
def products_edit(product_id):
|
||
product = Product.query.get_or_404(product_id)
|
||
categories = Category.query.order_by(Category.sort_order).all()
|
||
if request.method == "POST":
|
||
product.category_id = request.form.get("category_id", type=int)
|
||
_apply_product_form(product)
|
||
if product.pricing_mode == "per_100g" and not product.reference_weight_g:
|
||
flash("Вкажіть орієнтовну вагу порції для товару з ціною за 100 г", "error")
|
||
db.session.rollback()
|
||
return redirect(url_for("admin.products_edit", product_id=product_id))
|
||
db.session.commit()
|
||
log_action("product_edit", f"Оновлено товар «{product.name}»")
|
||
flash("Товар оновлено", "success")
|
||
return redirect(url_for("admin.products_list"))
|
||
return render_template("admin/product_form.html", product=product, categories=categories, allergens=ALLERGENS)
|
||
|
||
|
||
@admin_bp.route("/products/<int:product_id>/delete", methods=["POST"])
|
||
@admin_required
|
||
def products_delete(product_id):
|
||
product = Product.query.get_or_404(product_id)
|
||
name = product.name
|
||
db.session.delete(product)
|
||
db.session.commit()
|
||
log_action("product_delete", f"Видалено товар «{name}»")
|
||
flash("Товар видалено", "success")
|
||
return redirect(url_for("admin.products_list"))
|
||
|
||
|
||
@admin_bp.route("/products/<int:product_id>/toggle", methods=["POST"])
|
||
@admin_required
|
||
def products_toggle(product_id):
|
||
product = Product.query.get_or_404(product_id)
|
||
product.is_active = not product.is_active
|
||
log_action("product_toggle", f"Товар «{product.name}»: {'показано' if product.is_active else 'приховано'}")
|
||
db.session.commit()
|
||
return redirect(url_for("admin.products_list"))
|
||
|
||
|
||
def _apply_product_form(product: Product):
|
||
product.name = request.form.get("name", "").strip()
|
||
product.description = request.form.get("description", "").strip()
|
||
product.name_en = request.form.get("name_en", "").strip() or None
|
||
product.description_en = request.form.get("description_en", "").strip() or None
|
||
product.weight = request.form.get("weight", "").strip()
|
||
product.weight_en = request.form.get("weight_en", "").strip() or None
|
||
product.price = request.form.get("price", type=int) or 0
|
||
pricing_mode = request.form.get("pricing_mode", "fixed")
|
||
product.pricing_mode = pricing_mode if pricing_mode in ("fixed", "per_100g") else "fixed"
|
||
ref_weight = request.form.get("reference_weight_g", type=int)
|
||
product.reference_weight_g = ref_weight if product.pricing_mode == "per_100g" and ref_weight and ref_weight > 0 else None
|
||
price_override = request.form.get("estimated_price_override", type=int)
|
||
product.estimated_price_override = (
|
||
price_override if product.pricing_mode == "per_100g" and price_override and price_override > 0 else None
|
||
)
|
||
product.show_estimate_badge_price = bool(request.form.get("show_estimate_badge_price"))
|
||
product.show_estimate_badge_weight = bool(request.form.get("show_estimate_badge_weight"))
|
||
product.sort_order = request.form.get("sort_order", type=int) or 0
|
||
product.is_active = bool(request.form.get("is_active"))
|
||
product.allergens = ",".join(a for a in request.form.getlist("allergens") if a in ALLERGENS)
|
||
|
||
file = request.files.get("image")
|
||
if file and file.filename:
|
||
ext = file.filename.rsplit(".", 1)[-1].lower()
|
||
if ext in ALLOWED_IMAGE_EXT:
|
||
filename = f"{uuid.uuid4().hex}.{ext}"
|
||
upload_dir = Path(config.UPLOAD_FOLDER)
|
||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||
file.save(upload_dir / secure_filename(filename))
|
||
product.image_url = f"/static/uploads/{filename}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Orders
|
||
# ---------------------------------------------------------------------------
|
||
@admin_bp.route("/orders/new", methods=["GET", "POST"])
|
||
@login_required
|
||
def orders_new():
|
||
if request.method == "POST":
|
||
order_type = request.form.get("order_type")
|
||
if order_type not in ORDER_TYPES:
|
||
flash("Оберіть тип замовлення", "error")
|
||
return redirect(url_for("admin.orders_new"))
|
||
|
||
products_by_id = {p.id: p for p in Product.query.filter_by(is_active=True).all()}
|
||
product_ids = request.form.getlist("product_id")
|
||
quantities = request.form.getlist("quantity")
|
||
unit_prices = request.form.getlist("unit_price")
|
||
items = []
|
||
seen_ids = set()
|
||
for raw_id, raw_qty, raw_price in zip(product_ids, quantities, unit_prices):
|
||
if not raw_id:
|
||
continue
|
||
try:
|
||
pid = int(raw_id)
|
||
qty = int(raw_qty)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
product = products_by_id.get(pid)
|
||
if not product or qty <= 0 or pid in seen_ids:
|
||
continue
|
||
# unit_price lets staff override what's actually charged (e.g.
|
||
# after weighing a per_100g item) — falls back to the product's
|
||
# own price when left blank or entered invalid.
|
||
try:
|
||
unit_price = int(raw_price)
|
||
if unit_price < 0:
|
||
unit_price = product.effective_unit_price
|
||
except (TypeError, ValueError):
|
||
unit_price = product.effective_unit_price
|
||
seen_ids.add(pid)
|
||
items.append((product, qty, unit_price))
|
||
|
||
if not items:
|
||
flash("Додайте хоча б один товар", "error")
|
||
return redirect(url_for("admin.orders_new"))
|
||
|
||
payment_method = request.form.get("payment_method", "cash")
|
||
if payment_method not in PAYMENT_METHODS:
|
||
payment_method = "cash"
|
||
status = request.form.get("status", "new")
|
||
if status not in ORDER_STATUSES:
|
||
status = "new"
|
||
|
||
data = {
|
||
"name": request.form.get("customer_name", "").strip(),
|
||
"phone": request.form.get("customer_phone", "").strip(),
|
||
"payment_method": payment_method,
|
||
"payment_status": "paid" if request.form.get("payment_status") == "paid" else "unpaid",
|
||
"status": status,
|
||
"address_city": request.form.get("address_city", "").strip(),
|
||
"address_street": request.form.get("address_street", "").strip(),
|
||
"address_house": request.form.get("address_house", "").strip(),
|
||
"address_apartment": request.form.get("address_apartment", "").strip(),
|
||
"pickup_time": request.form.get("pickup_time", "").strip(),
|
||
"table_number": request.form.get("table_number", "").strip(),
|
||
}
|
||
|
||
order = order_service.create_manual_order(order_type, data, items)
|
||
log_action(
|
||
"order_manual_create",
|
||
f"Створено замовлення {order.display_number()} вручну ({ORDER_TYPE_LABELS[order_type]}, {current_user.username})",
|
||
)
|
||
checkbox_prro.fiscalize_in_background(order.id)
|
||
flash(f"Замовлення {order.display_number()} створено", "success")
|
||
return redirect(url_for("admin.orders_detail", order_id=order.id))
|
||
|
||
products = Product.query.filter_by(is_active=True).order_by(Product.name).all()
|
||
return render_template(
|
||
"admin/order_new.html",
|
||
products=products,
|
||
order_types=ORDER_TYPES,
|
||
order_type_labels=ORDER_TYPE_LABELS,
|
||
statuses=ORDER_STATUSES,
|
||
)
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>/edit", methods=["GET", "POST"])
|
||
@admin_required
|
||
def orders_edit(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
|
||
if request.method == "POST":
|
||
order_type = request.form.get("order_type")
|
||
if order_type not in ORDER_TYPES:
|
||
flash("Оберіть тип замовлення", "error")
|
||
return redirect(url_for("admin.orders_edit", order_id=order_id))
|
||
|
||
# Include products already on this order even if since deactivated,
|
||
# so editing an unrelated field doesn't silently drop a line whose
|
||
# product just got hidden from the menu in the meantime.
|
||
existing_product_ids = [i.product_id for i in order.items if i.product_id]
|
||
products_by_id = {
|
||
p.id: p
|
||
for p in Product.query.filter(db.or_(Product.is_active.is_(True), Product.id.in_(existing_product_ids))).all()
|
||
}
|
||
product_ids = request.form.getlist("product_id")
|
||
quantities = request.form.getlist("quantity")
|
||
unit_prices = request.form.getlist("unit_price")
|
||
items = []
|
||
seen_ids = set()
|
||
for raw_id, raw_qty, raw_price in zip(product_ids, quantities, unit_prices):
|
||
if not raw_id:
|
||
continue
|
||
try:
|
||
pid = int(raw_id)
|
||
qty = int(raw_qty)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
product = products_by_id.get(pid)
|
||
if not product or qty <= 0 or pid in seen_ids:
|
||
continue
|
||
# unit_price is whatever staff actually typed for this line —
|
||
# falls back to the product's own price only when left blank or
|
||
# invalid, NOT unconditionally re-snapshotted from the live
|
||
# catalog price like this used to do (that silently reset every
|
||
# line's charged price on any unrelated edit, e.g. fixing a
|
||
# phone number).
|
||
try:
|
||
unit_price = int(raw_price)
|
||
if unit_price < 0:
|
||
unit_price = product.effective_unit_price
|
||
except (TypeError, ValueError):
|
||
unit_price = product.effective_unit_price
|
||
seen_ids.add(pid)
|
||
items.append((product, qty, unit_price))
|
||
|
||
if not items:
|
||
flash("Додайте хоча б один товар", "error")
|
||
return redirect(url_for("admin.orders_edit", order_id=order_id))
|
||
|
||
payment_method = request.form.get("payment_method", "cash")
|
||
if payment_method not in PAYMENT_METHODS:
|
||
payment_method = "cash"
|
||
was_paid = order.payment_status == "paid"
|
||
now_paid = request.form.get("payment_status") == "paid"
|
||
|
||
# Reconcile stock around the item swap: undo whatever the old item
|
||
# list deducted, then deduct for the new one — skipped for a
|
||
# cancelled order, which has no stock reserved against it in the
|
||
# first place (cancelling already restored it).
|
||
reconcile_stock = inventory_enabled() and order.status != "cancelled"
|
||
if reconcile_stock:
|
||
restore_for_order(order)
|
||
|
||
OrderItem.query.filter_by(order_id=order.id).delete()
|
||
db.session.flush()
|
||
items_total = 0
|
||
for product, qty, unit_price in items:
|
||
db.session.add(
|
||
OrderItem(order_id=order.id, product_id=product.id, product_name=product.name, product_price=unit_price, quantity=qty)
|
||
)
|
||
items_total += unit_price * qty
|
||
|
||
if order_type == "delivery":
|
||
delivery_fee = calc_delivery_fee(items_total)
|
||
discount_amount = 0
|
||
elif order_type == "pickup":
|
||
delivery_fee = 0
|
||
discount_amount = calc_pickup_discount(items_total)
|
||
else:
|
||
delivery_fee = 0
|
||
discount_amount = 0
|
||
|
||
manual_discount = max(request.form.get("manual_discount_amount", type=int) or 0, 0)
|
||
|
||
order.order_type = order_type
|
||
order.customer_name = request.form.get("customer_name", "").strip() or "Гість"
|
||
order.customer_phone = request.form.get("customer_phone", "").strip() or None
|
||
order.address_city = request.form.get("address_city", "").strip() or None
|
||
order.address_street = request.form.get("address_street", "").strip() or None
|
||
order.address_house = request.form.get("address_house", "").strip() or None
|
||
order.address_apartment = request.form.get("address_apartment", "").strip() or None
|
||
order.pickup_time = request.form.get("pickup_time", "").strip() or None
|
||
order.table_number = request.form.get("table_number", "").strip() or None
|
||
order.payment_method = payment_method
|
||
order.payment_status = "paid" if now_paid else "unpaid"
|
||
order.items_total = items_total
|
||
order.delivery_fee = delivery_fee
|
||
order.discount_amount = discount_amount
|
||
order.manual_discount_amount = manual_discount
|
||
order.manual_discount_note = request.form.get("manual_discount_note", "").strip() or None
|
||
order.total = max(
|
||
0, items_total + delivery_fee - discount_amount - order.promo_discount_amount - manual_discount
|
||
)
|
||
if now_paid and not was_paid and order.paid_at is None:
|
||
order.paid_at = datetime.now(timezone.utc)
|
||
|
||
db.session.commit()
|
||
|
||
if reconcile_stock:
|
||
deduct_for_order(order)
|
||
|
||
log_action("order_edit", f"Замовлення {order.display_number()}: відредаговано вручну ({current_user.username})")
|
||
if now_paid and not was_paid:
|
||
checkbox_prro.fiscalize_in_background(order.id)
|
||
|
||
flash(f"Замовлення {order.display_number()} оновлено", "success")
|
||
return redirect(url_for("admin.orders_detail", order_id=order.id))
|
||
|
||
existing_product_ids = [i.product_id for i in order.items if i.product_id]
|
||
products = (
|
||
Product.query.filter(db.or_(Product.is_active.is_(True), Product.id.in_(existing_product_ids)))
|
||
.order_by(Product.name)
|
||
.all()
|
||
)
|
||
return render_template(
|
||
"admin/order_edit.html",
|
||
order=order,
|
||
products=products,
|
||
order_types=ORDER_TYPES,
|
||
order_type_labels=ORDER_TYPE_LABELS,
|
||
)
|
||
|
||
|
||
@admin_bp.route("/orders")
|
||
@login_required
|
||
def orders_list():
|
||
order_type = request.args.get("type")
|
||
status = request.args.get("status")
|
||
payment_status = request.args.get("payment_status")
|
||
q = request.args.get("q", "").strip()
|
||
|
||
query = Order.query
|
||
if order_type:
|
||
query = query.filter_by(order_type=order_type)
|
||
if status:
|
||
query = query.filter_by(status=status)
|
||
if payment_status:
|
||
query = query.filter_by(payment_status=payment_status)
|
||
if q:
|
||
like = f"%{q}%"
|
||
conditions = [Order.customer_name.ilike(like), Order.customer_phone.ilike(like)]
|
||
if q.lstrip("№#").isdigit():
|
||
conditions.append(Order.number == int(q.lstrip("№#")))
|
||
query = query.filter(db.or_(*conditions))
|
||
|
||
page = request.args.get("page", 1, type=int)
|
||
pagination = db.paginate(
|
||
query.order_by(Order.created_at.desc()),
|
||
page=page,
|
||
per_page=50,
|
||
error_out=False,
|
||
)
|
||
return render_template(
|
||
"admin/orders.html",
|
||
orders=pagination.items,
|
||
pagination=pagination,
|
||
statuses=ORDER_STATUSES,
|
||
order_type=order_type,
|
||
status=status,
|
||
payment_status=payment_status,
|
||
q=q,
|
||
)
|
||
|
||
|
||
def _order_timeline(order):
|
||
"""Steps for the visual progress stepper on the order detail page —
|
||
"courier" only appears for delivery orders."""
|
||
steps = ["new", "confirmed", "cooking", "ready"]
|
||
if order.order_type == "delivery":
|
||
steps.append("courier")
|
||
steps.append("completed")
|
||
|
||
if order.status == "cancelled":
|
||
current_index = -1 # nothing shown as done/current — see cancelled banner instead
|
||
else:
|
||
current_index = steps.index(order.status) if order.status in steps else -1
|
||
|
||
timeline = []
|
||
for i, key in enumerate(steps):
|
||
if current_index == -1:
|
||
state = "upcoming"
|
||
elif i < current_index:
|
||
state = "done"
|
||
elif i == current_index:
|
||
state = "current"
|
||
else:
|
||
state = "upcoming"
|
||
timeline.append({"key": key, "label": ORDER_STATUS_LABELS.get(key, key), "state": state})
|
||
return timeline
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>")
|
||
@login_required
|
||
def orders_detail(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
next_statuses = ORDER_STATUS_NEXT.get(order.status, [])
|
||
# "courier" only makes sense for delivery orders
|
||
if order.order_type != "delivery" and "courier" in next_statuses:
|
||
next_statuses = [s for s in next_statuses if s != "courier"]
|
||
return render_template(
|
||
"admin/order_detail.html",
|
||
order=order,
|
||
statuses=ORDER_STATUSES,
|
||
next_statuses=next_statuses,
|
||
timeline=_order_timeline(order),
|
||
checkbox_enabled=checkbox_prro.enabled(),
|
||
)
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>/status", methods=["POST"])
|
||
@login_required
|
||
def orders_status(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
new_status = request.form.get("status")
|
||
if new_status in ORDER_STATUSES:
|
||
old_status = order.status
|
||
if new_status == "cancelled":
|
||
order.cancel_reason = request.form.get("cancel_reason", "").strip() or None
|
||
order.status = new_status
|
||
db.session.commit()
|
||
if new_status == "cancelled" and old_status != "cancelled":
|
||
restore_for_order(order)
|
||
log_action(
|
||
"order_status_change",
|
||
f"Замовлення {order.display_number()}: «{ORDER_STATUS_LABELS.get(old_status, old_status)}» → "
|
||
f"«{ORDER_STATUS_LABELS.get(new_status, new_status)}»"
|
||
+ (f" (причина: {order.cancel_reason})" if order.cancel_reason and new_status == "cancelled" else ""),
|
||
)
|
||
notify_customer_status_change(order)
|
||
flash("Статус замовлення оновлено", "success")
|
||
return redirect(url_for("admin.orders_detail", order_id=order_id))
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>/mark-paid", methods=["POST"])
|
||
@login_required
|
||
def orders_mark_paid(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
if order.payment_method == "cash" and order.payment_status != "paid":
|
||
order.payment_status = "paid"
|
||
order.paid_at = datetime.now(timezone.utc)
|
||
db.session.commit()
|
||
log_action("order_mark_paid", f"Замовлення {order.display_number()}: позначено оплаченим (готівка)")
|
||
checkbox_prro.fiscalize_in_background(order.id)
|
||
flash_msg = "Замовлення позначено оплаченим"
|
||
if checkbox_prro.enabled():
|
||
flash_msg += " — чек формується у фоні, з'явиться на сторінці замовлення за кілька секунд"
|
||
flash(flash_msg, "success")
|
||
return redirect(request.referrer or url_for("admin.orders_detail", order_id=order_id))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Settings
|
||
# ---------------------------------------------------------------------------
|
||
CHECKBOX_SETTINGS = {
|
||
"allow_orders_outside_hours",
|
||
"force_closed",
|
||
"announcement_enabled",
|
||
"customer_status_notifications",
|
||
"order_rating_enabled",
|
||
"inventory_enabled",
|
||
"monobank_test_mode",
|
||
"monobank_show_basket",
|
||
"liqpay_sandbox_mode",
|
||
"checkbox_enabled",
|
||
"checkbox_deliver_receipt_to_customer",
|
||
"checkbox_auto_close_enabled",
|
||
"suppliers_manager_access",
|
||
"sidebar_collapsible_menu",
|
||
}
|
||
FILE_SETTINGS = {"logo_url"}
|
||
|
||
|
||
@admin_bp.route("/settings", methods=["GET", "POST"])
|
||
@admin_required
|
||
def settings_view():
|
||
if request.method == "POST":
|
||
for key in DEFAULT_SETTINGS.keys():
|
||
if key in FILE_SETTINGS:
|
||
continue
|
||
if key in CHECKBOX_SETTINGS:
|
||
Setting.set(key, "1" if request.form.get(key) else "0")
|
||
else:
|
||
Setting.set(key, request.form.get(key, DEFAULT_SETTINGS[key]))
|
||
|
||
logo_file = request.files.get("logo")
|
||
if logo_file and logo_file.filename:
|
||
ext = logo_file.filename.rsplit(".", 1)[-1].lower()
|
||
if ext in ALLOWED_IMAGE_EXT:
|
||
filename = f"logo-{uuid.uuid4().hex}.{ext}"
|
||
upload_dir = Path(config.UPLOAD_FOLDER)
|
||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||
logo_file.save(upload_dir / secure_filename(filename))
|
||
Setting.set("logo_url", f"/static/uploads/{filename}")
|
||
|
||
db.session.commit()
|
||
log_action("settings_update", "Оновлено налаштування системи")
|
||
flash("Налаштування збережено", "success")
|
||
return redirect(url_for("admin.settings_view"))
|
||
|
||
values = {key: Setting.get(key, default) for key, default in DEFAULT_SETTINGS.items()}
|
||
return render_template("admin/settings.html", values=values)
|
||
|
||
|
||
@admin_bp.route("/settings/test-monobank", methods=["POST"])
|
||
@admin_required
|
||
def settings_test_monobank():
|
||
from bober_bbq.payments.monobank import verify_token
|
||
|
||
field = request.args.get("field", "monobank_token")
|
||
if field not in ("monobank_token", "monobank_test_token"):
|
||
field = "monobank_token"
|
||
label = "Тестовий" if field == "monobank_test_token" else "Бойовий"
|
||
|
||
# Test whatever's in that field right now (even if not saved yet), so
|
||
# the admin can check a token before committing to it; if left blank,
|
||
# fall back to whatever's already saved for that specific field.
|
||
submitted = request.form.get(field, "").strip()
|
||
token = submitted or Setting.get(field, "")
|
||
ok, message = verify_token(token or None)
|
||
flash(f"{label} токен: {message}", "success" if ok else "error")
|
||
return redirect(url_for("admin.settings_view"))
|
||
|
||
|
||
@admin_bp.route("/settings/test-liqpay", methods=["POST"])
|
||
@admin_required
|
||
def settings_test_liqpay():
|
||
from bober_bbq.payments.liqpay import verify_token as liqpay_verify_token
|
||
|
||
# Same "test whatever's in the form right now, even if unsaved" UX as
|
||
# settings_test_monobank -- fall back to the saved value for whichever
|
||
# field was left blank.
|
||
public_key = request.form.get("liqpay_public_key", "").strip() or Setting.get("liqpay_public_key", "")
|
||
private_key = request.form.get("liqpay_private_key", "").strip() or Setting.get("liqpay_private_key", "")
|
||
ok, message = liqpay_verify_token(public_key or None, private_key or None)
|
||
flash(f"LiqPay: {message}", "success" if ok else "error")
|
||
return redirect(url_for("admin.settings_view"))
|
||
|
||
|
||
@admin_bp.route("/settings/test-checkbox", methods=["POST"])
|
||
@admin_required
|
||
def settings_test_checkbox():
|
||
login = request.form.get("checkbox_login", "").strip() or None
|
||
password = request.form.get("checkbox_password", "").strip() or None
|
||
ok, message = checkbox_prro.test_connection(login, password)
|
||
flash(f"Checkbox: {message}", "success" if ok else "error")
|
||
return redirect(url_for("admin.settings_view"))
|
||
|
||
|
||
@admin_bp.route("/settings/close-checkbox-shift", methods=["POST"])
|
||
@admin_required
|
||
def settings_close_checkbox_shift():
|
||
ok, message = checkbox_prro.close_shift()
|
||
flash(f"Зміна каси: {message}", "success" if ok else "error")
|
||
return redirect(url_for("admin.settings_view"))
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>/delete", methods=["POST"])
|
||
@admin_required
|
||
def orders_delete(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
display = order.display_number()
|
||
summary = f"{display} ({ORDER_TYPE_LABELS.get(order.order_type, order.order_type)}, {order.total} грн)"
|
||
|
||
# Cancelling already restores deducted stock — only do it again here if
|
||
# that never happened, or a cancelled order being deleted would get its
|
||
# ingredients credited back twice.
|
||
if order.status != "cancelled":
|
||
restore_for_order(order)
|
||
|
||
# StockMovement rows have no cascade from Order (they're a standalone
|
||
# ledger) — detach them instead of leaving order_id pointing at a row
|
||
# that no longer exists, which the "Рухи по складу" page already
|
||
# handles fine for movements that never had an order (manual adjustments).
|
||
StockMovement.query.filter_by(order_id=order.id).update({"order_id": None})
|
||
|
||
log_action("order_delete", f"Видалено замовлення {summary}")
|
||
db.session.delete(order)
|
||
db.session.commit()
|
||
flash(f"Замовлення {display} видалено остаточно", "success")
|
||
return redirect(url_for("admin.orders_list"))
|
||
|
||
|
||
@admin_bp.route("/orders/<int:order_id>/fiscalize", methods=["POST"])
|
||
@login_required
|
||
def orders_fiscalize(order_id):
|
||
order = Order.query.get_or_404(order_id)
|
||
checkbox_prro.fiscalize_in_background(order.id)
|
||
log_action("order_fiscalize", f"Замовлення {order.display_number()}: запущено формування чека")
|
||
flash("Чек формується у фоні — оновіть сторінку за кілька секунд", "success")
|
||
return redirect(url_for("admin.orders_detail", order_id=order.id))
|
||
|
||
|
||
def _slugify(name: str) -> str:
|
||
base = "".join(ch.lower() if ch.isalnum() else "-" for ch in name).strip("-")
|
||
slug = base or uuid.uuid4().hex[:8]
|
||
suffix = 1
|
||
candidate = slug
|
||
while Category.query.filter_by(slug=candidate).first():
|
||
suffix += 1
|
||
candidate = f"{slug}-{suffix}"
|
||
return candidate
|