from __future__ import annotations from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.calendars import Calendar from shared.services.registry import services from shared.services.relationships import attach_child, detach_child import unicodedata import re class CalendarError(ValueError): """Base error for calendar service operations.""" from shared.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: post = await services.blog.get_post_by_slug(sess, post_slug) if not post: return False cal = ( await sess.execute( select(Calendar).where( Calendar.container_type == "page", Calendar.container_id == post.id, 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() await detach_child(sess, "page", cal.container_id, "calendar", cal.id) 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 services.blog.get_post_by_id(sess, post_id) 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.container_type == "page", Calendar.container_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() await attach_child(sess, "page", post_id, "calendar", existing.id) return existing raise CalendarError(f'Calendar with slug "{slug}" already exists for post {post_id}.') cal = Calendar(container_type="page", container_id=post_id, name=name, slug=slug) sess.add(cal) await sess.flush() await attach_child(sess, "page", post_id, "calendar", cal.id) return cal