415 lines
13 KiB
Python
415 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
from fastapi import APIRouter, Depends, Form, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import delete, func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import is_admin, login_admin, logout_admin
|
|
from app.cart import available_stock
|
|
from app.config import get_settings
|
|
from app.database import get_db
|
|
from app.models import Category, Order, Product, ProductKey
|
|
|
|
router = APIRouter(prefix="/admin")
|
|
settings = get_settings()
|
|
|
|
|
|
def setup(templates: Jinja2Templates) -> None:
|
|
router.templates = templates # type: ignore[attr-defined]
|
|
|
|
|
|
def admin_ctx(request: Request, **kwargs):
|
|
data = {
|
|
"request": request,
|
|
"app_name": settings.app_name,
|
|
"admin_user": request.session.get("admin_user", "admin"),
|
|
}
|
|
data.update(kwargs)
|
|
return data
|
|
|
|
|
|
def guard(request: Request):
|
|
if not is_admin(request):
|
|
return RedirectResponse("/admin/login", status_code=303)
|
|
return None
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
def admin_login_page(request: Request):
|
|
if is_admin(request):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/login.html",
|
|
admin_ctx(request, error=None),
|
|
)
|
|
|
|
|
|
@router.post("/login", response_class=HTMLResponse)
|
|
def admin_login_submit(
|
|
request: Request,
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
):
|
|
if login_admin(request, username.strip(), password):
|
|
return RedirectResponse("/admin", status_code=303)
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/login.html",
|
|
admin_ctx(request, error="Неверный логин или пароль"),
|
|
status_code=401,
|
|
)
|
|
|
|
|
|
@router.post("/logout")
|
|
def admin_logout(request: Request):
|
|
logout_admin(request)
|
|
return RedirectResponse("/admin/login", status_code=303)
|
|
|
|
|
|
@router.get("", response_class=HTMLResponse)
|
|
@router.get("/", response_class=HTMLResponse)
|
|
def dashboard(request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
|
|
products_count = db.scalar(select(func.count()).select_from(Product)) or 0
|
|
keys_free = (
|
|
db.scalar(
|
|
select(func.count())
|
|
.select_from(ProductKey)
|
|
.where(ProductKey.is_sold.is_(False))
|
|
)
|
|
or 0
|
|
)
|
|
orders_count = db.scalar(select(func.count()).select_from(Order)) or 0
|
|
revenue = (
|
|
db.scalar(
|
|
select(func.coalesce(func.sum(Order.total), 0)).where(
|
|
Order.status == "delivered"
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
recent_orders = db.scalars(select(Order).order_by(Order.id.desc()).limit(8)).all()
|
|
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/dashboard.html",
|
|
admin_ctx(
|
|
request,
|
|
products_count=products_count,
|
|
keys_free=keys_free,
|
|
orders_count=orders_count,
|
|
revenue=revenue,
|
|
recent_orders=recent_orders,
|
|
),
|
|
)
|
|
|
|
|
|
@router.get("/products", response_class=HTMLResponse)
|
|
def products_list(request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
products = db.scalars(select(Product).order_by(Product.id.desc())).all()
|
|
stock_map = {p.id: available_stock(db, p.id) for p in products}
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/products.html",
|
|
admin_ctx(request, products=products, stock_map=stock_map),
|
|
)
|
|
|
|
|
|
@router.get("/products/new", response_class=HTMLResponse)
|
|
def product_new(request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
categories = db.scalars(select(Category).order_by(Category.sort_order)).all()
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=None,
|
|
categories=categories,
|
|
error=None,
|
|
form={},
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/products/new", response_class=HTMLResponse)
|
|
def product_create(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
title: str = Form(...),
|
|
slug: str = Form(...),
|
|
short_description: str = Form(...),
|
|
description: str = Form(""),
|
|
price: str = Form(...),
|
|
category_slug: str = Form(...),
|
|
image_gradient: str = Form("mint"),
|
|
delivery_note: str = Form("Мгновенная выдача"),
|
|
is_featured: str | None = Form(None),
|
|
is_active: str | None = Form(None),
|
|
):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
categories = db.scalars(select(Category).order_by(Category.sort_order)).all()
|
|
form = {
|
|
"title": title,
|
|
"slug": slug,
|
|
"short_description": short_description,
|
|
"description": description,
|
|
"price": price,
|
|
"category_slug": category_slug,
|
|
"image_gradient": image_gradient,
|
|
"delivery_note": delivery_note,
|
|
"is_featured": bool(is_featured),
|
|
"is_active": bool(is_active) if is_active is not None else True,
|
|
}
|
|
try:
|
|
price_val = Decimal(price.replace(",", ".").replace(" ", ""))
|
|
except (InvalidOperation, AttributeError):
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=None,
|
|
categories=categories,
|
|
error="Некорректная цена",
|
|
form=form,
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
slug_clean = slug.strip().lower().replace(" ", "-")
|
|
exists = db.scalar(select(Product.id).where(Product.slug == slug_clean))
|
|
if exists:
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=None,
|
|
categories=categories,
|
|
error="Slug уже занят",
|
|
form=form,
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
product = Product(
|
|
title=title.strip(),
|
|
slug=slug_clean,
|
|
short_description=short_description.strip()[:300],
|
|
description=description.strip(),
|
|
price=price_val,
|
|
category_slug=category_slug,
|
|
image_gradient=image_gradient,
|
|
delivery_note=delivery_note.strip()[:200],
|
|
is_featured=bool(is_featured),
|
|
is_active=bool(is_active),
|
|
)
|
|
db.add(product)
|
|
db.commit()
|
|
return RedirectResponse(f"/admin/products/{product.id}/keys", status_code=303)
|
|
|
|
|
|
@router.get("/products/{product_id}/edit", response_class=HTMLResponse)
|
|
def product_edit(product_id: int, request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
product = db.get(Product, product_id)
|
|
if not product:
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
categories = db.scalars(select(Category).order_by(Category.sort_order)).all()
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=product,
|
|
categories=categories,
|
|
error=None,
|
|
form={},
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/products/{product_id}/edit", response_class=HTMLResponse)
|
|
def product_update(
|
|
product_id: int,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
title: str = Form(...),
|
|
slug: str = Form(...),
|
|
short_description: str = Form(...),
|
|
description: str = Form(""),
|
|
price: str = Form(...),
|
|
category_slug: str = Form(...),
|
|
image_gradient: str = Form("mint"),
|
|
delivery_note: str = Form("Мгновенная выдача"),
|
|
is_featured: str | None = Form(None),
|
|
is_active: str | None = Form(None),
|
|
):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
product = db.get(Product, product_id)
|
|
if not product:
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
categories = db.scalars(select(Category).order_by(Category.sort_order)).all()
|
|
|
|
try:
|
|
price_val = Decimal(price.replace(",", ".").replace(" ", ""))
|
|
except (InvalidOperation, AttributeError):
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=product,
|
|
categories=categories,
|
|
error="Некорректная цена",
|
|
form={},
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
slug_clean = slug.strip().lower().replace(" ", "-")
|
|
clash = db.scalar(
|
|
select(Product.id).where(Product.slug == slug_clean, Product.id != product.id)
|
|
)
|
|
if clash:
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/product_form.html",
|
|
admin_ctx(
|
|
request,
|
|
product=product,
|
|
categories=categories,
|
|
error="Slug уже занят",
|
|
form={},
|
|
),
|
|
status_code=400,
|
|
)
|
|
|
|
product.title = title.strip()
|
|
product.slug = slug_clean
|
|
product.short_description = short_description.strip()[:300]
|
|
product.description = description.strip()
|
|
product.price = price_val
|
|
product.category_slug = category_slug
|
|
product.image_gradient = image_gradient
|
|
product.delivery_note = delivery_note.strip()[:200]
|
|
product.is_featured = bool(is_featured)
|
|
product.is_active = bool(is_active)
|
|
db.commit()
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
|
|
|
|
@router.post("/products/{product_id}/delete")
|
|
def product_delete(product_id: int, request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
product = db.get(Product, product_id)
|
|
if product:
|
|
db.execute(delete(ProductKey).where(ProductKey.product_id == product_id))
|
|
db.delete(product)
|
|
db.commit()
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
|
|
|
|
@router.get("/products/{product_id}/keys", response_class=HTMLResponse)
|
|
def product_keys(product_id: int, request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
product = db.get(Product, product_id)
|
|
if not product:
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
keys = db.scalars(
|
|
select(ProductKey)
|
|
.where(ProductKey.product_id == product_id)
|
|
.order_by(ProductKey.is_sold, ProductKey.id.desc())
|
|
).all()
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/keys.html",
|
|
admin_ctx(
|
|
request,
|
|
product=product,
|
|
keys=keys,
|
|
stock=available_stock(db, product_id),
|
|
message=None,
|
|
),
|
|
)
|
|
|
|
|
|
@router.post("/products/{product_id}/keys", response_class=HTMLResponse)
|
|
def product_keys_add(
|
|
product_id: int,
|
|
request: Request,
|
|
codes: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
product = db.get(Product, product_id)
|
|
if not product:
|
|
return RedirectResponse("/admin/products", status_code=303)
|
|
|
|
lines = [line.strip() for line in codes.splitlines() if line.strip()]
|
|
added = 0
|
|
for code in lines:
|
|
db.add(ProductKey(product_id=product_id, code=code))
|
|
added += 1
|
|
db.commit()
|
|
|
|
keys = db.scalars(
|
|
select(ProductKey)
|
|
.where(ProductKey.product_id == product_id)
|
|
.order_by(ProductKey.is_sold, ProductKey.id.desc())
|
|
).all()
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/keys.html",
|
|
admin_ctx(
|
|
request,
|
|
product=product,
|
|
keys=keys,
|
|
stock=available_stock(db, product_id),
|
|
message=f"Добавлено ключей: {added}",
|
|
),
|
|
)
|
|
|
|
|
|
@router.get("/orders", response_class=HTMLResponse)
|
|
def orders_list(request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
orders = db.scalars(select(Order).order_by(Order.id.desc()).limit(100)).all()
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/orders.html",
|
|
admin_ctx(request, orders=orders),
|
|
)
|
|
|
|
|
|
@router.get("/orders/{order_id}", response_class=HTMLResponse)
|
|
def order_detail(order_id: int, request: Request, db: Session = Depends(get_db)):
|
|
if (redir := guard(request)):
|
|
return redir
|
|
order = db.get(Order, order_id)
|
|
if not order:
|
|
return RedirectResponse("/admin/orders", status_code=303)
|
|
return router.templates.TemplateResponse( # type: ignore[attr-defined]
|
|
request,
|
|
"admin/order_detail.html",
|
|
admin_ctx(request, order=order),
|
|
)
|