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), )