130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
"""Логика работы с шаблонами."""
|
||
from typing import Dict, Any, Optional, List
|
||
|
||
from sqlalchemy import select
|
||
from app.db.session import async_session_maker
|
||
from app.models.templates import Template
|
||
from app.models.post import PostType
|
||
from app.bots.editor.messages import MessageType
|
||
|
||
class TemplateService:
|
||
@staticmethod
|
||
async def list_user_templates(owner_id: int) -> List[Template]:
|
||
"""Получить список шаблонов пользователя."""
|
||
async with async_session_maker() as session:
|
||
query = select(Template).where(Template.owner_id == owner_id)
|
||
result = await session.execute(query)
|
||
return list(result.scalars())
|
||
|
||
@staticmethod
|
||
async def get_template(template_id: str) -> Optional[Template]:
|
||
"""Получить шаблон по ID."""
|
||
async with async_session_maker() as session:
|
||
query = select(Template).where(Template.id == template_id)
|
||
result = await session.execute(query)
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
async def list_templates(owner_id: Optional[int] = None, limit: Optional[int] = None, offset: Optional[int] = None) -> list[Template]:
|
||
"""Получить список всех шаблонов.
|
||
|
||
Args:
|
||
owner_id: Опциональный ID владельца
|
||
|
||
Returns:
|
||
List[Template]: Список шаблонов
|
||
"""
|
||
async with async_session_maker() as session:
|
||
query = select(Template)
|
||
if owner_id is not None:
|
||
query = query.where(Template.owner_id == owner_id)
|
||
if offset is not None:
|
||
query = query.offset(offset)
|
||
if limit is not None:
|
||
query = query.limit(limit)
|
||
result = await session.execute(query)
|
||
return list(result.scalars())
|
||
|
||
from app.services.users import get_or_create_user
|
||
|
||
async def create_template(template_data: Dict[str, Any]) -> Template:
|
||
"""Создать новый шаблон.
|
||
|
||
Args:
|
||
template_data: Данные шаблона
|
||
|
||
Returns:
|
||
Template: Созданный шаблон
|
||
"""
|
||
# Проверяем owner_id и создаем пользователя если нужно
|
||
tg_user_id = template_data.pop("tg_user_id", None)
|
||
if tg_user_id:
|
||
user = await get_or_create_user(tg_user_id)
|
||
template_data["owner_id"] = user.id
|
||
|
||
async with async_session_maker() as session:
|
||
template = Template(**template_data)
|
||
session.add(template)
|
||
await session.commit()
|
||
return template
|
||
|
||
async def render_template_by_name(
|
||
name: str,
|
||
template_vars: Dict[str, Any],
|
||
context: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
"""Рендеринг шаблона по имени.
|
||
|
||
Args:
|
||
name: Имя шаблона
|
||
template_vars: Переменные для подстановки
|
||
context: Дополнительный контекст
|
||
|
||
Returns:
|
||
Dict[str, Any]: Отрендеренные данные для поста
|
||
"""
|
||
async with async_session_maker() as session:
|
||
stmt = Template.__table__.select().where(Template.__table__.c.name == name)
|
||
result = await session.execute(stmt)
|
||
template = result.scalar_one_or_none()
|
||
|
||
if not template:
|
||
raise ValueError(f"Шаблон {name} не найден")
|
||
|
||
text = template.content
|
||
keyboard = template.keyboard_tpl
|
||
|
||
# Подстановка переменных
|
||
for key, value in template_vars.items():
|
||
text = text.replace(f"{{${key}}}", str(value))
|
||
|
||
# Проверяем тип и конвертируем в MessageType
|
||
message_type = MessageType.TEXT
|
||
if template.type == PostType.photo:
|
||
message_type = MessageType.PHOTO
|
||
elif template.type == PostType.video:
|
||
message_type = MessageType.VIDEO
|
||
|
||
return {
|
||
"type": message_type,
|
||
"text": text,
|
||
"keyboard": keyboard,
|
||
"parse_mode": template.parse_mode
|
||
}
|
||
|
||
async def count_templates(owner_id: Optional[int] = None) -> int:
|
||
"""Посчитать количество шаблонов.
|
||
|
||
Args:
|
||
owner_id: Опциональный ID владельца
|
||
|
||
Returns:
|
||
int: Количество шаблонов
|
||
"""
|
||
async with async_session_maker() as session:
|
||
query = Template.__table__.select()
|
||
if owner_id is not None:
|
||
query = query.where(Template.__table__.c.owner_id == owner_id)
|
||
result = await session.execute(query)
|
||
return len(list(result.scalars()))
|