from __future__ import annotations from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.calendars import Calendar from models.ghost_content import Post # for FK existence checks import unicodedata import re class CalendarError(ValueError): """Base error for calendar service operations.""" from suma_browser.app.utils import ( utcnow ) def slugify(value: str, max_len: int = 255) -> str: """ Make a URL-friendly slug: - lowercase - remove accents - replace any non [a-z0-9]+ with '-' - no forward slashes - collapse multiple dashes - trim leading/trailing dashes """ if value is None: value = "" # normalize accents -> ASCII value = unicodedata.normalize("NFKD", value) value = value.encode("ascii", "ignore").decode("ascii") value = value.lower() # explicitly block forward slashes value = value.replace("/", "-") # replace non-alnum with hyphen value = re.sub(r"[^a-z0-9]+", "-", value) # collapse multiple hyphens value = re.sub(r"-{2,}", "-", value) # trim hyphens and enforce length value = value.strip("-")[:max_len].strip("-") # fallback if empty return value or "calendar" async def soft_delete(sess: AsyncSession, post_slug: str, calendar_slug: str) -> bool: cal = ( await sess.execute( select(Calendar) .join(Post, Calendar.post_id == Post.id) .where( Post.slug == post_slug, Calendar.slug == calendar_slug, Calendar.deleted_at.is_(None), ) ) ).scalar_one_or_none() if not cal: return False cal.deleted_at = utcnow() await sess.flush() return True async def create_calendar(sess: AsyncSession, post_id: int, name: str) -> Calendar: """ Create a calendar for a post. Name must be unique per post. If a calendar with the same (post_id, name) exists but is soft-deleted, it will be revived (deleted_at=None). """ name = (name or "").strip() if not name: raise CalendarError("Calendar name must not be empty.") slug=slugify(name) # Ensure post exists (avoid silent FK errors in some DBs) post = (await sess.execute(select(Post).where(Post.id == post_id))).scalar_one_or_none() if not post: raise CalendarError(f"Post {post_id} does not exist.") # Enforce: calendars can only be created on pages with the calendar feature if not post.is_page: raise CalendarError("Calendars can only be created on pages, not posts.") # Look for existing (including soft-deleted) q = await sess.execute( select(Calendar).where(Calendar.post_id == post_id, Calendar.name == name) ) existing = q.scalar_one_or_none() if existing: if existing.deleted_at is not None: existing.deleted_at = None # revive await sess.flush() return existing raise CalendarError(f'Calendar with slug "{slug}" already exists for post {post_id}.') cal = Calendar(post_id=post_id, name=name, slug=slug) sess.add(cal) await sess.flush() return cal