hexdev/shveitechbot
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

catalogue operations, db migr and so on.

Browse Source 
pre release version.
This commit is contained in:
hexdev
2026-03-11 20:26:38 +07:00
parent 1af0cf1826
commit 7f4f8a883d
17 changed files with 284 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
alembic/versions/1d2e799ed0f3_product_img_path_constraints_redacted.py Normal file
View File

@@ -0,0 +1,39 @@
"""product img_path constraints redacted.
Revision ID: 1d2e799ed0f3
Revises: 41a6b57158ea
Create Date: 2026-03-05 20:33:46.459295
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "1d2e799ed0f3"
down_revision: Union[str, Sequence[str], None] = "41a6b57158ea"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column("products", "img_path", existing_type=sa.TEXT(), nullable=True)
op.drop_constraint(op.f("products_img_path_key"), "products", type_="unique")
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_unique_constraint(
op.f("products_img_path_key"),
"products",
["img_path"],
postgresql_nulls_not_distinct=False,
)
op.alter_column("products", "img_path", existing_type=sa.TEXT(), nullable=False)
# ### end Alembic commands ###

2
db/models/products.py
View File

@@ -24,7 +24,7 @@ class Product(Base):
name: Mapped[str] = mapped_column(Text, nullable=False)
description: Mapped[Optional[str]] = mapped_column(Text)
price: Mapped[int] = mapped_column()
img_path: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
img_path: Mapped[str] = mapped_column(Text, nullable=True)
file_id: Mapped[str] = mapped_column(Text, nullable=True, unique=True)
search_vector: Mapped[str] = mapped_column(TSVECTOR)

16
dto/catalogue.py
View File

@@ -2,6 +2,8 @@ from dataclasses import dataclass
from enum import Enum
from typing import Optional, Union
from aiogram.types import Message
from db.models import Category
from db.models.products import Product
@@ -22,3 +24,17 @@ class CatalogueView:
parent_id: Union[int, str, None] = None
has_next: bool = False
show_menu: bool = False
@dataclass
class NewCatalogueElement:
msg: Message
parent_id: Union[int, str]
@dataclass
class NewCategoryContext(NewCatalogueElement): ...
@dataclass
class NewProductContext(NewCatalogueElement): ...

3
handlers/admins/__init__.py
View File

@@ -1,5 +1,6 @@
from .inline_mode import router as inline_router
from .menu import router as menu_router
from .product_mgmt import router as product_router
from .creation import router as creation_router
routers = [inline_router, menu_router, product_router]
routers = [inline_router, menu_router, product_router, creation_router]

70
handlers/admins/creation.py Normal file
View File

@@ -0,0 +1,70 @@
from aiogram import F, Router
from aiogram.types import CallbackQuery, Message
from aiogram.fsm.context import FSMContext
from sqlalchemy.ext.asyncio import AsyncSession
from dto.catalogue import NewCategoryContext
from misc.filters import IsVerified
from misc.kb.client import get_back_to_catalogue
from misc.mapper import parse_cat_id
from misc.states import AdminControlStorage
from misc.kb.common import return_menu
from repositories.categories import CategoriesRepository
from services.catalogue import CatalogueService
router = Router()
@router.callback_query(IsVerified(), F.data.startswith("new:"))
async def creation_init(
cb: CallbackQuery,
state: FSMContext,
session: AsyncSession,
catalogue_service: CatalogueService,
):
await state.clear()
data = cb.data.split(":")
mode = data[1]
cat_id = data[2]
path = await catalogue_service.get_path(session, cat_id)
if mode == "category":
await cb.message.edit_text(
f"{path} <i>-> ...</i>\n\n<b>🏷️ Введите название новой категории.</b>",
reply_markup=get_back_to_catalogue(cat_id),
)
ctx = NewCategoryContext(cb.message, cat_id)
await state.set_state(AdminControlStorage.new_category)
await state.set_data({"ctx": ctx})
if mode == "product":
...
@router.message(AdminControlStorage.new_category)
async def create_new_category(
msg: Message,
state: FSMContext,
session: AsyncSession,
categories_repo: CategoriesRepository,
):
await msg.delete()
data = await state.get_data()
ctx: NewCategoryContext | None = data.get("ctx")
if not ctx:
await msg.reply("❌ Ошибка при создании категории.", reply_markup=return_menu)
return
category = await categories_repo.add_category(
session, name=msg.text, parent_id=parse_cat_id(ctx.parent_id)
)
await ctx.msg.edit_text(
f"✅ Категория {msg.text} создана успешно.",
reply_markup=get_back_to_catalogue(category.id, text="🛒 Перейти"),
)

7
handlers/client/catalogue.py
View File

@@ -26,15 +26,16 @@ async def subcatalogue(
view = await catalogue_service.build_category_view(session, cat_id)
kb = render_catalogue(view, is_admin=is_admin)
path = await catalogue_service.get_path(session, cat_id)
if cb.message.photo:
await cb.message.delete()
await cb.answer(
"category selection",
await cb.message.answer(
f"{path}",
reply_markup=kb,
)
return
await cb.message.edit_text(
"category selection",
f"{path}",
reply_markup=kb,
)

32
handlers/client/products.py
View File

@@ -1,7 +1,9 @@
import os
import logging
from re import Match
from aiogram import Router, F
from aiogram.types import CallbackQuery, FSInputFile
from aiogram.exceptions import TelegramBadRequest
from aiogram.fsm.context import FSMContext
from sqlalchemy.ext.asyncio import AsyncSession
@@ -17,6 +19,7 @@ from services.catalogue import CatalogueService
from services.orders import OrderService
router = Router()
logger = logging.getLogger(__name__)
@router.callback_query(
@@ -73,17 +76,9 @@ async def product_card(
cart_amount=order_item,
)
if product.file_id:
await cb.message.delete()
msg = await cb.message.answer_photo(
product.file_id,
caption=get_product_description(product),
reply_markup=kb,
)
return
if not os.path.isfile(product.img_path):
if not product.file_id and not (
product.img_path and os.path.isfile(product.img_path)
):
await cb.message.edit_text(
get_product_description(product),
reply_markup=kb,
@@ -91,6 +86,20 @@ async def product_card(
return
await cb.message.delete()
if product.file_id:
msg = await cb.message.answer_photo(
product.file_id,
caption=get_product_description(product),
reply_markup=kb,
)
return
try:
await cb.message.edit_text("<b>⏳ Загружаем карточку товара...</b>")
except TelegramBadRequest:
...
msg = await cb.message.answer_photo(
FSInputFile(product.img_path),
caption=get_product_description(product),
@@ -98,6 +107,7 @@ async def product_card(
)
if not msg.photo:
logger.warning("didnt get a photo back wahhh :(")
return # TODO: Add logging
await products_repo.add_product_file_id(

8
main.py
View File

@@ -4,6 +4,7 @@ import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from redis.asyncio import Redis
from aiogram.client.session.aiohttp import AiohttpSession
from config import BOT_TOKEN, REDIS_URL
from handlers import admins_routers, client_routers
@@ -31,7 +32,12 @@ os.makedirs("static/img", exist_ok=True)
async def main():
dp = Dispatcher()
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode="HTML"))
aiohttp_session = AiohttpSession(proxy="http://127.0.0.1:2080")
bot = Bot(
token=BOT_TOKEN,
default=DefaultBotProperties(parse_mode="HTML"),
session=aiohttp_session,
)
# DB services creation
order_repository = OrderRepository()

13
misc/kb/admins.py
View File

@@ -1,21 +1,18 @@
from typing import Optional, Callable
from typing import Optional, Callable, Union
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
main_menu: InlineKeyboardMarkup = InlineKeyboardBuilder(
[
[InlineKeyboardButton(text="📦 Изменить товары", callback_data="cat:root")],
[
InlineKeyboardButton(
text="📜 Изменить категории", callback_data="edit:categories"
)
],
[InlineKeyboardButton(text="📦 Изменить каталог", callback_data="cat:root")],
# [InlineKeyboardButton(text="", callback_data="")],
]
).as_markup()
def back_to_product_kb(product_id: int, *, cb_factory: Callable[[int], str]):
def back_to_product_kb(
product_id: Union[int, str], *, cb_factory: Callable[[int | str], str]
):
return InlineKeyboardBuilder(
[[InlineKeyboardButton(text="⬅️", callback_data=cb_factory(product_id))]]
).as_markup()

59
misc/kb/client.py
View File

@@ -1,4 +1,4 @@
from typing import Optional, Callable
from typing import Optional, Callable, Union
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
@@ -7,7 +7,7 @@ from config import PAGE_SIZE
from db.models import Category, Product
from dto.cart import CartItemDTO
from dto.catalogue import CatalogueView, CatalogueType
from .common import return_menu
from misc.mapper import serialize_cat_id
def pagination_row(
@@ -37,11 +37,33 @@ def cart_btn(cart_items: int) -> InlineKeyboardButton:
def get_back_to_catalogue(
cat_id: int, *, text: Optional[str] = "⬅️ Назад"
parent_id: int,
*,
text: Optional[str] = "⬅️ Назад",
show_controls: bool = False,
cat_id: Optional[Union[str, int]] = None,
fallback_cb: Optional[str] = "menu:main",
) -> InlineKeyboardMarkup:
return InlineKeyboardBuilder(
[[InlineKeyboardButton(text=text, callback_data=f"cat:{cat_id}")]]
).as_markup()
builder = InlineKeyboardBuilder()
if show_controls and cat_id:
builder.row(
InlineKeyboardButton(
text="➕ Категория", callback_data=f"new:category:{cat_id}"
),
InlineKeyboardButton(
text="➕ Товар", callback_data=f"new:product:{cat_id}"
),
)
builder.row(
InlineKeyboardButton(
text=text, callback_data=f"cat:{parent_id}" if parent_id else fallback_cb
),
)
return builder.as_markup()
def main_menu_kb(cart_items: int) -> InlineKeyboardMarkup:
@@ -58,9 +80,10 @@ def main_menu_kb(cart_items: int) -> InlineKeyboardMarkup:
def render_category(
children: list[Category],
parent_id: Optional[int] = None,
parent_id: Optional[Union[str, int]] = None,
show_menu: bool = False,
*,
cat_id: Optional[Union[str, int]] = None,
category_cb_factory: Callable[[Category], str],
is_admin: bool = False,
) -> InlineKeyboardMarkup:
@@ -75,12 +98,16 @@ def render_category(
nav: list[list[InlineKeyboardButton]] = []
if is_admin:
cat_id = serialize_cat_id(cat_id)
nav.append(
[
InlineKeyboardButton(
text="📄 Добавить категорию",
callback_data=f"new:category:{parent_id}",
)
text="➕ Категория",
callback_data=f"new:category:{cat_id}",
),
InlineKeyboardButton(
text="✍️", callback_data=f"edit:category:{cat_id}"
),
]
)
@@ -124,9 +151,7 @@ def render_products(
if is_admin:
builder.row(
InlineKeyboardButton(
text="📦 Добавить товар", callback_data=f"new:product:{cat_id}"
)
InlineKeyboardButton(text="➕ Товар", callback_data=f"new:product:{cat_id}")
)
nav_buttons = []
@@ -144,11 +169,12 @@ def render_products(
def render_catalogue(view: CatalogueView, *, is_admin: bool):
if view.view_type == CatalogueType.CATEGORIES:
return render_category(
children=view.children[:PAGE_SIZE], # type: ignore
children=view.children,
parent_id=view.parent_id,
show_menu=view.show_menu,
category_cb_factory=lambda c: f"cat:{c.id}",
is_admin=is_admin,
cat_id=view.category,
)
if view.view_type == CatalogueType.PRODUCTS:
return render_products(
@@ -162,9 +188,12 @@ def render_catalogue(view: CatalogueView, *, is_admin: bool):
f"products:{view.category}:{view.page + 1}" if view.has_next else None
),
is_admin=is_admin,
cat_id=view.category,
)
return return_menu
return get_back_to_catalogue(
view.parent_id, show_controls=is_admin, cat_id=serialize_cat_id(view.category)
)
def render_product_interactions(

11
misc/mapper.py Normal file
View File

@@ -0,0 +1,11 @@
def parse_cat_id(value: str) -> int | None | bool:
if value == "root":
return None
if str(value).isdigit():
return int(value)
raise Exception("Invalid Category Id.")
def serialize_cat_id(cat_id: int | None) -> str:
return "root" if cat_id is None else str(cat_id)

1
misc/states.py
View File

@@ -18,3 +18,4 @@ class SearchStorage(StatesGroup):
class AdminControlStorage(StatesGroup):
edit_product = State()
new_category = State()

5
misc/texts.py
View File

@@ -24,3 +24,8 @@ def get_order_item_list(cart: CartDTO):
text += f"\n━━━━━━━━━━━━━━\n💰 Итого: {cart.total}₽"
return text
def get_breadcrumps_path(products: list[Product]) -> str:
products_str = f" -> {' -> '.join([x.name for x in products])}" if products else ""
return f"<b>🏠 Каталог{products_str}</b>"

36
repositories/categories.py
View File

@@ -1,6 +1,7 @@
from typing import Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Optional
from sqlalchemy.orm import aliased
from db.models import Category
@@ -19,3 +20,36 @@ class CategoriesRepository:
result = await session.scalars(stmt)
return list(result)
async def get_category_path(self, session: AsyncSession, category_id: int):
base = select(Category.id, Category.parent_id, Category.name).where(
Category.id == category_id
)
cte = base.cte(name="path", recursive=True)
parent = aliased(Category)
cte = cte.union_all(
select(parent.id, parent.parent_id, parent.name).join(
cte, cte.c.parent_id == parent.id
)
)
stmt = select(cte)
result = await session.execute(stmt)
rows = result.all()
return list(reversed(rows))
async def add_category(
self, session: AsyncSession, *, name: str, parent_id: Optional[int]
) -> Category:
category = Category(name=name, parent_id=parent_id)
session.add(category)
await session.commit()
return category

12
repositories/products.py
View File

@@ -1,5 +1,5 @@
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from sqlalchemy import select
from sqlalchemy.sql.expression import func
from typing import Optional
@@ -32,13 +32,9 @@ class ProductRepository:
async def add_product_file_id(
self, session: AsyncSession, product_id: int, *, file_id: str
) -> None:
stmt = (
update(Product)
.where(Product.id == product_id)
.where(Product.file_id is None)
.values(file_id=file_id)
)
await session.execute(stmt)
product = await self.get_product_by_id(session, product_id)
product.file_id = file_id
await session.commit()
async def search(

1
requirements.txt
View File

@@ -8,3 +8,4 @@ psycopg2
phonenumbers
ruff
redis
aiohttp-socks

30
services/catalogue.py
View File

@@ -2,6 +2,8 @@ from typing import Union, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from dto.catalogue import CatalogueType, CatalogueView
from misc.mapper import parse_cat_id, serialize_cat_id
from misc.texts import get_breadcrumps_path
from repositories import ProductRepository, CategoriesRepository
from config import PAGE_SIZE
@@ -16,11 +18,7 @@ class CatalogueService:
async def build_category_view(
self, session: AsyncSession, cat_id: Optional[Union[int, str]], page: int = 0
) -> CatalogueView:
cat_id = (
int(cat_id)
if isinstance(cat_id, (int, str)) and str(cat_id).isdigit()
else None
)
cat_id = parse_cat_id(cat_id)
view = CatalogueView(page=page)
children = await self.categories_repo.get_categories_by_parent_id(
@@ -33,22 +31,28 @@ class CatalogueService:
view.show_menu = True
else: # Subcategories
view.category = cat_id
view.parent_id = category.parent_id or "root"
view.parent_id = serialize_cat_id(category.parent_id)
if not children: # Products
view.view_type = CatalogueType.PRODUCTS
products = await self.product_repo.get_product_by_category(
session, cat_id, limit=PAGE_SIZE + 1, offset=view.page * PAGE_SIZE
)
if products:
view.view_type = CatalogueType.PRODUCTS
view.products = products[:PAGE_SIZE]
view.has_next = len(products) > PAGE_SIZE
if not products:
view.view_type = CatalogueType.EMPTY
view.products = products[:PAGE_SIZE]
view.has_next = len(products) > PAGE_SIZE
return view
view.view_type = CatalogueType.EMPTY
return view
view.view_type = CatalogueType.CATEGORIES
view.children = children[:PAGE_SIZE]
view.children = children
return view
async def get_path(self, session: AsyncSession, cat_id: Union[int, str]):
category = int(cat_id) if isinstance(cat_id, int) or cat_id.isdigit() else None
products = await self.categories_repo.get_category_path(session, category)
return get_breadcrumps_path(products)