Sergey Vanyushkin
7ff3fa0992
All checks were successful
ci/woodpecker/pr/pipeline Pipeline was successful
Implement full comments system: domain entities (Comment, CommentLike), value objects (CommentContent), use cases (CRUD, like toggle), SQLAlchemy repository, API v1 endpoints, web UI with comment form and nested replies, i18n translations (EN/RU/FR/DE), and E2E tests. Fix nested reply (reply-to-reply) not displaying — the flat reply_comments dict was only queried for top-level comment IDs, so deeply nested replies were saved to DB (incrementing comment count) but never rendered. Switch to a recursive Jinja2 macro that renders any nesting depth.
771 lines
23 KiB
Python
771 lines
23 KiB
Python
"""Web UI routes for blog application with real use case integration.
|
|
|
|
This module provides HTML endpoints for the blog web interface
|
|
with role-based access control, user authentication, and full
|
|
integration with the application's use cases and domain layer.
|
|
"""
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from dishka.integrations.fastapi import DishkaRoute, FromDishka
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from markdown_it import MarkdownIt
|
|
from pygments import highlight
|
|
from pygments.formatters import HtmlFormatter
|
|
from pygments.lexers import get_lexer_by_name
|
|
from pygments.util import ClassNotFound
|
|
|
|
from app.application.dtos import CreatePostDTO, UpdatePostDTO
|
|
from app.application.use_cases import (
|
|
CreateCommentUseCase,
|
|
CreatePostUseCase,
|
|
DeletePostUseCase,
|
|
GetPostUseCase,
|
|
ListCommentsUseCase,
|
|
ListPostsUseCase,
|
|
PublishPostUseCase,
|
|
ToggleCommentLikeUseCase,
|
|
TogglePostLikeUseCase,
|
|
UpdatePostUseCase,
|
|
)
|
|
from app.domain.exceptions import (
|
|
AlreadyExistsException,
|
|
NotFoundException,
|
|
ValidationException,
|
|
)
|
|
from app.domain.repositories import CommentRepository
|
|
from app.domain.roles import Role, get_effective_role
|
|
from app.infrastructure.auth import TokenInfo
|
|
from app.infrastructure.config.settings import settings
|
|
from app.infrastructure.i18n.translator import DEFAULT_LOCALE, SUPPORTED_LOCALES, _
|
|
from app.presentation.web.deps import (
|
|
OptionalUserDep,
|
|
RequireUserDep,
|
|
can_create_post,
|
|
can_delete_post,
|
|
can_edit_post,
|
|
can_see_draft,
|
|
)
|
|
from app.presentation.web.flash import flash
|
|
|
|
router = APIRouter(prefix="/web", tags=["web"], route_class=DishkaRoute)
|
|
templates = Jinja2Templates(directory="app/presentation/templates")
|
|
|
|
|
|
def _jinja_translate(key: str, locale: str = DEFAULT_LOCALE) -> str:
|
|
"""Jinja2 global function for template translation.
|
|
|
|
Args:
|
|
key: Translation key to look up.
|
|
locale: Target locale code.
|
|
|
|
Returns:
|
|
Translated string or the key itself if no translation found.
|
|
"""
|
|
return _(key, locale)
|
|
|
|
|
|
templates.env.globals["_"] = _jinja_translate
|
|
|
|
|
|
_md = MarkdownIt("commonmark", {"html": False}).enable("table")
|
|
|
|
|
|
def _highlight_code(code: str, lang: str, _: Any) -> str:
|
|
try:
|
|
lexer = get_lexer_by_name(lang)
|
|
except ClassNotFound:
|
|
lexer = get_lexer_by_name("text")
|
|
formatter = HtmlFormatter(nowrap=True)
|
|
result: str = highlight(code, lexer, formatter)
|
|
return result
|
|
|
|
|
|
def markdown_filter(value: str) -> str:
|
|
md = MarkdownIt("commonmark", {"html": False, "highlight": _highlight_code}).enable("table")
|
|
return str(md.render(value))
|
|
|
|
|
|
templates.env.filters["markdown"] = markdown_filter
|
|
|
|
|
|
_DEFAULT_PAGE_SIZE = 10
|
|
|
|
|
|
def _get_user_role(user: TokenInfo | None) -> Role:
|
|
"""Get effective role from user token.
|
|
|
|
Args:
|
|
user: User token info or None for guest.
|
|
|
|
Returns:
|
|
Effective role for the user.
|
|
"""
|
|
if not user:
|
|
return Role.GUEST
|
|
return get_effective_role(user.roles)
|
|
|
|
|
|
def _get_base_context(
|
|
user: TokenInfo | None, current_locale: str = DEFAULT_LOCALE
|
|
) -> dict[str, Any]:
|
|
"""Get base template context with user info and permissions.
|
|
|
|
Args:
|
|
user: Current user or None for guest.
|
|
current_locale: Active locale code for i18n.
|
|
|
|
Returns:
|
|
Dictionary with user, user_role, can_create, and current_locale.
|
|
"""
|
|
user_role = _get_user_role(user)
|
|
|
|
return {
|
|
"user": user,
|
|
"user_role": user_role.value if user_role else None,
|
|
"can_create": can_create_post(user),
|
|
"current_locale": current_locale,
|
|
}
|
|
|
|
|
|
async def _get_visible_posts(
|
|
list_use_case: ListPostsUseCase,
|
|
user: TokenInfo | None,
|
|
limit: int,
|
|
offset: int,
|
|
) -> tuple[list[Any], bool]:
|
|
"""Fetch posts visible to the user with pagination.
|
|
|
|
For guests: only published posts.
|
|
For users: published posts plus own drafts.
|
|
For admins: all posts.
|
|
|
|
Args:
|
|
list_use_case: Use case for listing posts.
|
|
user: Current user or None for guest.
|
|
limit: Maximum number of posts to return.
|
|
offset: Number of posts to skip.
|
|
|
|
Returns:
|
|
Tuple of (visible posts, has_next flag).
|
|
"""
|
|
user_role = _get_user_role(user)
|
|
|
|
if user_role == Role.ADMIN:
|
|
posts = await list_use_case.all_posts()
|
|
posts = sorted(posts, key=lambda p: p.created_at, reverse=True)
|
|
total = len(posts)
|
|
posts = posts[offset : offset + limit]
|
|
has_next = offset + limit < total
|
|
return posts, has_next
|
|
|
|
published = await list_use_case.published_posts(limit=limit + 1, offset=offset)
|
|
has_next = len(published) > limit
|
|
published = published[:limit]
|
|
|
|
if user_role == Role.USER and user is not None:
|
|
own = await list_use_case.by_author(user.user_id)
|
|
published_ids = {p.id for p in published}
|
|
own_drafts = [p for p in own if p.id not in published_ids and not p.published]
|
|
merged = list(published) + own_drafts
|
|
merged.sort(key=lambda p: p.created_at, reverse=True)
|
|
return merged[:limit], has_next
|
|
|
|
return published, has_next
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
async def home(
|
|
request: Request,
|
|
user: OptionalUserDep,
|
|
list_use_case: FromDishka[ListPostsUseCase],
|
|
comment_repo: FromDishka[CommentRepository],
|
|
) -> HTMLResponse:
|
|
"""Render the home page with list of posts.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
user: Current user from dependency.
|
|
list_use_case: Use case for listing posts.
|
|
comment_repo: Repository for fetching comment counts.
|
|
|
|
Returns:
|
|
HTMLResponse with rendered posts list template.
|
|
"""
|
|
from dataclasses import replace
|
|
|
|
page_str = request.query_params.get("page", "1")
|
|
page = max(1, int(page_str) if page_str.isdigit() else 1)
|
|
offset = (page - 1) * _DEFAULT_PAGE_SIZE
|
|
|
|
visible_posts, has_next = await _get_visible_posts(
|
|
list_use_case, user, _DEFAULT_PAGE_SIZE, offset
|
|
)
|
|
|
|
for i, post in enumerate(visible_posts):
|
|
count = await comment_repo.count_by_post(post.id)
|
|
visible_posts[i] = replace(post, comment_count=count)
|
|
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/index.html",
|
|
{
|
|
**context,
|
|
"posts": visible_posts,
|
|
"active_page": "home",
|
|
"current_page": page,
|
|
"has_prev": page > 1,
|
|
"has_next": has_next,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/posts", response_class=HTMLResponse)
|
|
async def list_posts(
|
|
request: Request,
|
|
user: OptionalUserDep,
|
|
list_use_case: FromDishka[ListPostsUseCase],
|
|
comment_repo: FromDishka[CommentRepository],
|
|
) -> HTMLResponse:
|
|
"""Render the posts listing page.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
user: Current user from dependency.
|
|
list_use_case: Use case for listing posts.
|
|
comment_repo: Repository for fetching comment counts.
|
|
|
|
Returns:
|
|
HTMLResponse with rendered posts list template.
|
|
"""
|
|
from dataclasses import replace
|
|
|
|
page_str = request.query_params.get("page", "1")
|
|
page = max(1, int(page_str) if page_str.isdigit() else 1)
|
|
offset = (page - 1) * _DEFAULT_PAGE_SIZE
|
|
|
|
visible_posts, has_next = await _get_visible_posts(
|
|
list_use_case, user, _DEFAULT_PAGE_SIZE, offset
|
|
)
|
|
|
|
for i, post in enumerate(visible_posts):
|
|
count = await comment_repo.count_by_post(post.id)
|
|
visible_posts[i] = replace(post, comment_count=count)
|
|
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/index.html",
|
|
{
|
|
**context,
|
|
"posts": visible_posts,
|
|
"active_page": "posts",
|
|
"current_page": page,
|
|
"has_prev": page > 1,
|
|
"has_next": has_next,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/posts/new", response_class=HTMLResponse)
|
|
async def new_post_form(
|
|
request: Request,
|
|
user: RequireUserDep,
|
|
) -> HTMLResponse:
|
|
"""Render the new post creation form.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
user: Current user (required).
|
|
|
|
Returns:
|
|
HTMLResponse with rendered post form template.
|
|
"""
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/post_form.html",
|
|
{
|
|
**context,
|
|
"is_edit": False,
|
|
"post": None,
|
|
"active_page": "posts",
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/posts/new")
|
|
async def create_post(
|
|
request: Request,
|
|
user: RequireUserDep,
|
|
create_use_case: FromDishka[CreatePostUseCase],
|
|
publish_use_case: FromDishka[PublishPostUseCase],
|
|
) -> RedirectResponse:
|
|
"""Handle new post creation form submission.
|
|
|
|
Args:
|
|
request: The HTTP request object containing form data.
|
|
user: Current user (required).
|
|
create_use_case: Use case for creating posts.
|
|
publish_use_case: Use case for publishing posts.
|
|
|
|
Returns:
|
|
RedirectResponse to the new post or form page.
|
|
"""
|
|
form = await request.form()
|
|
title = str(form.get("title", "")).strip()
|
|
content = str(form.get("content", "")).strip()
|
|
tags_str = str(form.get("tags", "")).strip()
|
|
action = str(form.get("action", "draft")).strip()
|
|
|
|
tags = [t.strip() for t in tags_str.split(",") if t.strip()]
|
|
|
|
try:
|
|
dto = CreatePostDTO(
|
|
title=title,
|
|
content=content,
|
|
author_id=user.user_id,
|
|
tags=tags,
|
|
)
|
|
result = await create_use_case.execute(dto)
|
|
|
|
user_role = _get_user_role(user)
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
if action == "publish":
|
|
await publish_use_case.publish(result.id, user.user_id, user_role)
|
|
flash(request, _("flash.post_published", locale), "success")
|
|
else:
|
|
flash(request, _("flash.post_saved_draft", locale), "success")
|
|
|
|
return RedirectResponse(url=f"/web/posts/{result.slug}", status_code=303)
|
|
except AlreadyExistsException as exc:
|
|
flash(request, str(exc), "error")
|
|
return RedirectResponse(url="/web/posts/new", status_code=303)
|
|
except ValidationException as exc:
|
|
flash(request, str(exc), "error")
|
|
return RedirectResponse(url="/web/posts/new", status_code=303)
|
|
|
|
|
|
@router.get("/posts/{post_slug}", response_class=HTMLResponse)
|
|
async def post_detail(
|
|
request: Request,
|
|
post_slug: str,
|
|
user: OptionalUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
list_comments_use_case: FromDishka[ListCommentsUseCase],
|
|
comment_repo: FromDishka[CommentRepository],
|
|
) -> HTMLResponse:
|
|
"""Render a single post detail page with comments.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
post_slug: The URL-friendly slug of the post to display.
|
|
user: Current user from dependency.
|
|
get_use_case: Use case for retrieving posts.
|
|
list_comments_use_case: Use case for listing comments.
|
|
comment_repo: Repository for fetching comment count.
|
|
|
|
Returns:
|
|
HTMLResponse with rendered post detail template.
|
|
|
|
Raises:
|
|
HTTPException: If post not found or not visible to user.
|
|
"""
|
|
from dataclasses import replace
|
|
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
if not post.published and not can_see_draft(user, post.author_id):
|
|
raise HTTPException(status_code=404, detail="Post not found")
|
|
|
|
comments = await list_comments_use_case.execute(post.id)
|
|
comment_count = await comment_repo.count_by_post(post.id)
|
|
post = replace(post, comment_count=comment_count)
|
|
|
|
children: dict[str, list[Any]] = {}
|
|
for c in comments:
|
|
pid = str(c.parent_id) if c.parent_id else ""
|
|
if pid not in children:
|
|
children[pid] = []
|
|
children[pid].append(c)
|
|
|
|
top_level = children.pop("", [])
|
|
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/post_detail.html",
|
|
{
|
|
**context,
|
|
"post": post,
|
|
"top_level_comments": top_level,
|
|
"reply_comments": children,
|
|
"active_page": "posts",
|
|
"can_edit": can_edit_post(user, post.author_id),
|
|
"can_delete": can_delete_post(user, post.author_id),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/posts/{post_slug}/comments")
|
|
async def create_comment_web(
|
|
request: Request,
|
|
post_slug: str,
|
|
user: OptionalUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
create_use_case: FromDishka[CreateCommentUseCase],
|
|
) -> dict[str, object]:
|
|
"""Create a comment on a post via web UI.
|
|
|
|
Args:
|
|
request: The HTTP request object with JSON body.
|
|
post_slug: The URL-friendly slug of the post.
|
|
user: Current user from cookie or None.
|
|
get_use_case: Use case for retrieving post.
|
|
create_use_case: Use case for creating comments.
|
|
|
|
Returns:
|
|
JSON dict with created comment data.
|
|
|
|
Raises:
|
|
HTTPException: If user not authenticated or post not found.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
body = await request.json()
|
|
content = str(body.get("content", "")).strip()
|
|
parent_id_str = body.get("parent_id")
|
|
|
|
parent_id: UUID | None = None
|
|
if parent_id_str:
|
|
parent_id = UUID(parent_id_str)
|
|
|
|
result = await create_use_case.execute(
|
|
post_id=post.id,
|
|
author_id=user.user_id,
|
|
content=content,
|
|
parent_id=parent_id,
|
|
)
|
|
|
|
return {
|
|
"id": str(result.id),
|
|
"post_id": str(result.post_id),
|
|
"author_id": result.author_id,
|
|
"content": result.content,
|
|
"parent_id": str(result.parent_id) if result.parent_id else None,
|
|
"like_count": result.like_count,
|
|
"created_at": result.created_at.isoformat() if result.created_at else None,
|
|
}
|
|
|
|
|
|
@router.post("/comments/{comment_id}/like")
|
|
async def toggle_comment_like_web(
|
|
comment_id: UUID,
|
|
user: OptionalUserDep,
|
|
toggle_use_case: FromDishka[ToggleCommentLikeUseCase],
|
|
) -> dict[str, object]:
|
|
"""Toggle like on a comment via web UI.
|
|
|
|
Args:
|
|
comment_id: UUID of the comment.
|
|
user: Current user from cookie or None.
|
|
toggle_use_case: Use case for toggling comment likes.
|
|
|
|
Returns:
|
|
JSON dict with updated like_count.
|
|
|
|
Raises:
|
|
HTTPException: If user not authenticated.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
|
|
result = await toggle_use_case.execute(comment_id, user.user_id)
|
|
return {"like_count": result.like_count, "id": str(result.id)}
|
|
|
|
|
|
@router.get("/posts/{post_slug}/edit", response_class=HTMLResponse)
|
|
async def edit_post_form(
|
|
request: Request,
|
|
post_slug: str,
|
|
user: RequireUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
) -> HTMLResponse:
|
|
"""Render the post edit form.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
post_slug: The URL-friendly slug of the post to edit.
|
|
user: Current user (required).
|
|
get_use_case: Use case for retrieving posts.
|
|
|
|
Returns:
|
|
HTMLResponse with rendered post form template.
|
|
|
|
Raises:
|
|
HTTPException: If post not found or user cannot edit it.
|
|
"""
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
if not can_edit_post(user, post.author_id):
|
|
raise HTTPException(status_code=403, detail="Not authorized to edit this post")
|
|
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/post_form.html",
|
|
{
|
|
**context,
|
|
"is_edit": True,
|
|
"post": post,
|
|
"active_page": "posts",
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/posts/{post_slug}/edit")
|
|
async def update_post(
|
|
request: Request,
|
|
post_slug: str,
|
|
user: RequireUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
update_use_case: FromDishka[UpdatePostUseCase],
|
|
publish_use_case: FromDishka[PublishPostUseCase],
|
|
) -> RedirectResponse:
|
|
"""Handle post update form submission.
|
|
|
|
Args:
|
|
request: The HTTP request object containing form data.
|
|
post_slug: The URL-friendly slug of the post to update.
|
|
user: Current user (required).
|
|
get_use_case: Use case for retrieving posts.
|
|
update_use_case: Use case for updating posts.
|
|
publish_use_case: Use case for publishing posts.
|
|
|
|
Returns:
|
|
RedirectResponse to the updated post or form page.
|
|
"""
|
|
form = await request.form()
|
|
title = str(form.get("title", "")).strip()
|
|
content = str(form.get("content", "")).strip()
|
|
tags_str = str(form.get("tags", "")).strip()
|
|
action = str(form.get("action", "draft")).strip()
|
|
|
|
tags = [t.strip() for t in tags_str.split(",") if t.strip()]
|
|
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
if not can_edit_post(user, post.author_id):
|
|
raise HTTPException(status_code=403, detail="Not authorized to edit this post")
|
|
|
|
try:
|
|
dto = UpdatePostDTO(
|
|
title=title if title else None,
|
|
content=content if content else None,
|
|
tags=tags if tags else None,
|
|
)
|
|
user_role = _get_user_role(user)
|
|
result = await update_use_case.execute(post.id, dto, user.user_id, user_role)
|
|
|
|
if action == "publish":
|
|
if not result.published:
|
|
await publish_use_case.publish(result.id, user.user_id, user_role)
|
|
else:
|
|
if result.published:
|
|
await publish_use_case.unpublish(result.id, user.user_id, user_role)
|
|
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
flash(request, _("flash.post_updated", locale), "success")
|
|
return RedirectResponse(url=f"/web/posts/{result.slug}", status_code=303)
|
|
except (AlreadyExistsException, ValidationException) as exc:
|
|
flash(request, str(exc), "error")
|
|
return RedirectResponse(url=f"/web/posts/{post_slug}/edit", status_code=303)
|
|
|
|
|
|
@router.post("/posts/{post_slug}/delete")
|
|
async def delete_post(
|
|
request: Request,
|
|
post_slug: str,
|
|
user: RequireUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
delete_use_case: FromDishka[DeletePostUseCase],
|
|
) -> RedirectResponse:
|
|
"""Handle post deletion.
|
|
|
|
Args:
|
|
request: The HTTP request object.
|
|
post_slug: The URL-friendly slug of the post to delete.
|
|
user: Current user (required).
|
|
get_use_case: Use case for retrieving posts.
|
|
delete_use_case: Use case for deleting posts.
|
|
|
|
Returns:
|
|
RedirectResponse redirecting to the home page.
|
|
"""
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
if not can_delete_post(user, post.author_id):
|
|
raise HTTPException(status_code=403, detail="Not authorized to delete this post")
|
|
|
|
try:
|
|
user_role = _get_user_role(user)
|
|
await delete_use_case.execute(post.id, user.user_id, user_role)
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
flash(request, _("flash.post_deleted", locale), "success")
|
|
except NotFoundException:
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
flash(request, _("flash.post_not_found", locale), "error")
|
|
|
|
return RedirectResponse(url="/web/", status_code=303)
|
|
|
|
|
|
@router.post("/posts/{post_slug}/like")
|
|
async def toggle_like_web(
|
|
post_slug: str,
|
|
user: OptionalUserDep,
|
|
get_use_case: FromDishka[GetPostUseCase],
|
|
toggle_use_case: FromDishka[TogglePostLikeUseCase],
|
|
) -> dict[str, object]:
|
|
"""Toggle like on a post via web UI.
|
|
|
|
Args:
|
|
post_slug: The URL-friendly slug of the post.
|
|
user: Current user from cookie or None.
|
|
get_use_case: Use case for retrieving posts.
|
|
toggle_use_case: Use case for toggling likes.
|
|
|
|
Returns:
|
|
JSON dict with updated like_count.
|
|
|
|
Raises:
|
|
HTTPException: If post not found or user not authenticated.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Authentication required")
|
|
|
|
try:
|
|
post = await get_use_case.by_slug(post_slug)
|
|
except NotFoundException:
|
|
raise HTTPException(status_code=404, detail="Post not found") from None
|
|
|
|
result = await toggle_use_case.execute(post.id, user.user_id)
|
|
return {"like_count": result.like_count}
|
|
|
|
|
|
@router.get("/profile", response_class=HTMLResponse)
|
|
async def profile(
|
|
request: Request,
|
|
user: RequireUserDep,
|
|
) -> HTMLResponse:
|
|
"""Render user profile page.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
user: Current user (required).
|
|
|
|
Returns:
|
|
HTMLResponse with rendered profile template.
|
|
"""
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/profile.html",
|
|
{
|
|
**context,
|
|
"active_page": "profile",
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/about", response_class=HTMLResponse)
|
|
async def about(
|
|
request: Request,
|
|
user: OptionalUserDep,
|
|
) -> HTMLResponse:
|
|
"""Render the about page.
|
|
|
|
Args:
|
|
request: The HTTP request object for template context.
|
|
user: Current user from dependency.
|
|
|
|
Returns:
|
|
HTMLResponse with rendered about page template.
|
|
"""
|
|
locale = getattr(request.state, "locale", DEFAULT_LOCALE)
|
|
context = _get_base_context(user, locale)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"pages/about.html",
|
|
{
|
|
**context,
|
|
"active_page": "about",
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/lang/{locale}")
|
|
async def set_language(
|
|
request: Request,
|
|
locale: str,
|
|
) -> RedirectResponse:
|
|
"""Set the active language and redirect back to the previous page.
|
|
|
|
Stores the locale choice in a persistent cookie so that subsequent
|
|
requests use the selected language. Falls back to browser preference
|
|
or English default.
|
|
|
|
Args:
|
|
request: HTTP request object.
|
|
locale: Target locale code (en, ru, fr, de).
|
|
|
|
Returns:
|
|
RedirectResponse back to the referrer or home page.
|
|
"""
|
|
if locale not in SUPPORTED_LOCALES:
|
|
locale = DEFAULT_LOCALE
|
|
|
|
referer = request.headers.get("referer", "/web/")
|
|
response = RedirectResponse(url=referer, status_code=303)
|
|
response.set_cookie(
|
|
key="locale",
|
|
value=locale,
|
|
httponly=True,
|
|
secure=not settings.is_dev,
|
|
samesite="lax",
|
|
max_age=365 * 24 * 3600,
|
|
)
|
|
return response
|