|
import logging
|
|
import re
|
|
from typing import Optional, List
|
|
from snek.system.service import BaseService
|
|
from snek.system.exceptions import ValidationError, DuplicateResourceError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ProfilePageService(BaseService):
|
|
mapper_name = "profile_page"
|
|
|
|
def slugify(self, title: str) -> str:
|
|
slug = title.lower().strip()
|
|
slug = re.sub(r'[^\w\s-]', '', slug)
|
|
slug = re.sub(r'[-\s]+', '-', slug)
|
|
return slug[:100]
|
|
|
|
async def create_page(self, user_uid: str, title: str, content: str = "", is_published: bool = True) -> dict:
|
|
slug = self.slugify(title)
|
|
|
|
existing = await self.get(user_uid=user_uid, slug=slug, deleted_at=None)
|
|
if existing:
|
|
raise DuplicateResourceError(f"A page with slug '{slug}' already exists")
|
|
|
|
pages = [p async for p in self.find(user_uid=user_uid, deleted_at=None)]
|
|
max_order = max([p["order_index"] for p in pages], default=-1)
|
|
|
|
model = await self.new()
|
|
model["user_uid"] = user_uid
|
|
model["title"] = title
|
|
model["slug"] = slug
|
|
model["content"] = content
|
|
model["order_index"] = max_order + 1
|
|
model["is_published"] = is_published
|
|
|
|
await self.save(model)
|
|
return model
|
|
|
|
async def update_page(self, page_uid: str, title: Optional[str] = None,
|
|
content: Optional[str] = None, is_published: Optional[bool] = None) -> dict:
|
|
page = await self.get(uid=page_uid, deleted_at=None)
|
|
if not page:
|
|
raise ValidationError("Page not found")
|
|
|
|
if title is not None:
|
|
page["title"] = title
|
|
new_slug = self.slugify(title)
|
|
existing = await self.get(user_uid=page["user_uid"], slug=new_slug, deleted_at=None)
|
|
if existing and existing["uid"] != page_uid:
|
|
raise DuplicateResourceError(f"A page with slug '{new_slug}' already exists")
|
|
page["slug"] = new_slug
|
|
|
|
if content is not None:
|
|
page["content"] = content
|
|
|
|
if is_published is not None:
|
|
page["is_published"] = is_published
|
|
|
|
return await self.save(page)
|
|
|
|
async def get_user_pages(self, user_uid: str, include_unpublished: bool = False) -> List[dict]:
|
|
if include_unpublished:
|
|
pages = [p.record async for p in self.find(user_uid=user_uid, deleted_at=None)]
|
|
else:
|
|
pages = [p.record async for p in self.find(user_uid=user_uid, is_published=True, deleted_at=None)]
|
|
|
|
return sorted(pages, key=lambda p: p["order_index"])
|
|
|
|
async def get_page_by_slug(self, user_uid: str, slug: str, include_unpublished: bool = False) -> Optional[dict]:
|
|
page = await self.get(user_uid=user_uid, slug=slug, deleted_at=None)
|
|
if page and (include_unpublished or page["is_published"]):
|
|
return page
|
|
return None
|
|
|
|
async def reorder_pages(self, user_uid: str, page_uids: List[str]) -> None:
|
|
for index, page_uid in enumerate(page_uids):
|
|
page = await self.get(uid=page_uid, user_uid=user_uid, deleted_at=None)
|
|
if page:
|
|
page["order_index"] = index
|
|
await self.save(page)
|
|
|
|
async def delete_page(self, page_uid: str) -> None:
|
|
page = await self.get(uid=page_uid, deleted_at=None)
|
|
if page:
|
|
await self.delete(page)
|