import copy import json import logging import os import re import uuid from typing import Any from jsonschema import Draft202012Validator from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import select from constants.presentation import MAX_NUMBER_OF_SLIDES from models.image_prompt import ImagePrompt from models.presentation_outline_model import PresentationOutlineModel, SlideOutlineModel from models.sql.image_asset import ImageAsset from models.sql.key_value import KeyValueSqlModel from models.sql.presentation import PresentationModel from models.sql.slide import SlideModel from models.sql.template_v2 import TemplateV2 from services.icon_finder_service import ICON_FINDER_SERVICE from services.documents_loader import DocumentsLoader from services.image_generation_service import ImageGenerationService from services.mem0_presentation_memory_service import MEM0_PRESENTATION_MEMORY_SERVICE from services.temp_file_service import TEMP_FILE_SERVICE from templates.presentation_layout import PresentationLayoutModel, SlideLayoutModel from templates.v2.schema import get_template_schema from utils.asset_directory_utils import ( filesystem_image_path_to_app_data_url, get_images_directory, normalize_slide_asset_url, ) from utils.icon_weights import DEFAULT_ICON_WEIGHT from utils.outline_utils import get_presentation_title_from_presentation_outline from utils.outline_limits import normalize_outline_content from utils.process_slides import ( process_old_and_new_slides_and_fetch_assets, process_slide_and_fetch_assets, ) LOGGER = logging.getLogger(__name__) MAX_SCHEMA_ERRORS = 10 DEFAULT_SOURCE_DOCUMENT_CHARS = 12000 MAX_SOURCE_DOCUMENT_CHARS = 30000 SLIDE_STAGE_WIDTH = 1280.0 SLIDE_STAGE_HEIGHT = 720.0 BLANK_SLIDE_LAYOUT_ID = "__blank_slide__" BLANK_TEMPLATE_V2_LAYOUT: dict[str, Any] = { "id": BLANK_SLIDE_LAYOUT_ID, "description": "Empty slide.", "background": "#FFFFFF", "components": [], "elements": [ { "type": "rectangle", "position": {"x": 0, "y": 0}, "size": {"width": 1280, "height": 720}, "fill": {"color": "#FFFFFF"}, "decorative": True, } ], } TEMPLATE_V2_GENERATED_ELEMENT_TYPES = {"text", "image", "text-list", "table", "chart"} # Keep URL runtime fields during validation because many slide schemas require them. # Speaker note is handled separately and should not affect JSON-schema checks. RUNTIME_CONTENT_FIELDS = {"__speaker_note__"} DEFAULT_INSERT_BOXES = { "chart": { "position": {"x": 128.0, "y": 108.0}, "size": {"width": 1024.0, "height": 460.0}, "min_size": {"width": 320.0, "height": 180.0}, }, "table": { "position": {"x": 128.0, "y": 120.0}, "size": {"width": 1024.0, "height": 410.0}, "min_size": {"width": 420.0, "height": 160.0}, }, } THEMES_STORAGE_KEY = "presentation_custom_themes" CHAT_BUILTIN_THEMES: list[dict[str, Any]] = [ { "id": "edge-yellow", "name": "Edge Yellow", "description": "Yellow and dark theme for professionalish and edge.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#f5f547", "background": "#1f1f1f", "card": "#424242", "stroke": "#585858", "primary_text": "#161616", "background_text": "#f5f547", "graph_0": "#ffff54", "graph_1": "#f1f142", "graph_2": "#dada15", "graph_3": "#c1bf00", "graph_4": "#a8a600", "graph_5": "#908c00", "graph_6": "#797400", "graph_7": "#625c00", "graph_8": "#4d4500", "graph_9": "#382f00", }, "fonts": { "textFont": { "name": "Playfair Display", "url": "https://fonts.googleapis.com/css2?family=Playfair+Display:wght@400..900&display=swap", } }, }, }, { "id": "light-rose", "name": "Light Rose", "description": "Rose background with punchy font.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#030204", "background": "#f69c9c", "card": "#ffaeb4", "stroke": "#bf6a6b", "primary_text": "#bebebe", "background_text": "#030202", "graph_0": "#2f2c32", "graph_1": "#444147", "graph_2": "#5a565d", "graph_3": "#706d73", "graph_4": "#88848b", "graph_5": "#a09da4", "graph_6": "#b9b6bd", "graph_7": "#d3cfd6", "graph_8": "#eae6ed", "graph_9": "#f7f3fb", }, "fonts": { "textFont": { "name": "Overpass", "url": "https://fonts.googleapis.com/css2?family=Overpass:wght@100..900&display=swap", } }, }, }, { "id": "mint-blue", "name": "Mint Blue", "description": "Mint green with blue heading.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#3b3172", "background": "#ffffff", "card": "#80e7cf", "stroke": "#d1d1d1", "primary_text": "#ffffff", "background_text": "#3b3172", "graph_0": "#003d2d", "graph_1": "#005341", "graph_2": "#006a57", "graph_3": "#00826d", "graph_4": "#2b9a85", "graph_5": "#4ab39d", "graph_6": "#65cdb6", "graph_7": "#80e7cf", "graph_8": "#98ffe6", "graph_9": "#a5fff4", }, "fonts": { "textFont": { "name": "Prompt", "url": "https://fonts.googleapis.com/css2?family=Prompt:wght@100..900&display=swap", } }, }, }, { "id": "professional-blue", "name": "Professional Blue", "description": "Clean and professional blue theme.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#161616", "background": "#ffffff", "card": "#dae6ff", "stroke": "#d1d1d1", "primary_text": "#eeeaea", "background_text": "#000000", "graph_0": "#2e2e2e", "graph_1": "#424242", "graph_2": "#585858", "graph_3": "#6f6f6f", "graph_4": "#868686", "graph_5": "#9e9e9e", "graph_6": "#b7b7b7", "graph_7": "#d1d1d1", "graph_8": "#e8e8e8", "graph_9": "#f5f5f5", }, "fonts": { "textFont": { "name": "Inter", "url": "https://fonts.googleapis.com/css2?family=Inter:wght@100..900&display=swap", } }, }, }, { "id": "professional-dark", "name": "Professional Dark", "description": "Clean and professional for dark corporate usage.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#eff5f1", "background": "#050505", "card": "#424242", "stroke": "#585858", "primary_text": "#050505", "background_text": "#eff5f1", "graph_0": "#ebf6ff", "graph_1": "#dee8fa", "graph_2": "#c7d2e3", "graph_3": "#aeb8c9", "graph_4": "#959fb0", "graph_5": "#7d8797", "graph_6": "#666f7f", "graph_7": "#505867", "graph_8": "#3a4351", "graph_9": "#262e3c", }, "fonts": { "textFont": { "name": "Instrument Sans", "url": "https://fonts.googleapis.com/css2?family=Instrument+Sans:ital,wght@0,400..700;1,400..700&display=swap", } }, }, }, ] THEME_COLOR_KEYS = [ "primary", "background", "card", "stroke", "primary_text", "background_text", "graph_0", "graph_1", "graph_2", "graph_3", "graph_4", "graph_5", "graph_6", "graph_7", "graph_8", "graph_9", ] DEFAULT_THEME_FONT = { "name": "Inter", "url": "https://fonts.googleapis.com/css2?family=Inter:wght@100..900&display=swap", } class PresentationChatMemoryLayer: """ Memory abstraction for chat tools and context retrieval. This layer intentionally hides where data comes from (SQL-backed persisted state and mem0 retrieval) behind `get` and `search`-style methods so chat logic stays decoupled from storage details. """ def __init__(self, sql_session: AsyncSession, presentation_id: uuid.UUID): self._sql_session = sql_session self._presentation_id = presentation_id async def get(self, key: str) -> Any: if key != "presentation_outline": return None # Prefer live slides from SQL so slide count and slide indices are always current. slides_result = await self._sql_session.scalars( select(SlideModel) .where(SlideModel.presentation == self._presentation_id) .order_by(SlideModel.index) ) slides = list(slides_result) if slides: LOGGER.info( "Chat outline loaded from slides table (presentation_id=%s, slides=%d)", self._presentation_id, len(slides), ) return { "source": "slides_table", "slide_count": len(slides), "slides": [ { "slide_id": str(slide.id), "index": slide.index, "layout_id": slide.layout, "content": slide.content, "speaker_note": slide.speaker_note, } for slide in slides ], } presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation or not presentation.outlines: LOGGER.info( "Chat memory miss for outline (presentation_id=%s)", self._presentation_id, ) return None LOGGER.info( "Chat outline fallback hit from presentation.outlines (presentation_id=%s)", self._presentation_id, ) return presentation.outlines async def search(self, query: str, limit: int = 5) -> list[dict[str, Any]]: """ Search slides directly from SQL-backed slide rows. Results are intentionally compact (snippet-first) to keep tool-call payloads small for models with limited context windows. """ trimmed_query = (query or "").strip() if not trimmed_query: return [] slides_result = await self._sql_session.scalars( select(SlideModel).where(SlideModel.presentation == self._presentation_id) ) slides = sorted(list(slides_result), key=lambda slide: slide.index) if not slides: LOGGER.info( "Chat memory miss for slide search (presentation_id=%s, reason=no_slides)", self._presentation_id, ) return [] query_lower = trimmed_query.lower() query_tokens = set(re.findall(r"[a-z0-9]{2,}", query_lower)) ranked: list[tuple[int, dict[str, Any]]] = [] for slide in slides: serialized = self._serialize_slide(slide) searchable = serialized.lower() score = 0 if query_lower in searchable: score += 8 if query_tokens: score += sum(1 for token in query_tokens if token in searchable) if score <= 0: continue ranked.append( ( score, { "slide_id": str(slide.id), "index": slide.index, "slide_number": slide.index + 1, "layout_id": slide.layout, "snippet": self._build_snippet(serialized, query_lower), "score": score, }, ) ) ranked.sort(key=lambda item: (-item[0], item[1]["index"])) results = [entry for _, entry in ranked[: max(1, limit)]] LOGGER.info( "Chat DB slide search completed (presentation_id=%s, query=%r, hits=%d)", self._presentation_id, trimmed_query, len(results), ) return results async def get_slide_at_index( self, index: int, *, include_full_content: bool = False ) -> dict[str, Any] | None: slide = await self._sql_session.scalar( select(SlideModel).where( SlideModel.presentation == self._presentation_id, SlideModel.index == index, ) ) if not slide: LOGGER.info( "Chat memory miss for slide by index (presentation_id=%s, index=%d)", self._presentation_id, index, ) return None response: dict[str, Any] = { "slide_id": str(slide.id), "index": slide.index, "slide_number": slide.index + 1, "layout_id": slide.layout, "content_preview": self._build_snippet( self._serialize_slide(slide), query_lower="", window=420, ), "speaker_note": slide.speaker_note, } if include_full_content: response["content"] = slide.content response["ui"] = slide.ui ui = self._slide_ui_layout(slide) if ui is not None: response["ui_summary"] = await self.get_slide_ui_elements( index=slide.index, include_full_json=False, ) return response async def get_outline_draft(self) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "found": False, "message": "Presentation not found.", "slide_count": 0, "slides": [], } slides = self._normalize_outline_slides(presentation.outlines) if not slides: return { "found": False, "message": "No outline draft is available yet.", "slide_count": 0, "slides": [], } return { "found": True, "message": "Outline draft fetched successfully.", "slide_count": len(slides), "slides": [ { "index": index, "slide_number": index + 1, "title": self._extract_outline_title(slide["content"]), "content": slide["content"], } for index, slide in enumerate(slides) ], } async def add_outline( self, *, content: str, index: int | None = None, ) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "saved": False, "message": "Presentation not found.", } slides = self._normalize_outline_slides(presentation.outlines) if len(slides) >= MAX_NUMBER_OF_SLIDES: return { "saved": False, "message": f"Outline slide limit reached. You can have at most {MAX_NUMBER_OF_SLIDES} outlines.", "slide_count": len(slides), "max_slide_count": MAX_NUMBER_OF_SLIDES, } insert_index = len(slides) if index is None else min(max(0, index), len(slides)) slides.insert(insert_index, {"content": normalize_outline_content(content.strip())}) await self._save_outline_slides(presentation, slides) return { "saved": True, "action": "created", "message": f"Outline slide added at index {insert_index}.", "index": insert_index, "slide_count": len(slides), } async def update_outline(self, *, index: int, content: str) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "saved": False, "message": "Presentation not found.", } slides = self._normalize_outline_slides(presentation.outlines) target_index = max(0, index) if target_index >= len(slides): return { "saved": False, "message": f"No outline slide found at index {target_index}.", "index": target_index, "slide_count": len(slides), } slides[target_index] = {"content": normalize_outline_content(content.strip())} await self._save_outline_slides(presentation, slides) return { "saved": True, "action": "updated", "message": f"Outline slide at index {target_index} was updated.", "index": target_index, "slide_count": len(slides), } async def delete_outline(self, *, index: int) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "deleted": False, "message": "Presentation not found.", } slides = self._normalize_outline_slides(presentation.outlines) target_index = max(0, index) if target_index >= len(slides): return { "deleted": False, "message": f"No outline slide found at index {target_index}.", "index": target_index, "slide_count": len(slides), } slides.pop(target_index) await self._save_outline_slides(presentation, slides) return { "deleted": True, "action": "deleted", "message": f"Outline slide at index {target_index} was deleted.", "index": target_index, "slide_count": len(slides), } async def move_outline(self, *, from_index: int, to_index: int) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "saved": False, "message": "Presentation not found.", } slides = self._normalize_outline_slides(presentation.outlines) source_index = max(0, from_index) if source_index >= len(slides): return { "saved": False, "message": f"No outline slide found at index {source_index}.", "from_index": source_index, "slide_count": len(slides), } destination_index = min(max(0, to_index), len(slides) - 1) [slide] = slides[source_index : source_index + 1] del slides[source_index] slides.insert(destination_index, slide) await self._save_outline_slides(presentation, slides) return { "saved": True, "action": "moved", "message": ( f"Outline slide moved from index {source_index} " f"to index {destination_index}." ), "from_index": source_index, "to_index": destination_index, "slide_count": len(slides), } async def get_available_layouts(self) -> list[dict[str, Any]]: presentation = await self._sql_session.get( PresentationModel, self._presentation_id ) layout_model = await self._get_layout_model(presentation) if not layout_model: return [] return [ { "id": layout.id, "name": layout.name, "description": layout.description, } for layout in layout_model.slides ] async def get_available_blocks( self, *, query: str | None = None, layout_id: str | None = None, element_type: str | None = None, block_id: str | None = None, include_full_content: bool = False, max_results: int = 20, ) -> dict[str, Any]: presentation = await self._sql_session.get( PresentationModel, self._presentation_id ) if not presentation: return { "found": False, "count": 0, "blocks": [], "message": "Presentation not found.", } candidates = await self._collect_template_block_candidates(presentation) query_text = (query or "").strip().lower() layout_filter = (layout_id or "").strip() element_filter = (element_type or "").strip().lower() block_filter = (block_id or "").strip() matches: list[tuple[int, int, dict[str, Any]]] = [] for source_index, candidate in enumerate(candidates): if block_filter and candidate.get("block_id") != block_filter: continue if layout_filter and candidate.get("layout_id") != layout_filter: continue element_types = { str(item).lower() for item in candidate.get("element_types", []) if item is not None } if element_filter and element_filter not in element_types: continue score = self._block_match_score(candidate, query_text) if query_text and score <= 0: continue matches.append((score, source_index, candidate)) matches.sort( key=lambda item: ( -item[0], bool(item[2].get("decorative")), item[1], ) ) limit = min(max(max_results, 1), 50) blocks = [ self._format_available_block( candidate, include_full_content=include_full_content, ) for _, _, candidate in matches[:limit] ] return { "found": bool(blocks), "count": len(blocks), "total_matches": len(matches), "blocks": blocks, "truncated": len(matches) > len(blocks), "message": ( f"Found {len(blocks)} matching block(s)." if blocks else "No matching blocks were found." ), } async def read_source_documents( self, *, query: str | None = None, max_chars: int | None = None, ) -> dict[str, Any]: presentation = await self._sql_session.get( PresentationModel, self._presentation_id ) if not presentation: return { "found": False, "message": "Presentation not found.", "documents": [], } char_budget = min( max(max_chars or DEFAULT_SOURCE_DOCUMENT_CHARS, 1000), MAX_SOURCE_DOCUMENT_CHARS, ) source_paths = [ path for path in (presentation.file_paths or []) if isinstance(path, str) and path.strip() ] documents: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] remaining_chars = char_budget for source_index, raw_path in enumerate(source_paths): if remaining_chars <= 0: break name = os.path.basename(raw_path) or f"Document {source_index + 1}" try: resolved_path = TEMP_FILE_SERVICE.resolve_temp_path( raw_path, must_exist=True, ) loader = DocumentsLoader( file_paths=[resolved_path], presentation_language=presentation.language, ) temp_dir = TEMP_FILE_SERVICE.create_temp_dir(str(uuid.uuid4())) await loader.load_documents(temp_dir=temp_dir) parsed_text = loader.documents[0] if loader.documents else "" except Exception as exc: errors.append({"name": name, "error": str(exc)}) continue trimmed = self._trim_document_text(parsed_text, remaining_chars) if not trimmed: errors.append({"name": name, "error": "No text was extracted."}) continue documents.append( { "index": source_index, "name": name, "content": trimmed, "truncated": len(parsed_text.strip()) > len(trimmed), } ) remaining_chars -= len(trimmed) if documents: omitted_count = max(0, len(source_paths) - len(documents) - len(errors)) return { "found": True, "source": "uploaded_files", "count": len(documents), "documents": documents, "errors": errors, "omitted_count": omitted_count, "message": f"Read {len(documents)} source document(s).", } fallback_query = ( (query or "").strip() or "uploaded source document extracted PDF document text summary" ) fallback_context = await MEM0_PRESENTATION_MEMORY_SERVICE.retrieve_context( self._presentation_id, fallback_query, ) if fallback_context.strip(): return { "found": True, "source": "presentation_memory", "count": 1, "documents": [ { "index": 0, "name": "Indexed source document context", "content": self._trim_document_text( fallback_context, char_budget, ), "truncated": len(fallback_context.strip()) > char_budget, } ], "errors": errors, "message": ( "Source upload files were unavailable, so indexed document " "memory was returned instead." ), } if source_paths: return { "found": False, "source": "uploaded_files", "count": 0, "documents": [], "errors": errors, "message": ( "Source document files are recorded for this presentation, " "but no readable text could be extracted." ), } return { "found": False, "source": "presentation", "count": 0, "documents": [], "errors": errors, "message": "No uploaded source documents are linked to this presentation.", } async def get_content_schema_from_layout_id( self, layout_id: str ) -> dict[str, Any] | None: layout = await self._get_layout_by_id(layout_id) if not layout: return None return layout.json_schema async def _get_presentation_icon_weight( self, presentation: PresentationModel | None = None ) -> str: if presentation is None: presentation = await self._sql_session.get( PresentationModel, self._presentation_id ) layout_model = await self._get_layout_model(presentation) return layout_model.icon_weight if layout_model else DEFAULT_ICON_WEIGHT async def generate_image(self, prompt: str) -> str: image_generation_service = ImageGenerationService(get_images_directory()) image = await image_generation_service.generate_image(ImagePrompt(prompt=prompt)) if isinstance(image, ImageAsset): self._sql_session.add(image) await self._sql_session.commit() return filesystem_image_path_to_app_data_url(image.path) return normalize_slide_asset_url(str(image)) async def generate_icon(self, query: str) -> str: icons = await ICON_FINDER_SERVICE.search_icons( query, k=1, weight=await self._get_presentation_icon_weight(), ) if icons: return normalize_slide_asset_url(icons[0]) return normalize_slide_asset_url("/static/icons/placeholder.svg") async def add_blank_slide(self, *, index: int | None = None) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return {"added": False, "message": "Presentation not found."} slides_result = await self._sql_session.scalars( select(SlideModel) .where(SlideModel.presentation == self._presentation_id) .order_by(SlideModel.index) ) slides = list(slides_result) if len(slides) >= MAX_NUMBER_OF_SLIDES: return { "added": False, "message": f"Slide limit reached. You can have at most {MAX_NUMBER_OF_SLIDES} slides.", "slide_count": len(slides), "max_slide_count": MAX_NUMBER_OF_SLIDES, } insert_index = ( len(slides) if index is None else min(max(0, index), len(slides)) ) for slide in sorted( [slide for slide in slides if slide.index >= insert_index], key=lambda each: each.index, reverse=True, ): slide.index += 1 self._sql_session.add(slide) presentation.n_slides = len(slides) + 1 self._sql_session.add(presentation) new_slide = SlideModel( presentation=self._presentation_id, layout_group=self._resolve_layout_group(presentation=presentation), layout=BLANK_SLIDE_LAYOUT_ID, index=insert_index, content={}, speaker_note="", ui=self._blank_slide_ui(), ) self._sql_session.add(new_slide) await self._sql_session.commit() await self._sql_session.refresh(new_slide) return { "added": True, "message": f"Blank slide added at index {insert_index}.", "slide_id": str(new_slide.id), "index": insert_index, "slide_number": insert_index + 1, } async def save_slide( self, *, content: dict[str, Any], layout_id: str, index: int, replace_old_slide_at_index: bool, ) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "saved": False, "message": "Presentation not found.", "validation_errors": [], } layout = await self._get_layout_by_id(layout_id, presentation=presentation) if not layout: return { "saved": False, "message": f"Layout '{layout_id}' was not found in this presentation.", "validation_errors": [f"Unknown layout_id '{layout_id}'."], } icon_weight = await self._get_presentation_icon_weight(presentation) layout_group = self._resolve_layout_group(presentation=presentation) validation_errors = self._validate_slide_content( content=content, schema=layout.json_schema, ) if validation_errors: return { "saved": False, "message": "Slide content failed schema validation.", "validation_errors": validation_errors, } target_index = max(0, index) image_generation_service = ImageGenerationService(get_images_directory()) if replace_old_slide_at_index: existing_slide = await self._sql_session.scalar( select(SlideModel).where( SlideModel.presentation == self._presentation_id, SlideModel.index == target_index, ) ) if not existing_slide: return { "saved": False, "message": f"No existing slide found at index {target_index} to replace.", "validation_errors": [], } updated_content = copy.deepcopy(content) image_warnings: list[dict] = [] new_assets = await process_old_and_new_slides_and_fetch_assets( image_generation_service=image_generation_service, old_slide_content=existing_slide.content or {}, new_slide_content=updated_content, icon_weight=icon_weight, use_template_v2_asset_fields=existing_slide.layout_group.startswith( "template-v2" ), allow_image_fallback=True, image_warnings=image_warnings, ) for warning in image_warnings: LOGGER.warning( "Chat slide replacement image generation warning: " "presentation_id=%s slide_index=%s detail=%s", self._presentation_id, target_index, warning.get("detail"), ) existing_slide.id = uuid.uuid4() existing_slide.layout = layout_id existing_slide.layout_group = layout_group existing_slide.content = updated_content existing_slide.ui = await self._build_template_v2_slide_ui( presentation=presentation, layout_id=layout_id, content=updated_content, ) existing_slide.speaker_note = self._extract_speaker_note(updated_content) self._sql_session.add(existing_slide) self._sql_session.add_all(new_assets) await self._sql_session.commit() await MEM0_PRESENTATION_MEMORY_SERVICE.store_slide_edit( presentation_id=self._presentation_id, slide_index=target_index, edit_prompt=f"[chat_tool_save_slide_replace] layout_id={layout_id}", edited_slide_content=updated_content, ) return { "saved": True, "action": "replaced", "message": f"Slide at index {target_index} was replaced successfully.", "slide_id": str(existing_slide.id), "index": target_index, } slides_result = await self._sql_session.scalars( select(SlideModel) .where(SlideModel.presentation == self._presentation_id) .order_by(SlideModel.index) ) slides = list(slides_result) if len(slides) >= MAX_NUMBER_OF_SLIDES: return { "saved": False, "message": f"Slide limit reached. You can have at most {MAX_NUMBER_OF_SLIDES} slides.", "validation_errors": [], "slide_count": len(slides), "max_slide_count": MAX_NUMBER_OF_SLIDES, } if slides: max_index = max(slide.index for slide in slides) insert_index = min(target_index, max_index + 1) slides_to_shift = [slide for slide in slides if slide.index >= insert_index] else: insert_index = 0 slides_to_shift = [] for slide in sorted(slides_to_shift, key=lambda each: each.index, reverse=True): slide.index += 1 self._sql_session.add(slide) new_slide_content = copy.deepcopy(content) new_slide = SlideModel( presentation=self._presentation_id, layout_group=layout_group, layout=layout_id, index=insert_index, content=new_slide_content, speaker_note=self._extract_speaker_note(new_slide_content), ) image_warnings: list[dict] = [] new_assets = await process_slide_and_fetch_assets( image_generation_service=image_generation_service, slide=new_slide, icon_weight=icon_weight, allow_image_fallback=True, image_warnings=image_warnings, ) for warning in image_warnings: LOGGER.warning( "Chat slide image generation warning: presentation_id=%s detail=%s", self._presentation_id, warning.get("detail"), ) new_slide.ui = await self._build_template_v2_slide_ui( presentation=presentation, layout_id=layout_id, content=new_slide.content, ) self._sql_session.add(new_slide) self._sql_session.add_all(new_assets) await self._sql_session.commit() await self._sql_session.refresh(new_slide) await MEM0_PRESENTATION_MEMORY_SERVICE.store_slide_edit( presentation_id=self._presentation_id, slide_index=insert_index, edit_prompt=f"[chat_tool_save_slide_new] layout_id={layout_id}", edited_slide_content=new_slide.content, ) return { "saved": True, "action": "created", "message": f"New slide saved at index {insert_index}.", "slide_id": str(new_slide.id), "index": insert_index, "shifted_slide_count": len(slides_to_shift), } async def delete_slide(self, *, index: int) -> dict[str, Any]: target_index = max(0, index) slide = await self._sql_session.scalar( select(SlideModel).where( SlideModel.presentation == self._presentation_id, SlideModel.index == target_index, ) ) if not slide: return { "deleted": False, "message": f"No slide found at index {target_index}.", "index": target_index, } presentation = await self._sql_session.get(PresentationModel, self._presentation_id) slides_result = await self._sql_session.scalars( select(SlideModel) .where(SlideModel.presentation == self._presentation_id) .order_by(SlideModel.index) ) slides = sorted(list(slides_result), key=lambda each: each.index) deleted_slide_id = str(slide.id) if len(slides) <= 1: fallback_slide = self._create_blank_slide_from_reference( presentation=presentation, source_slide=slide, index=0, ) await self._sql_session.delete(slide) if presentation: presentation.n_slides = 1 self._sql_session.add(presentation) self._sql_session.add(fallback_slide) await self._sql_session.commit() await self._sql_session.refresh(fallback_slide) return { "deleted": True, "message": "Deleted the final slide and added a blank fallback slide.", "deleted_slide_id": deleted_slide_id, "slide_id": str(fallback_slide.id), "index": 0, "slide_number": 1, "shifted_slide_count": 0, "blank_fallback": True, } await self._sql_session.delete(slide) remaining_slides = [ each_slide for each_slide in slides if each_slide.id != slide.id ] shifted_count = 0 for each_slide in remaining_slides: if each_slide.index <= target_index: continue each_slide.index -= 1 self._sql_session.add(each_slide) shifted_count += 1 if presentation: presentation.n_slides = len(remaining_slides) self._sql_session.add(presentation) await self._sql_session.commit() return { "deleted": True, "message": f"Slide at index {target_index} was deleted successfully.", "deleted_slide_id": deleted_slide_id, "index": target_index, "shifted_slide_count": shifted_count, } async def _get_slide_by_index(self, index: int) -> SlideModel | None: return await self._sql_session.scalar( select(SlideModel).where( SlideModel.presentation == self._presentation_id, SlideModel.index == max(0, index), ) ) async def _get_current_theme(self) -> dict[str, Any] | None: presentation = await self._sql_session.get( PresentationModel, self._presentation_id, ) return ( copy.deepcopy(presentation.theme) if presentation and isinstance(presentation.theme, dict) else None ) @staticmethod def _blank_slide_ui() -> dict[str, Any]: return copy.deepcopy(BLANK_TEMPLATE_V2_LAYOUT) def _create_blank_slide_from_reference( self, *, presentation: PresentationModel | None, source_slide: SlideModel, index: int, ) -> SlideModel: layout_group = ( source_slide.layout_group.strip() if isinstance(source_slide.layout_group, str) else "" ) if not layout_group and presentation: layout_group = self._resolve_layout_group(presentation=presentation) if not layout_group: layout_group = "presentation" layout = ( f"{layout_group}:{BLANK_SLIDE_LAYOUT_ID}" if layout_group.startswith("custom-") else BLANK_SLIDE_LAYOUT_ID ) return SlideModel( presentation=self._presentation_id, layout_group=layout_group, layout=layout, index=index, content={}, speaker_note="", ui=self._blank_slide_ui(), ) @staticmethod def _slide_ui_layout(slide: SlideModel) -> dict[str, Any] | None: ui = slide.ui if not isinstance(ui, dict): return None if not isinstance(ui.get("components"), list): return None return ui async def _save_slide_ui(self, slide: SlideModel, ui: dict[str, Any]) -> None: # Persist the mutated raw layout dict directly. We intentionally avoid # round-tripping through pydantic so richer runtime fields on the slide # UI (assets, tiptap ids, etc.) are preserved untouched. slide.ui = ui self._sql_session.add(slide) await self._sql_session.commit() await self._sql_session.refresh(slide) async def get_slide_ui_elements( self, *, index: int, include_full_json: bool = False ) -> dict[str, Any]: # Imported lazily so chat startup does not load rendered-slide helper code # unless the assistant actually inspects UI elements. from services.chat.slide_ui_helpers import ( _collect_editable_elements, _compact_components, ) slide = await self._get_slide_by_index(index) if not slide: return { "found": False, "message": f"No slide found at index {max(0, index)}.", } ui = self._slide_ui_layout(slide) if ui is None: return { "found": True, "editable": False, "index": slide.index, "slide_number": slide.index + 1, "message": ( "This slide is not a rendered template (ui) slide. Use " "saveSlide or updateSlide to edit it instead." ), } editable = _collect_editable_elements(ui, include_visual_elements=True) response: dict[str, Any] = { "found": True, "editable": True, "index": slide.index, "slide_number": slide.index + 1, "layout_id": ui.get("id"), "description": ui.get("description"), "component_count": len(ui.get("components", [])), "components": _compact_components(ui), "editable_count": len(editable), "elements": editable, "message": ( f"Slide {slide.index + 1} renders from its ui layout with " f"{len(ui.get('components', []))} component(s) and " f"{len(editable)} editable element(s)." ), } if include_full_json: response["ui"] = ui return response async def update_slide_ui_element( self, *, index: int, element_path: str, text: str | None = None, items: list[str] | None = None, table_cell: dict[str, Any] | None = None, table: dict[str, Any] | None = None, chart: dict[str, Any] | None = None, element_patch: dict[str, Any] | None = None, position: dict[str, Any] | None = None, size: dict[str, Any] | None = None, ) -> dict[str, Any]: from services.chat.slide_ui_helpers import ( _apply_element_style_patch, _apply_image_element_value, _component_id_for_path, _content_update_requested_for_type, _looks_like_asset_reference, _normalize_chart_element, _resolve_element_path, _resolve_image_update_payload, _update_chart_element, _update_table_element, _update_table_cell, _update_text_element, _update_text_list_element, ) slide = await self._get_slide_by_index(index) if not slide: return {"updated": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "updated": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } ui = copy.deepcopy(ui) element = _resolve_element_path(ui, element_path) element_type = str(element.get("type") or "") theme = await self._get_current_theme() element_updated = False if element_patch is not None: if "type" in element_patch and not isinstance(element_patch.get("type"), str): raise ValueError("element.type must be a string when provided.") self._merge_ui_patch(element, element_patch) self._sync_ui_text_fields(element) _apply_element_style_patch(element, element_patch) element_type = str(element.get("type") or element_type) if element_type == "chart": _normalize_chart_element(element, theme) element_updated = True content_update_requested = _content_update_requested_for_type( element_type, text=text, items=items, table_cell=table_cell, table=table, chart=chart, ) if content_update_requested and element_type == "text": if text is None: raise ValueError("text is required for text elements.") _update_text_element(element, text) elif content_update_requested and element_type == "text-list": if items is None: raise ValueError("items is required for text-list elements.") _update_text_list_element(element, items) elif content_update_requested and element_type == "table": if table is not None: _update_table_element(element, table) elif table_cell is not None: _update_table_cell(element, table_cell) else: raise ValueError("table or tableCell is required for table elements.") elif content_update_requested and element_type == "chart": if chart is None: raise ValueError("chart is required for chart elements.") _update_chart_element(element, chart, theme) elif content_update_requested and element_type == "image": payload = _resolve_image_update_payload(text, items) if payload is None: raise ValueError( "Image/icon updates require `text` with a URL returned by " "generateAssets, generateImage, or generateIcon." ) if isinstance(payload, str) and not _looks_like_asset_reference(payload): generated_url = await ( self.generate_icon(payload) if element.get("is_icon") is True else self.generate_image(payload) ) element["data"] = generated_url element["prompt"] = payload.strip() else: _apply_image_element_value(element, payload) elif content_update_requested: raise ValueError(f"Element type '{element_type}' is not content-editable.") geometry_updated = self._update_ui_box(element, position=position, size=size) if not content_update_requested and not geometry_updated and not element_updated: raise ValueError("No element content or geometry update was provided.") await self._save_slide_ui(slide, ui) component_id = _component_id_for_path(ui, element_path) return { "updated": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "element_path": element_path, "element_type": element_type, "position": element.get("position"), "size": element.get("size"), "message": ( f"Updated {element_type} content on slide {slide.index + 1}." ), } async def update_slide_ui_component( self, *, index: int, component_id: str, action: str | None = None, component_ids: list[str] | None = None, position: dict[str, Any] | None = None, size: dict[str, Any] | None = None, replacement_component: dict[str, Any] | None = None, ) -> dict[str, Any]: slide = await self._get_slide_by_index(index) if not slide: return {"updated": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "updated": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } ui = copy.deepcopy(ui) components = ui.get("components") if not isinstance(components, list): return {"updated": False, "message": "Slide has no components list."} normalized_action = self._normalize_component_action(action) if normalized_action == "ungroup": return await self._ungroup_slide_ui_component( slide=slide, ui=ui, components=components, component_id=component_id, ) if normalized_action == "group": return await self._group_slide_ui_components( slide=slide, ui=ui, components=components, component_id=component_id, component_ids=component_ids or [component_id], ) if normalized_action == "duplicate": return await self._duplicate_slide_ui_component( slide=slide, ui=ui, components=components, component_id=component_id, ) if normalized_action in { "bring-to-front", "bring-forward", "send-backward", "send-to-back", }: return await self._reorder_slide_ui_component( slide=slide, ui=ui, components=components, component_id=component_id, action=normalized_action, ) component = next( ( candidate for candidate in components if isinstance(candidate, dict) and candidate.get("id") == component_id ), None, ) if component is None: return { "updated": False, "message": f"Component '{component_id}' was not found.", } updated = False if replacement_component is not None: from services.chat.slide_ui_helpers import ( _normalize_chart_tree, _normalize_image_tree, _validate_chart_insert_tree, ) if not isinstance(replacement_component.get("elements"), list): raise ValueError("replacement component must include elements.") replacement = copy.deepcopy(replacement_component) _normalize_image_tree(replacement) _validate_chart_insert_tree(replacement) replacement["id"] = component_id replacement.setdefault("description", component.get("description")) replacement.setdefault("position", component.get("position")) replacement.setdefault("size", component.get("size")) component_index = components.index(component) components[component_index] = replacement component = replacement self._sync_ui_text_fields(component) self._normalize_added_visual_block(component) _normalize_chart_tree(component, await self._get_current_theme()) self._fit_component_to_stage(component) updated = True if size is not None: # Match the editor's corner-resize behavior. Updating only the # component frame leaves a standalone chart/image/table at its old # size and creates empty space inside the enlarged selection box. self._scale_component_contents_for_resize(component, size) if self._update_ui_box(component, position=position, size=size): updated = True if not updated: raise ValueError("No component geometry update was provided.") await self._save_slide_ui(slide, ui) return { "updated": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "position": component.get("position"), "size": component.get("size"), "message": f"Updated component '{component_id}' on slide {slide.index + 1}.", } async def _ungroup_slide_ui_component( self, *, slide: SlideModel, ui: dict[str, Any], components: list[Any], component_id: str, ) -> dict[str, Any]: from services.chat.slide_ui_helpers import _ungrouped_components_from_component component_index, component = next( ( (idx, candidate) for idx, candidate in enumerate(components) if isinstance(candidate, dict) and candidate.get("id") == component_id ), (None, None), ) if component_index is None or not isinstance(component, dict): return { "updated": False, "message": f"Component '{component_id}' was not found.", } parts = _ungrouped_components_from_component( component, component_index, used_ids={ str(item.get("id")) for idx, item in enumerate(components) if idx != component_index and isinstance(item, dict) }, ) if not parts: raise ValueError("Component does not contain safely separable elements.") components[component_index : component_index + 1] = parts await self._save_slide_ui(slide, ui) return { "updated": True, "action": "ungrouped", "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "created_component_ids": [part["id"] for part in parts], "message": f"Ungrouped component '{component_id}' on slide {slide.index + 1}.", } async def _group_slide_ui_components( self, *, slide: SlideModel, ui: dict[str, Any], components: list[Any], component_id: str, component_ids: list[str], ) -> dict[str, Any]: requested_ids = [] for raw_id in [component_id, *component_ids]: if raw_id and raw_id not in requested_ids: requested_ids.append(raw_id) if len(requested_ids) < 2: raise ValueError("Grouping requires at least two component ids.") selected: list[tuple[int, dict[str, Any]]] = [] for idx, component in enumerate(components): if isinstance(component, dict) and component.get("id") in requested_ids: selected.append((idx, component)) found_ids = {str(component.get("id")) for _, component in selected} missing_ids = [item for item in requested_ids if item not in found_ids] if missing_ids: raise ValueError(f"Component(s) not found: {', '.join(missing_ids)}") boxes = [ self._component_box(component) for _, component in selected ] if any(box is None for box in boxes): raise ValueError("Cannot group components without valid position and size.") typed_boxes = [box for box in boxes if box is not None] left = min(box["x"] for box in typed_boxes) top = min(box["y"] for box in typed_boxes) right = max(box["x"] + box["width"] for box in typed_boxes) bottom = max(box["y"] + box["height"] for box in typed_boxes) grouped_elements: list[dict[str, Any]] = [] for (_, component), box in zip(selected, typed_boxes): elements = component.get("elements") if not isinstance(elements, list): continue for element in elements: if not isinstance(element, dict): continue copied = copy.deepcopy(element) element_position = copied.get("position") if isinstance(element_position, dict): copied["position"] = { "x": float(box["x"]) + float(element_position.get("x") or 0) - left, "y": float(box["y"]) + float(element_position.get("y") or 0) - top, } else: copied["position"] = { "x": float(box["x"]) - left, "y": float(box["y"]) - top, } grouped_elements.append(copied) if not grouped_elements: raise ValueError("Cannot group components with no elements.") first_index = min(idx for idx, _ in selected) selected_indices = {idx for idx, _ in selected} group_id = self._unique_ui_component_id(component_id or "group", [ component for idx, component in enumerate(components) if idx not in selected_indices ]) group_component = { "id": group_id, "description": "Grouped component", "position": {"x": left, "y": top}, "size": {"width": right - left, "height": bottom - top}, "elements": grouped_elements, } next_components = [ component for idx, component in enumerate(components) if idx not in selected_indices ] next_components.insert(first_index, group_component) ui["components"] = next_components await self._save_slide_ui(slide, ui) return { "updated": True, "action": "grouped", "index": slide.index, "slide_number": slide.index + 1, "component_id": group_id, "grouped_component_ids": requested_ids, "message": f"Grouped {len(selected)} components on slide {slide.index + 1}.", } async def _duplicate_slide_ui_component( self, *, slide: SlideModel, ui: dict[str, Any], components: list[Any], component_id: str, ) -> dict[str, Any]: component_index, component = self._find_ui_component(components, component_id) if component_index is None or not isinstance(component, dict): return { "updated": False, "message": f"Component '{component_id}' was not found.", } duplicate = copy.deepcopy(component) duplicate_id = self._unique_ui_component_id(f"{component_id}_copy", components) duplicate["id"] = duplicate_id position = duplicate.get("position") if isinstance(position, dict): duplicate["position"] = { **position, "x": float(position.get("x") or 0) + 16, "y": float(position.get("y") or 0) + 16, } components.insert(component_index + 1, duplicate) await self._save_slide_ui(slide, ui) return { "updated": True, "action": "duplicated", "index": slide.index, "slide_number": slide.index + 1, "component_id": duplicate_id, "source_component_id": component_id, "message": f"Duplicated component '{component_id}' on slide {slide.index + 1}.", } async def _reorder_slide_ui_component( self, *, slide: SlideModel, ui: dict[str, Any], components: list[Any], component_id: str, action: str, ) -> dict[str, Any]: component_index, component = self._find_ui_component(components, component_id) if component_index is None or component is None: return { "updated": False, "message": f"Component '{component_id}' was not found.", } target_index = self._component_layer_target_index( component_index, len(components), action, ) if target_index == component_index: return { "updated": False, "action": action, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "message": f"Component '{component_id}' is already at that layer.", } moved = components.pop(component_index) components.insert(target_index, moved) await self._save_slide_ui(slide, ui) return { "updated": True, "action": action, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "component_index": target_index, "message": f"Reordered component '{component_id}' on slide {slide.index + 1}.", } @staticmethod def _normalize_component_action(action: str | None) -> str: return { "bringToFront": "bring-to-front", "bringForward": "bring-forward", "sendBackward": "send-backward", "sendToBack": "send-to-back", }.get(action or "update", action or "update") @staticmethod def _component_layer_target_index( component_index: int, component_count: int, action: str, ) -> int: if component_count <= 1: return component_index if action == "send-to-back": return 0 if action == "send-backward": return max(0, component_index - 1) if action == "bring-forward": return min(component_count - 1, component_index + 1) if action == "bring-to-front": return component_count - 1 return component_index @staticmethod def _find_ui_component( components: list[Any], component_id: str, ) -> tuple[int | None, dict[str, Any] | None]: return next( ( (idx, candidate) for idx, candidate in enumerate(components) if isinstance(candidate, dict) and candidate.get("id") == component_id ), (None, None), ) @staticmethod def _component_box(component: dict[str, Any]) -> dict[str, float] | None: position = component.get("position") size = component.get("size") if not isinstance(position, dict) or not isinstance(size, dict): return None x = position.get("x") y = position.get("y") width = size.get("width") height = size.get("height") if not all(isinstance(value, (int, float)) for value in (x, y, width, height)): return None if width <= 0 or height <= 0: return None return { "x": float(x), "y": float(y), "width": float(width), "height": float(height), } @classmethod def _scale_component_contents_for_resize( cls, component: dict[str, Any], target_size: dict[str, Any], ) -> None: current_size = component.get("size") if not isinstance(current_size, dict): return current_width = current_size.get("width") current_height = current_size.get("height") target_width = target_size.get("width") target_height = target_size.get("height") if not all( isinstance(value, (int, float)) for value in (current_width, current_height, target_width, target_height) ): return if current_width <= 0 or current_height <= 0: return scale_x = float(target_width) / float(current_width) scale_y = float(target_height) / float(current_height) if abs(scale_x - 1.0) < 0.001 and abs(scale_y - 1.0) < 0.001: return elements = component.get("elements") if not isinstance(elements, list): return font_scale = (scale_x * scale_y) ** 0.5 for element in elements: if isinstance(element, dict): cls._scale_ui_element_for_component_resize( element, scale_x=scale_x, scale_y=scale_y, font_scale=font_scale, ) @classmethod def _scale_ui_element_for_component_resize( cls, element: dict[str, Any], *, scale_x: float, scale_y: float, font_scale: float, ) -> None: position = element.get("position") if isinstance(position, dict): x = position.get("x") y = position.get("y") if isinstance(x, (int, float)) and isinstance(y, (int, float)): element["position"] = { "x": float(x) * scale_x, "y": float(y) * scale_y, } size = element.get("size") if isinstance(size, dict): width = size.get("width") height = size.get("height") if isinstance(width, (int, float)) and isinstance(height, (int, float)): element["size"] = { "width": max(1.0, float(width) * scale_x), "height": max(1.0, float(height) * scale_y), } if str(element.get("type") or "").lower() in {"text", "text-list", "table"}: cls._scale_font_metrics_for_component_resize(element, font_scale) for key in ("children", "elements"): children = element.get(key) if isinstance(children, list): for child in children: if isinstance(child, dict): cls._scale_ui_element_for_component_resize( child, scale_x=scale_x, scale_y=scale_y, font_scale=font_scale, ) child = element.get("child") if isinstance(child, dict): cls._scale_ui_element_for_component_resize( child, scale_x=scale_x, scale_y=scale_y, font_scale=font_scale, ) @classmethod def _scale_font_metrics_for_component_resize( cls, value: Any, scale: float, ) -> None: if isinstance(value, list): for item in value: cls._scale_font_metrics_for_component_resize(item, scale) return if not isinstance(value, dict): return font = value.get("font") if isinstance(font, dict): size = font.get("size") if isinstance(size, (int, float)): font["size"] = min(512.0, max(1.0, round(float(size) * scale, 2))) for key in ("letter_spacing", "letterSpacing"): spacing = font.get(key) if isinstance(spacing, (int, float)): font[key] = round(float(spacing) * scale, 2) # Child layout nodes are scaled separately so their text metrics are # not applied twice. These fields contain only text/table content. for key, child_value in value.items(): if key in {"font", "children", "elements", "child"}: continue if isinstance(child_value, (dict, list)): cls._scale_font_metrics_for_component_resize(child_value, scale) async def delete_slide_ui_component( self, *, index: int, component_id: str ) -> dict[str, Any]: slide = await self._get_slide_by_index(index) if not slide: return {"deleted": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "deleted": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } ui = copy.deepcopy(ui) components = ui.get("components") before = len(components) ui["components"] = [ component for component in components if not ( isinstance(component, dict) and component.get("id") == component_id ) ] if len(ui["components"]) == before: return { "deleted": False, "message": f"Component '{component_id}' was not found on slide {slide.index + 1}.", } await self._save_slide_ui(slide, ui) return { "deleted": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "message": ( f"Deleted component '{component_id}' from slide {slide.index + 1}." ), } async def delete_slide_ui_element( self, *, index: int, element_path: str ) -> dict[str, Any]: from services.chat.slide_ui_helpers import _component_id_for_path slide = await self._get_slide_by_index(index) if not slide: return {"deleted": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "deleted": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } ui = copy.deepcopy(ui) removed = self._remove_element_at_path(ui, element_path) if not removed: return { "deleted": False, "message": ( f"Could not delete element at '{element_path}'. Only indexed " "elements[] / children[] entries can be removed; to remove a whole " "component use deleteComponent." ), } component_id = _component_id_for_path(ui, element_path) await self._save_slide_ui(slide, ui) return { "deleted": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "element_path": element_path, "message": f"Deleted element '{element_path}' on slide {slide.index + 1}.", } async def add_slide_ui_component( self, *, index: int, component: dict[str, Any], insert_index: int | None = None, ) -> dict[str, Any]: slide = await self._get_slide_by_index(index) if not slide: return {"added": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "added": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } if not isinstance(component, dict): raise ValueError("component must be a JSON object.") elements = component.get("elements") if not isinstance(elements, list) or not elements: raise ValueError("component.elements must be a non-empty list.") from services.chat.slide_ui_helpers import ( _normalize_chart_tree, _normalize_image_tree, _validate_chart_insert_tree, ) ui = copy.deepcopy(ui) components = ui.get("components") new_component = copy.deepcopy(component) _normalize_image_tree(new_component) _validate_chart_insert_tree(new_component) existing_ids = { str(existing.get("id")) for existing in components if isinstance(existing, dict) and existing.get("id") } new_id = str(new_component.get("id") or "").strip() if not new_id or new_id in existing_ids: base = new_id or "component" suffix = len(components) + 1 candidate = f"{base}_{suffix}" while candidate in existing_ids: suffix += 1 candidate = f"{base}_{suffix}" new_id = candidate new_component["id"] = new_id new_component.setdefault( "description", f"Component {new_id} added via assistant." ) # Keep assistant-authored text content in the canonical renderable shape. self._sync_ui_text_fields(new_component) self._normalize_added_visual_block(new_component) _normalize_chart_tree(new_component, await self._get_current_theme()) self._fit_component_to_stage(new_component) position = ( len(components) if insert_index is None else min(max(0, insert_index), len(components)) ) components.insert(position, new_component) await self._save_slide_ui(slide, ui) return { "added": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": new_id, "component_index": position, "message": ( f"Added component '{new_id}' to slide {slide.index + 1}." ), } async def add_slide_ui_element( self, *, index: int, element: dict[str, Any], component_id: str | None = None, insert_index: int | None = None, ) -> dict[str, Any]: slide = await self._get_slide_by_index(index) if not slide: return {"added": False, "message": f"No slide found at index {max(0, index)}."} ui = self._slide_ui_layout(slide) if ui is None: return { "added": False, "message": "This slide has no editable ui layout; use saveSlide instead.", } from services.chat.slide_ui_helpers import ( _normalize_chart_element, _normalize_image_tree, _validate_chart_insert_tree, ) ui = copy.deepcopy(ui) components = ui.get("components") if not isinstance(components, list): return {"added": False, "message": "Slide has no components list."} new_element = copy.deepcopy(element) _normalize_image_tree(new_element) _validate_chart_insert_tree(new_element) if component_id: component_index, component = next( ( (idx, candidate) for idx, candidate in enumerate(components) if isinstance(candidate, dict) and candidate.get("id") == component_id ), (None, None), ) if component_index is None or not isinstance(component, dict): return { "added": False, "message": f"Component '{component_id}' was not found.", } elements = component.setdefault("elements", []) if not isinstance(elements, list): raise ValueError("Target component has no elements list.") position = len(elements) if insert_index is None else min(max(0, insert_index), len(elements)) elements.insert(position, new_element) path = f"components[{component_index}].elements[{position}]" else: position_box = new_element.get("position") if isinstance(new_element.get("position"), dict) else {} size_box = new_element.get("size") if isinstance(new_element.get("size"), dict) else {} component = { "id": self._unique_ui_component_id( str(new_element.get("name") or new_element.get("type") or "element"), components, ), "description": f"Element {new_element.get('type') or ''} added via assistant.".strip(), "position": { "x": float(position_box.get("x") or 128), "y": float(position_box.get("y") or 120), }, "size": { "width": float(size_box.get("width") or 320), "height": float(size_box.get("height") or 120), }, "elements": [new_element], } new_element["position"] = {"x": 0, "y": 0} self._normalize_added_visual_block(component) self._fit_component_to_stage(component) if isinstance(component.get("size"), dict): new_element["size"] = copy.deepcopy(component["size"]) component_position = len(components) if insert_index is None else min(max(0, insert_index), len(components)) components.insert(component_position, component) component_id = str(component["id"]) path = f"components[{component_position}].elements[0]" self._sync_ui_text_fields(new_element) if new_element.get("type") == "chart": _normalize_chart_element(new_element, await self._get_current_theme()) await self._save_slide_ui(slide, ui) return { "added": True, "index": slide.index, "slide_number": slide.index + 1, "component_id": component_id, "element_path": path, "message": f"Added element on slide {slide.index + 1}.", } @staticmethod def _unique_ui_component_id(base: str, components: list[Any]) -> str: normalized = re.sub(r"[^a-z0-9]+", "_", base.strip().lower()).strip("_") or "component" used = { str(component.get("id")) for component in components if isinstance(component, dict) and component.get("id") } candidate = normalized[:80] suffix = 2 while candidate in used: suffix_text = f"_{suffix}" candidate = f"{normalized[: 80 - len(suffix_text)]}{suffix_text}" suffix += 1 return candidate @staticmethod def _merge_ui_patch(target: dict[str, Any], patch: dict[str, Any]) -> None: for key, value in patch.items(): if isinstance(value, dict) and isinstance(target.get(key), dict): PresentationChatMemoryLayer._merge_ui_patch(target[key], value) else: target[key] = copy.deepcopy(value) @staticmethod def _sync_ui_text_fields(node: Any) -> None: """Recursively keep assistant-authored text content renderable.""" if isinstance(node, dict): if node.get("type") == "text": runs = node.get("runs") if isinstance(runs, list) and runs: joined = "".join( str(run.get("text") or "") for run in runs if isinstance(run, dict) ) node["text"] = joined elif isinstance(node.get("text"), str): node["runs"] = [{"text": node["text"]}] elif node.get("type") == "table": PresentationChatMemoryLayer._sync_ui_table_cells(node) for value in node.values(): PresentationChatMemoryLayer._sync_ui_text_fields(value) elif isinstance(node, list): for value in node: PresentationChatMemoryLayer._sync_ui_text_fields(value) @staticmethod def _sync_ui_table_cells(element: dict[str, Any]) -> None: fallback_font = element.get("font") columns = element.get("columns") if not isinstance(columns, list) and isinstance(element.get("headers"), list): columns = element["headers"] if isinstance(columns, list): element["columns"] = [ PresentationChatMemoryLayer._normalized_ui_table_cell( cell, fallback_font, ) for cell in columns ] rows = element.get("rows") if isinstance(rows, list): element["rows"] = [ [ PresentationChatMemoryLayer._normalized_ui_table_cell( cell, fallback_font, ) for cell in row ] if isinstance(row, list) else row for row in rows ] @staticmethod def _normalized_ui_table_cell(cell: Any, fallback_font: Any) -> dict[str, Any]: if isinstance(cell, dict): normalized = copy.deepcopy(cell) cell_font = normalized.get("font") or fallback_font existing_runs = normalized.get("runs") normalized_runs = ( PresentationChatMemoryLayer._normalized_ui_text_runs( existing_runs, cell_font, ) if isinstance(existing_runs, list) else [] ) alias_text = PresentationChatMemoryLayer._table_cell_alias_text(normalized) if normalized_runs and ( PresentationChatMemoryLayer._runs_plain_text(normalized_runs) or not alias_text ): normalized["runs"] = normalized_runs else: normalized["runs"] = PresentationChatMemoryLayer._replacement_runs_from_existing( existing_runs, alias_text, cell_font, ) return normalized return { "runs": PresentationChatMemoryLayer._replacement_runs_from_existing( None, PresentationChatMemoryLayer._table_cell_text_value(cell), fallback_font, ) } @staticmethod def _normalized_ui_text_runs(runs: list[Any], fallback_font: Any) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for run in runs: if isinstance(run, dict): next_run = copy.deepcopy(run) next_run["text"] = PresentationChatMemoryLayer._table_cell_text_value( next_run.get("text") if "text" in next_run else next_run.get("content") if "content" in next_run else next_run.get("value") ) normalized.append(next_run) else: next_run = { "text": PresentationChatMemoryLayer._table_cell_text_value(run) } if isinstance(fallback_font, dict): next_run["font"] = copy.deepcopy(fallback_font) normalized.append(next_run) return normalized @staticmethod def _runs_plain_text(runs: list[dict[str, Any]]) -> str: return "".join(str(run.get("text") or "") for run in runs) @staticmethod def _table_cell_alias_text(cell: dict[str, Any]) -> str: for key in ("text", "content", "value", "label", "data"): if key in cell: return PresentationChatMemoryLayer._table_cell_text_value(cell[key]) return "" @staticmethod def _table_cell_text_value(value: Any) -> str: if value is None: return "" if isinstance(value, (str, int, float, bool)): return str(value) if isinstance(value, list): return "".join( PresentationChatMemoryLayer._table_cell_text_value(item) for item in value ) if isinstance(value, dict): runs = value.get("runs") if isinstance(runs, list): text = "".join( PresentationChatMemoryLayer._table_cell_text_value(run) for run in runs ) if text: return text for key in ("text", "content", "value", "label", "data"): if key in value: return PresentationChatMemoryLayer._table_cell_text_value(value[key]) return "" return str(value) @staticmethod def _normalize_added_visual_block(component: dict[str, Any]) -> None: kind = PresentationChatMemoryLayer._insert_visual_kind(component) if kind is None: return defaults = DEFAULT_INSERT_BOXES[kind] min_size = defaults["min_size"] if PresentationChatMemoryLayer._box_too_small( component.get("size"), min_size, ): component["position"] = copy.deepcopy(defaults["position"]) component["size"] = copy.deepcopy(defaults["size"]) component_size = component.get("size") if not isinstance(component_size, dict): return for element in component.get("elements", []): if not isinstance(element, dict) or element.get("type") != kind: continue if PresentationChatMemoryLayer._box_too_small( element.get("size"), min_size, ): element["position"] = {"x": 0, "y": 0} element["size"] = { "width": float(component_size["width"]), "height": float(component_size["height"]), } @staticmethod def _fit_component_to_stage(component: dict[str, Any]) -> None: position = component.get("position") size = component.get("size") if not isinstance(position, dict) or not isinstance(size, dict): return x = position.get("x") y = position.get("y") width = size.get("width") height = size.get("height") if not all(isinstance(value, (int, float)) for value in (x, y, width, height)): return width = min(max(1.0, float(width)), SLIDE_STAGE_WIDTH) height = min(max(1.0, float(height)), SLIDE_STAGE_HEIGHT) component["size"] = {"width": width, "height": height} component["position"] = { "x": min(max(0.0, float(x)), SLIDE_STAGE_WIDTH - width), "y": min(max(0.0, float(y)), SLIDE_STAGE_HEIGHT - height), } @staticmethod def _insert_visual_kind(component: dict[str, Any]) -> str | None: elements = component.get("elements") if not isinstance(elements, list): return None types = { element.get("type") for element in elements if isinstance(element, dict) } if "chart" in types: return "chart" if "table" in types: return "table" return None @staticmethod def _box_too_small(size: Any, min_size: dict[str, float]) -> bool: if not isinstance(size, dict): return True width = size.get("width") height = size.get("height") if not isinstance(width, (int, float)) or not isinstance(height, (int, float)): return True return width < min_size["width"] or height < min_size["height"] @staticmethod def _remove_element_at_path(ui: dict[str, Any], path: str) -> bool: segments = path.split(".") if not segments: return False last = segments[-1] match = re.match(r"^(components|elements|children)\[(\d+)\]$", last) if not match: return False key = match.group(1) target_index = int(match.group(2)) parent: Any = ui for segment in segments[:-1]: if segment == "child": if not isinstance(parent, dict) or not isinstance( parent.get("child"), dict ): return False parent = parent["child"] continue seg_match = re.match(r"^(components|elements|children)\[(\d+)\]$", segment) if not seg_match: return False seg_key = seg_match.group(1) seg_index = int(seg_match.group(2)) if not isinstance(parent, dict) or not isinstance(parent.get(seg_key), list): return False values = parent[seg_key] if seg_index >= len(values) or not isinstance(values[seg_index], dict): return False parent = values[seg_index] if not isinstance(parent, dict) or not isinstance(parent.get(key), list): return False values = parent[key] if target_index >= len(values): return False values.pop(target_index) return True @staticmethod def _update_ui_box( target: dict[str, Any], *, position: dict[str, Any] | None = None, size: dict[str, Any] | None = None, ) -> bool: updated = False if position is not None: target["position"] = { "x": float(position["x"]), "y": float(position["y"]), } updated = True if size is not None: target["size"] = { "width": float(size["width"]), "height": float(size["height"]), } updated = True return updated async def set_presentation_theme( self, *, theme_query: str | None = None, custom_theme: dict[str, Any] | None = None, save_custom_theme: bool = True, ) -> dict[str, Any]: requested_theme = (theme_query or "").strip() has_custom_theme = isinstance(custom_theme, dict) if not requested_theme and not has_custom_theme: return { "applied": False, "message": "Theme query or custom theme payload is required.", } presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "applied": False, "message": "Presentation not found.", } current_theme = ( presentation.theme if isinstance(presentation.theme, dict) else None ) available_themes = await self._get_chat_available_themes() selected_theme: dict[str, Any] | None = None custom_theme_saved = False selected_source = "query" if has_custom_theme: custom_theme_payload = custom_theme if isinstance(custom_theme, dict) else {} selected_theme = self._build_custom_theme_from_payload( custom_theme=custom_theme_payload, requested_theme=requested_theme, current_theme=current_theme, available_themes=available_themes, ) if not selected_theme: return { "applied": False, "message": ( "Invalid custom theme payload. Include colors and optional font " "details (name/url), or use a theme name/id query." ), "requested_theme": requested_theme or None, } selected_source = "custom" if save_custom_theme: await self._upsert_custom_theme_in_store(selected_theme) custom_theme_saved = True else: selected_theme = self._select_theme_for_query( requested_theme, available_themes, current_theme, ) if not selected_theme: return { "applied": False, "message": ( "No matching theme found. Try a specific theme name/id, " "use 'dark'/'light'/'another', or provide customTheme." ), "requested_theme": requested_theme, "available_themes": [ {"id": str(theme.get("id") or ""), "name": str(theme.get("name") or "")} for theme in available_themes ], } previous_theme = copy.deepcopy(current_theme) if current_theme else None presentation.theme = copy.deepcopy(selected_theme) self._sql_session.add(presentation) await self._sql_session.commit() selected_name = str(selected_theme.get("name") or "selected theme") selected_id = str(selected_theme.get("id") or "") previous_name = self._extract_theme_name(previous_theme) return { "applied": True, "message": f"Theme changed to '{selected_name}'.", "requested_theme": requested_theme or None, "theme": selected_theme, "theme_id": selected_id, "theme_name": selected_name, "theme_source": selected_source, "custom_theme_saved": custom_theme_saved, "previous_theme_name": previous_name, } async def get_presentation_theme_catalog(self) -> dict[str, Any]: presentation = await self._sql_session.get(PresentationModel, self._presentation_id) if not presentation: return { "found": False, "message": "Presentation not found.", "current_theme": None, "available_themes": [], "count": 0, } current_theme = ( copy.deepcopy(presentation.theme) if isinstance(presentation.theme, dict) else None ) current_theme_id = ( str((current_theme or {}).get("id") or "").strip().lower() if current_theme else "" ) builtin_theme_ids = { str(theme.get("id") or "").strip().lower() for theme in CHAT_BUILTIN_THEMES } available_themes = await self._get_chat_available_themes() catalog: list[dict[str, Any]] = [] for theme in available_themes: theme_id = str(theme.get("id") or "").strip() theme_name = str(theme.get("name") or "").strip() if not theme_id and not theme_name: continue normalized_theme_id = theme_id.lower() catalog.append( { "id": theme_id, "name": theme_name or theme_id, "description": str(theme.get("description") or "").strip(), "source": ( "built_in" if normalized_theme_id in builtin_theme_ids else "custom" ), "is_current": bool( current_theme_id and normalized_theme_id and normalized_theme_id == current_theme_id ), } ) current_theme_summary: dict[str, Any] | None = None if current_theme: current_theme_colors = self._extract_theme_colors(current_theme) current_theme_summary = { "id": str(current_theme.get("id") or "").strip(), "name": str(current_theme.get("name") or "").strip(), "description": str(current_theme.get("description") or "").strip(), "colors": current_theme_colors, "chart_colors": self._chart_palette_from_theme_colors( current_theme_colors, ), } return { "found": True, "count": len(catalog), "current_theme": current_theme_summary, "available_themes": catalog, "available_theme_ids": [theme["id"] for theme in catalog if theme.get("id")], "message": "Theme catalog fetched successfully.", } async def retrieve_context(self, query: str) -> str: context = await MEM0_PRESENTATION_MEMORY_SERVICE.retrieve_context( self._presentation_id, query, ) if context: LOGGER.info( "Chat memory semantic context hit (presentation_id=%s, chars=%d)", self._presentation_id, len(context), ) else: LOGGER.info( "Chat memory semantic context miss (presentation_id=%s)", self._presentation_id, ) return context async def _collect_template_block_candidates( self, presentation: PresentationModel, ) -> list[dict[str, Any]]: sources: list[tuple[str, dict[str, Any]]] = [] if isinstance(presentation.layout, dict): sources.append(("presentation_layout", presentation.layout)) seen_template_ids: set[str] = set() for key in ("name", "template_id", "template_v2_id"): template_id = self._extract_template_v2_id( presentation.layout.get(key), allow_bare=key in {"template_id", "template_v2_id"}, ) if not template_id or template_id in seen_template_ids: continue seen_template_ids.add(template_id) template = await self._sql_session.get(TemplateV2, template_id) if template: sources.extend(self._template_block_sources(template)) candidates: list[dict[str, Any]] = [] seen_block_ids: set[str] = set() for source, payload in sources: candidates.extend( self._block_candidates_from_merged_components( payload, source=source, seen_block_ids=seen_block_ids, ) ) candidates.extend( self._block_candidates_from_layouts( payload, source=source, seen_block_ids=seen_block_ids, ) ) return candidates @staticmethod def _template_block_sources(template: TemplateV2) -> list[tuple[str, dict[str, Any]]]: sources: list[tuple[str, dict[str, Any]]] = [] for label, payload in ( ("template_merged_components", template.merged_components), ("template_components", template.components), ("template_layouts", template.layouts), ("template_raw_layouts", template.raw_layouts), ): if isinstance(payload, dict): sources.append((label, payload)) return sources @classmethod def _block_candidates_from_merged_components( cls, payload: dict[str, Any], *, source: str, seen_block_ids: set[str], ) -> list[dict[str, Any]]: raw_components = payload.get("components") if not isinstance(raw_components, list): return [] candidates: list[dict[str, Any]] = [] for group_index, group in enumerate(raw_components): if not isinstance(group, dict): continue variants = group.get("variants") if not isinstance(variants, list): variants = [group] variant_count = len([item for item in variants if isinstance(item, dict)]) for variant_index, component in enumerate(variants): if not isinstance(component, dict): continue component_id = str( component.get("id") or group.get("id") or f"component_{group_index + 1}" ) block_id = cls._unique_block_id( f"merged:{group.get('id') or component_id}:{variant_index}", seen_block_ids, ) candidates.append( cls._build_block_candidate( source=source, block_id=block_id, layout_id=None, layout_description=None, component=component, component_id=component_id, description=( group.get("description") or component.get("description") or "" ), variant_index=variant_index, variant_count=variant_count, ) ) return candidates @classmethod def _block_candidates_from_layouts( cls, payload: dict[str, Any], *, source: str, seen_block_ids: set[str], ) -> list[dict[str, Any]]: layouts = payload.get("layouts") if not isinstance(layouts, list): return [] candidates: list[dict[str, Any]] = [] for layout_index, layout in enumerate(layouts): if not isinstance(layout, dict): continue components = layout.get("components") if not isinstance(components, list): continue layout_id = str(layout.get("id") or f"layout_{layout_index + 1}") layout_description = ( str(layout.get("description")) if layout.get("description") is not None else None ) for component_index, component in enumerate(components): if not isinstance(component, dict): continue component_id = str( component.get("id") or f"component_{component_index + 1}" ) block_id = cls._unique_block_id( f"layout:{layout_id}:{component_id}", seen_block_ids, ) candidates.append( cls._build_block_candidate( source=source, block_id=block_id, layout_id=layout_id, layout_description=layout_description, component=component, component_id=component_id, description=str(component.get("description") or ""), ) ) return candidates @classmethod def _build_block_candidate( cls, *, source: str, block_id: str, layout_id: str | None, layout_description: str | None, component: dict[str, Any], component_id: str, description: str, variant_index: int | None = None, variant_count: int | None = None, ) -> dict[str, Any]: candidate = { "block_id": block_id, "source": source, "layout_id": layout_id, "layout_description": layout_description, "component_id": component_id, "description": description, "position": copy.deepcopy(component.get("position")), "size": copy.deepcopy(component.get("size")), "element_count": len(component.get("elements", [])) if isinstance(component.get("elements"), list) else 0, "element_types": cls._component_element_types(component), "element_names": cls._component_element_names(component), "decorative": cls._component_is_decorative(component), "component": copy.deepcopy(component), } if variant_index is not None: candidate["variant_index"] = variant_index if variant_count is not None: candidate["variant_count"] = variant_count return candidate @staticmethod def _component_element_types(component: dict[str, Any]) -> list[str]: seen: set[str] = set() ordered: list[str] = [] def visit(element: Any) -> None: if not isinstance(element, dict): return element_type = element.get("type") if element_type is not None: value = str(element_type) if value not in seen: seen.add(value) ordered.append(value) child = element.get("child") if isinstance(child, dict): visit(child) children = element.get("children") if isinstance(children, list): for nested in children: visit(nested) elements = component.get("elements") if isinstance(elements, list): for element in elements: visit(element) return ordered @staticmethod def _component_element_names(component: dict[str, Any]) -> list[str]: seen: set[str] = set() ordered: list[str] = [] def visit(element: Any) -> None: if not isinstance(element, dict): return name = element.get("name") if name is not None: value = str(name).strip() if value and value not in seen: seen.add(value) ordered.append(value) child = element.get("child") if isinstance(child, dict): visit(child) children = element.get("children") if isinstance(children, list): for nested in children: visit(nested) elements = component.get("elements") if isinstance(elements, list): for element in elements: visit(element) return ordered @staticmethod def _component_is_decorative(component: dict[str, Any]) -> bool: elements = component.get("elements") if not isinstance(elements, list) or not elements: return False return all( isinstance(element, dict) and element.get("decorative") is True for element in elements ) @staticmethod def _unique_block_id(block_id: str, seen_block_ids: set[str]) -> str: normalized = re.sub(r"[^a-zA-Z0-9:_-]+", "_", block_id).strip("_") normalized = normalized or "block" candidate = normalized suffix = 2 while candidate in seen_block_ids: candidate = f"{normalized}:{suffix}" suffix += 1 seen_block_ids.add(candidate) return candidate @staticmethod def _block_match_score(candidate: dict[str, Any], query: str) -> int: if not query: return 1 haystack = " ".join( str(value or "") for value in ( candidate.get("block_id"), candidate.get("component_id"), candidate.get("description"), candidate.get("layout_id"), candidate.get("layout_description"), " ".join(str(item) for item in candidate.get("element_types", [])), " ".join(str(item) for item in candidate.get("element_names", [])), ) ).lower() score = 0 if query in haystack: score += 10 for word in re.findall(r"[a-z0-9_-]+", query): if word in haystack: score += 1 return score @staticmethod def _format_available_block( candidate: dict[str, Any], *, include_full_content: bool, ) -> dict[str, Any]: keys = ( "block_id", "source", "layout_id", "layout_description", "component_id", "description", "position", "size", "element_count", "element_types", "element_names", "decorative", "variant_index", "variant_count", ) block = {key: candidate.get(key) for key in keys if key in candidate} if include_full_content: block["component"] = copy.deepcopy(candidate.get("component")) return block async def _get_layout_by_id( self, layout_id: str, presentation: PresentationModel | None = None, ) -> SlideLayoutModel | None: if not presentation: presentation = await self._sql_session.get( PresentationModel, self._presentation_id ) layout_model = await self._get_layout_model(presentation) if not layout_model: return None for layout in layout_model.slides: if layout.id == layout_id: return layout return None async def _get_layout_model( self, presentation: PresentationModel | None, ) -> PresentationLayoutModel | None: if not presentation or not isinstance(presentation.layout, dict): return None if self._is_template_v2_layout_payload(presentation.layout): return self._build_template_v2_layout_model( presentation.layout, layout_name=str(presentation.layout.get("name") or "template-v2"), ) try: return presentation.get_layout() except Exception: template_model = await self._resolve_template_v2_layout_model(presentation) if template_model: return template_model LOGGER.exception( "Failed to parse presentation layout (presentation_id=%s)", self._presentation_id, ) return None async def _resolve_template_v2_layout_model( self, presentation: PresentationModel, ) -> PresentationLayoutModel | None: candidate_ids: list[str] = [] seen_ids: set[str] = set() if isinstance(presentation.layout, dict): for key in ("name", "template_id", "template_v2_id"): template_id = self._extract_template_v2_id( presentation.layout.get(key), allow_bare=key in {"template_id", "template_v2_id"}, ) if template_id and template_id not in seen_ids: candidate_ids.append(template_id) seen_ids.add(template_id) for template_id in candidate_ids: template = await self._sql_session.get(TemplateV2, template_id) if not template or not isinstance(template.layouts, dict): continue return self._build_template_v2_layout_model( template.layouts, layout_name=f"template-v2-{template.id}", ) return None @staticmethod def _is_template_v2_layout_payload(layout_payload: Any) -> bool: return ( isinstance(layout_payload, dict) and isinstance(layout_payload.get("layouts"), list) ) @staticmethod def _extract_template_v2_id(value: Any, *, allow_bare: bool = False) -> str | None: if not isinstance(value, str) or not value: return None candidate = value.strip() if not candidate: return None for prefix in ("template-v2-", "template-v2:"): if candidate.startswith(prefix): candidate = candidate[len(prefix) :].strip() break else: if not allow_bare: return None return candidate or None @staticmethod def _build_template_v2_layout_model( layout_payload: dict[str, Any], *, layout_name: str, ) -> PresentationLayoutModel: template_schema = get_template_schema(layout_payload) source_layouts = layout_payload.get("layouts") if not isinstance(source_layouts, list): source_layouts = [] slides: list[SlideLayoutModel] = [] for index, schema_layout in enumerate(template_schema["layouts"]): if not isinstance(schema_layout, dict): continue source_layout = ( source_layouts[index] if index < len(source_layouts) and isinstance(source_layouts[index], dict) else {} ) layout_id = ( schema_layout.get("layout_id") or source_layout.get("id") or f"layout_{index + 1}" ) layout_schema = schema_layout.get("schema") if not isinstance(layout_schema, dict): layout_schema = { "title": str(layout_id), "description": source_layout.get("description"), } slides.append( SlideLayoutModel( id=str(layout_id), name=source_layout.get("name") or layout_schema.get("title"), description=source_layout.get("description") or layout_schema.get("description"), json_schema=layout_schema, ) ) return PresentationLayoutModel( name=layout_name, ordered=False, slides=slides, ) async def _build_template_v2_slide_ui( self, *, presentation: PresentationModel, layout_id: str, content: dict[str, Any], ) -> dict[str, Any] | None: source_layout = await self._get_template_v2_raw_layout_by_id( presentation=presentation, layout_id=layout_id, ) if source_layout is None: return None ui = copy.deepcopy(source_layout) theme = presentation.theme if isinstance(presentation.theme, dict) else None self._apply_template_v2_content_to_ui(ui, content, theme=theme) self._sync_ui_text_fields(ui) from services.chat.slide_ui_helpers import _normalize_chart_tree _normalize_chart_tree(ui, theme) return ui async def _get_template_v2_raw_layout_by_id( self, *, presentation: PresentationModel, layout_id: str, ) -> dict[str, Any] | None: if isinstance(presentation.layout, dict): source = self._raw_layout_from_payload(presentation.layout, layout_id) if source is not None: return source for key in ("name", "template_id", "template_v2_id"): template_id = self._extract_template_v2_id( presentation.layout.get(key), allow_bare=key in {"template_id", "template_v2_id"}, ) if not template_id: continue template = await self._sql_session.get(TemplateV2, template_id) if template and isinstance(template.layouts, dict): source = self._raw_layout_from_payload(template.layouts, layout_id) if source is not None: return source return None @staticmethod def _raw_layout_from_payload( layout_payload: dict[str, Any], layout_id: str, ) -> dict[str, Any] | None: layouts = layout_payload.get("layouts") if not isinstance(layouts, list): return None for layout in layouts: if isinstance(layout, dict) and str(layout.get("id")) == str(layout_id): return layout return None @classmethod def _apply_template_v2_content_to_ui( cls, ui: dict[str, Any], content: dict[str, Any], *, theme: dict[str, Any] | None = None, ) -> None: components = ui.get("components") if not isinstance(components, list): return component_counts: dict[str, int] = {} for component in components: if isinstance(component, dict): component_id = str(component.get("id") or "") component_counts[component_id] = ( component_counts.get(component_id, 0) + 1 ) component_seen: dict[str, int] = {} used_keys: set[str] = set() for component in components: if not isinstance(component, dict): continue component_id = str(component.get("id") or "") occurrence_index = component_seen.get(component_id, 0) component_seen[component_id] = occurrence_index + 1 candidate_keys = [component_id] if component_counts.get(component_id, 0) > 1: candidate_keys.insert(0, f"{component_id}_{occurrence_index}") component_content = None for key in candidate_keys: if key in used_keys: continue value = content.get(key) if isinstance(value, dict): component_content = value used_keys.add(key) break if component_content is None: continue cls._apply_template_v2_component_content( component, component_content, theme=theme, ) @classmethod def _apply_template_v2_component_content( cls, component: dict[str, Any], content: dict[str, Any], *, theme: dict[str, Any] | None = None, ) -> None: elements = component.get("elements") if not isinstance(elements, list): return for element in elements: if isinstance(element, dict): cls._apply_template_v2_element_content( element, content, theme=theme, ) @classmethod def _apply_template_v2_element_content( cls, element: dict[str, Any], content: Any, *, theme: dict[str, Any] | None = None, direct_value: bool = False, ) -> None: content_values = content if isinstance(content, dict) else {} element_type = element.get("type") name = element.get("name") has_value = False value = None if isinstance(name, str): has_value, value = cls._template_v2_content_value( content_values, name, ) if ( has_value and element.get("decorative") is False and element_type in TEMPLATE_V2_GENERATED_ELEMENT_TYPES ): cls._set_template_v2_element_value( element, value, theme=theme, ) return if ( direct_value and not has_value and element.get("decorative") is False and element_type in TEMPLATE_V2_GENERATED_ELEMENT_TYPES ): cls._set_template_v2_element_value( element, content, theme=theme, ) return nested_content = value if isinstance(value, dict) else content_values nested_direct_value = direct_value and not has_value child = element.get("child") if isinstance(child, dict): cls._apply_template_v2_element_content( child, nested_content, theme=theme, direct_value=nested_direct_value, ) children = element.get("children") if isinstance(children, list): if isinstance(value, list) and children: next_children: list[Any] = [] for index, item in enumerate(value): source_child = copy.deepcopy(children[min(index, len(children) - 1)]) if isinstance(source_child, dict): cls._apply_template_v2_element_content( source_child, item, theme=theme, direct_value=True, ) next_children.append(source_child) element["children"] = next_children return for child_element in children: if isinstance(child_element, dict): cls._apply_template_v2_element_content( child_element, nested_content, theme=theme, direct_value=nested_direct_value, ) @staticmethod def _template_v2_content_value( content: dict[str, Any], name: str, ) -> tuple[bool, Any]: for candidate in PresentationChatMemoryLayer._template_v2_content_name_candidates( name ): if candidate in content: return True, content[candidate] return False, None @staticmethod def _template_v2_content_name_candidates(name: str) -> list[str]: without_numeric_token = re.sub(r"_\d+(?=_|$)", "", name) without_prefix = ( without_numeric_token.split("_", 1)[1] if "_" in without_numeric_token else without_numeric_token ) candidates: list[str] = [] for candidate in (name, without_numeric_token, without_prefix): if candidate and candidate not in candidates: candidates.append(candidate) return candidates @classmethod def _set_template_v2_element_value( cls, element: dict[str, Any], value: Any, *, theme: dict[str, Any] | None = None, ) -> None: element_type = element.get("type") if element_type == "text": text = cls._template_v2_text_value(value) if text is None or text == "": return cls._set_template_v2_runs_text(element, text) element["text"] = text return if element_type == "text-list" and isinstance(value, list): source_items = ( element.get("items") if isinstance(element.get("items"), list) else [] ) element["items"] = [ cls._replacement_runs_from_existing( source_items[index] if index < len(source_items) else None, cls._template_v2_text_value(item) or "", element.get("font"), ) for index, item in enumerate(value) ] return if element_type == "image": asset_url = cls._template_v2_asset_url(value) if asset_url: element["data"] = asset_url prompt = cls._template_v2_asset_prompt( value, element.get("is_icon") is True, ) if prompt: element["prompt"] = prompt return if element_type == "chart" and isinstance(value, dict): from services.chat.slide_ui_helpers import _apply_chart_content_update _apply_chart_content_update(element, value, theme) return if element_type == "table": if isinstance(value, dict): cls._set_template_v2_table_content(element, value) elif isinstance(value, list): cls._set_template_v2_table_rows(element, value) @classmethod def _set_template_v2_runs_text(cls, element: dict[str, Any], text: str) -> None: element["runs"] = cls._replacement_runs_from_existing( element.get("runs"), text, element.get("font"), ) @staticmethod def _template_v2_text_value(value: Any) -> str | None: if isinstance(value, str): return value if isinstance(value, (int, float)) and not isinstance(value, bool): return str(value) if isinstance(value, dict): text = value.get("text") if isinstance(text, str): return text if isinstance(text, (int, float)) and not isinstance(text, bool): return str(text) return None @staticmethod def _replacement_runs_from_existing( existing_runs: Any, text: str, fallback_font: Any, ) -> list[dict[str, Any]]: if isinstance(existing_runs, list) and existing_runs: first = existing_runs[0] if isinstance(first, dict): run = copy.deepcopy(first) run["text"] = text return [run] run: dict[str, Any] = {"text": text} if isinstance(fallback_font, dict): run["font"] = copy.deepcopy(fallback_font) return [run] @staticmethod def _template_v2_asset_url(value: Any) -> str | None: if isinstance(value, str): return normalize_slide_asset_url(value) if not isinstance(value, dict): return None fallback_url: str | None = None for key in ( "data", "url", "image_url", "icon_url", "__image_url__", "__icon_url__", ): asset_url = value.get(key) if isinstance(asset_url, str) and asset_url.strip(): normalized_url = normalize_slide_asset_url(asset_url) if normalized_url.strip().startswith( ("http://", "https://", "/app_data/", "/static/", "data:", "blob:") ): return normalized_url if fallback_url is None: fallback_url = normalized_url return fallback_url @staticmethod def _template_v2_asset_prompt(value: Any, is_icon: bool) -> str | None: if not isinstance(value, dict): return None prompt_keys = ( ("icon_query", "__icon_query__", "query", "prompt") if is_icon else ("image_prompt", "__image_prompt__", "prompt", "query") ) for key in prompt_keys: prompt = value.get(key) if isinstance(prompt, str) and prompt.strip(): return prompt return None @classmethod def _set_template_v2_table_content( cls, element: dict[str, Any], value: dict[str, Any], ) -> None: columns = value.get("columns") if isinstance(columns, list): existing_columns = ( element.get("columns") if isinstance(element.get("columns"), list) else [] ) element["columns"] = cls._template_v2_table_cells_from_values( existing_columns, columns, element.get("font"), ) rows = value.get("rows") if isinstance(rows, list): cls._set_template_v2_table_rows(element, rows) @classmethod def _template_v2_table_cells_from_values( cls, existing_cells: list[Any], values: list[Any], fallback_font: Any, ) -> list[dict[str, Any]]: fallback_cell = existing_cells[-1] if existing_cells else None cells: list[dict[str, Any]] = [] for index, value in enumerate(values): existing_cell = ( existing_cells[index] if index < len(existing_cells) else fallback_cell ) existing_runs = ( existing_cell.get("runs") if isinstance(existing_cell, dict) else None ) cell = copy.deepcopy(existing_cell) if isinstance(existing_cell, dict) else {} cell["runs"] = cls._replacement_runs_from_existing( existing_runs, cls._table_cell_text_value(value), cell.get("font") or fallback_font, ) cells.append(cell) return cells @classmethod def _set_template_v2_table_rows( cls, element: dict[str, Any], rows: list[Any], ) -> None: existing_rows = ( element.get("rows") if isinstance(element.get("rows"), list) else [] ) next_rows: list[list[dict[str, Any]]] = [] for row_index, row in enumerate(rows): if not isinstance(row, list): continue existing_row = ( existing_rows[row_index] if row_index < len(existing_rows) and isinstance(existing_rows[row_index], list) else [] ) next_rows.append( cls._template_v2_table_cells_from_values( existing_row, row, element.get("font"), ) ) element["rows"] = next_rows def _validate_slide_content( self, *, content: dict[str, Any], schema: dict[str, Any], ) -> list[str]: validation_content = self._strip_runtime_fields(content) validator = Draft202012Validator(schema) errors = sorted(validator.iter_errors(validation_content), key=lambda err: err.path) if not errors: return [] formatted_errors: list[str] = [] for err in errors[:MAX_SCHEMA_ERRORS]: location = ".".join([str(part) for part in err.path]) or "$" formatted_errors.append(f"{location}: {err.message}") return formatted_errors @staticmethod def _strip_runtime_fields(value: Any) -> Any: if isinstance(value, dict): sanitized: dict[str, Any] = {} for key, nested_value in value.items(): if key in RUNTIME_CONTENT_FIELDS: continue sanitized[key] = PresentationChatMemoryLayer._strip_runtime_fields( nested_value ) return sanitized if isinstance(value, list): return [ PresentationChatMemoryLayer._strip_runtime_fields(item) for item in value ] return value @staticmethod def _extract_speaker_note(content: dict[str, Any]) -> str: value = content.get("__speaker_note__") if isinstance(value, str): return value return "" @staticmethod def _resolve_layout_group( *, presentation: PresentationModel, fallback: str = "presentation", ) -> str: if isinstance(presentation.layout, dict): name = str(presentation.layout.get("name") or "").strip() if name: return name if PresentationChatMemoryLayer._is_template_v2_layout_payload( presentation.layout ): return "template-v2" return fallback @staticmethod def _normalize_outline_slides(outlines: Any) -> list[dict[str, str]]: if not isinstance(outlines, dict): return [] raw_slides = outlines.get("slides") if not isinstance(raw_slides, list): return [] slides: list[dict[str, str]] = [] for raw_slide in raw_slides: raw_content: Any if isinstance(raw_slide, dict): raw_content = raw_slide.get("content", "") else: raw_content = raw_slide if isinstance(raw_content, str): content = raw_content elif raw_content is None: content = "" else: try: content = json.dumps(raw_content, ensure_ascii=False) except Exception: content = str(raw_content) slides.append({"content": normalize_outline_content(content)}) return slides async def _save_outline_slides( self, presentation: PresentationModel, slides: list[dict[str, str]], ) -> None: outline_model = PresentationOutlineModel( slides=[SlideOutlineModel(content=slide["content"]) for slide in slides] ) presentation.outlines = outline_model.model_dump(mode="json") presentation.n_slides = len(outline_model.slides) presentation.title = get_presentation_title_from_presentation_outline( outline_model ) self._sql_session.add(presentation) await self._sql_session.commit() await MEM0_PRESENTATION_MEMORY_SERVICE.store_generated_outlines( presentation.id, presentation.outlines, ) @staticmethod def _extract_outline_title(markdown_content: str) -> str: for line in markdown_content.splitlines(): stripped = line.strip() if not stripped: continue heading_match = re.match(r"^#{1,6}\s*(.+?)\s*$", stripped) if heading_match: return heading_match.group(1).strip() return stripped[:120] return "Untitled outline" @staticmethod def _serialize_slide(slide: SlideModel) -> str: content_text = "" try: content_text = json.dumps(slide.content or {}, ensure_ascii=False) except Exception: content_text = str(slide.content) speaker_note = slide.speaker_note or "" return f"slide_index={slide.index}\nlayout_id={slide.layout}\n{content_text}\n{speaker_note}" @staticmethod def _build_snippet(text: str, query_lower: str, window: int = 320) -> str: normalized = " ".join(text.split()) if not normalized: return "" offset = normalized.lower().find(query_lower) if offset == -1: return normalized[:window] start = max(0, offset - window // 3) end = min(len(normalized), start + window) return normalized[start:end] @staticmethod def _trim_document_text(text: str, limit: int) -> str: normalized = (text or "").strip() if not normalized: return "" if len(normalized) <= limit: return normalized return f"{normalized[:limit].rstrip()}\n[Document content truncated]" @staticmethod def _extract_theme_colors(theme: dict[str, Any]) -> dict[str, Any]: data = theme.get("data") colors = data.get("colors") if isinstance(data, dict) else None if isinstance(colors, dict): return copy.deepcopy(colors) colors = theme.get("colors") return copy.deepcopy(colors) if isinstance(colors, dict) else {} @staticmethod def _chart_palette_from_theme_colors(colors: dict[str, Any]) -> list[str]: palette: list[str] = [] for index in range(10): value = colors.get(f"graph_{index}") if not isinstance(value, str): continue color = PresentationChatMemoryLayer._normalize_hex_color(value) if color: palette.append(color) if palette: return palette fallback_keys = ("primary", "card", "stroke", "background_text", "primary_text") fallback_palette: list[str] = [] for key in fallback_keys: value = colors.get(key) if not isinstance(value, str): continue color = PresentationChatMemoryLayer._normalize_hex_color(value) if color: fallback_palette.append(color) return fallback_palette async def _get_chat_available_themes(self) -> list[dict[str, Any]]: merged_themes: list[dict[str, Any]] = [copy.deepcopy(theme) for theme in CHAT_BUILTIN_THEMES] row = await self._sql_session.scalar( select(KeyValueSqlModel).where(KeyValueSqlModel.key == THEMES_STORAGE_KEY) ) if not row or not isinstance(row.value, dict): return merged_themes custom_themes = row.value.get("themes") if not isinstance(custom_themes, list): return merged_themes existing_ids = { str(theme.get("id") or "").strip().lower() for theme in merged_themes } for custom_theme in custom_themes: if not isinstance(custom_theme, dict): continue theme_data = custom_theme.get("data") colors = theme_data.get("colors") if isinstance(theme_data, dict) else None if not isinstance(colors, dict) or "background" not in colors: continue custom_theme_copy = copy.deepcopy(custom_theme) custom_theme_copy.setdefault("user", "local") theme_id = str(custom_theme_copy.get("id") or "").strip().lower() if theme_id and theme_id in existing_ids: continue if theme_id: existing_ids.add(theme_id) merged_themes.append(custom_theme_copy) return merged_themes async def _upsert_custom_theme_in_store(self, theme: dict[str, Any]) -> None: row = await self._sql_session.scalar( select(KeyValueSqlModel).where(KeyValueSqlModel.key == THEMES_STORAGE_KEY) ) themes: list[dict[str, Any]] = [] if row and isinstance(row.value, dict): raw_themes = row.value.get("themes") if isinstance(raw_themes, list): themes = copy.deepcopy(raw_themes) theme_id = str(theme.get("id") or "").strip().lower() replaced = False if theme_id: for idx, existing_theme in enumerate(themes): existing_id = str(existing_theme.get("id") or "").strip().lower() if existing_id == theme_id: themes[idx] = copy.deepcopy(theme) replaced = True break if not replaced: themes.append(copy.deepcopy(theme)) if row: row.value = {"themes": themes} self._sql_session.add(row) return self._sql_session.add(KeyValueSqlModel(key=THEMES_STORAGE_KEY, value={"themes": themes})) @staticmethod def _resolve_base_theme_for_customization( current_theme: dict[str, Any] | None, available_themes: list[dict[str, Any]], ) -> dict[str, Any]: if isinstance(current_theme, dict): data = current_theme.get("data") colors = data.get("colors") if isinstance(data, dict) else None if isinstance(colors, dict): return copy.deepcopy(current_theme) preferred_base = PresentationChatMemoryLayer._find_theme_by_id( available_themes, "professional-blue" ) if preferred_base: return copy.deepcopy(preferred_base) if available_themes: return copy.deepcopy(available_themes[0]) return { "id": "professional-blue", "name": "Professional Blue", "description": "Fallback base theme.", "user": "system", "logo": None, "logo_url": None, "company_name": None, "data": { "colors": { "primary": "#161616", "background": "#ffffff", "card": "#dae6ff", "stroke": "#d1d1d1", "primary_text": "#eeeaea", "background_text": "#000000", "graph_0": "#2e2e2e", "graph_1": "#424242", "graph_2": "#585858", "graph_3": "#6f6f6f", "graph_4": "#868686", "graph_5": "#9e9e9e", "graph_6": "#b7b7b7", "graph_7": "#d1d1d1", "graph_8": "#e8e8e8", "graph_9": "#f5f5f5", }, "fonts": {"textFont": DEFAULT_THEME_FONT}, }, } @staticmethod def _build_custom_theme_from_payload( *, custom_theme: dict[str, Any], requested_theme: str, current_theme: dict[str, Any] | None, available_themes: list[dict[str, Any]], ) -> dict[str, Any] | None: base_theme = PresentationChatMemoryLayer._resolve_base_theme_for_customization( current_theme, available_themes ) payload = copy.deepcopy(custom_theme) data_block = payload.get("data") if isinstance(payload.get("data"), dict) else payload if not isinstance(data_block, dict): return None colors_override = PresentationChatMemoryLayer._extract_colors_from_payload(data_block) if not colors_override: return None merged_colors = PresentationChatMemoryLayer._merge_theme_colors( base_theme=base_theme, color_overrides=colors_override, ) if not merged_colors: return None text_font = PresentationChatMemoryLayer._extract_text_font_from_payload( data_block, base_theme ) if not text_font: return None name_candidates = [ payload.get("name"), data_block.get("name"), requested_theme, "Custom Theme", ] theme_name = next( ( str(candidate).strip() for candidate in name_candidates if isinstance(candidate, str) and str(candidate).strip() ), "Custom Theme", ) theme_id = PresentationChatMemoryLayer._sanitize_theme_id( str(payload.get("id") or "") ) if not theme_id: theme_id = PresentationChatMemoryLayer._sanitize_theme_id(theme_name) if not theme_id: theme_id = f"chat-custom-{uuid.uuid4().hex[:8]}" description = str( payload.get("description") or data_block.get("description") or f"Custom theme generated from chat request: {theme_name}" ).strip() theme_data = payload.get("data") final_data = copy.deepcopy(theme_data) if isinstance(theme_data, dict) else {} final_data["colors"] = merged_colors final_data["fonts"] = {"textFont": text_font} return { "id": theme_id, "name": theme_name, "description": description, "user": str(payload.get("user") or "local"), "logo": payload.get("logo"), "logo_url": payload.get("logo_url"), "company_name": payload.get("company_name"), "data": final_data, } @staticmethod def _extract_colors_from_payload(data_block: dict[str, Any]) -> dict[str, str]: raw_colors = data_block.get("colors") if not isinstance(raw_colors, dict): return {} normalized_colors: dict[str, str] = {} for key in THEME_COLOR_KEYS: value = raw_colors.get(key) if not isinstance(value, str): continue normalized_hex = PresentationChatMemoryLayer._normalize_hex_color(value) if normalized_hex: normalized_colors[key] = normalized_hex return normalized_colors @staticmethod def _merge_theme_colors( *, base_theme: dict[str, Any], color_overrides: dict[str, str], ) -> dict[str, str] | None: data = base_theme.get("data") base_colors = data.get("colors") if isinstance(data, dict) else None if not isinstance(base_colors, dict): return None merged: dict[str, str] = {} for key in THEME_COLOR_KEYS: override = color_overrides.get(key) if override: merged[key] = override continue base_value = base_colors.get(key) if isinstance(base_value, str): normalized = PresentationChatMemoryLayer._normalize_hex_color(base_value) merged[key] = normalized or base_value continue # Keep resulting theme always complete for frontend variable mapping. merged[key] = "#000000" return merged @staticmethod def _extract_text_font_from_payload( data_block: dict[str, Any], base_theme: dict[str, Any], ) -> dict[str, str] | None: candidate: dict[str, Any] | None = None fonts = data_block.get("fonts") if isinstance(fonts, dict): text_font = fonts.get("textFont") if isinstance(text_font, dict): candidate = text_font if not candidate: text_font = data_block.get("textFont") if isinstance(text_font, dict): candidate = text_font if not candidate: base_data = base_theme.get("data") base_fonts = base_data.get("fonts") if isinstance(base_data, dict) else None base_text_font = base_fonts.get("textFont") if isinstance(base_fonts, dict) else None if isinstance(base_text_font, dict): candidate = base_text_font if not candidate: candidate = DEFAULT_THEME_FONT name = candidate.get("name") url = candidate.get("url") if not isinstance(name, str) or not name.strip(): return None if not isinstance(url, str) or not url.strip(): return None return {"name": name.strip(), "url": url.strip()} @staticmethod def _sanitize_theme_id(value: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-") return slug[:64] @staticmethod def _normalize_hex_color(value: str) -> str | None: normalized = value.strip().lower() if not normalized: return None if normalized.startswith("#"): normalized = normalized[1:] if len(normalized) == 3: expanded = "".join(ch * 2 for ch in normalized) if re.fullmatch(r"[0-9a-f]{6}", expanded): return f"#{expanded}" return None if len(normalized) != 6: return None if not re.fullmatch(r"[0-9a-f]{6}", normalized): return None return f"#{normalized}" @staticmethod def _select_theme_for_query( requested_theme: str, available_themes: list[dict[str, Any]], current_theme: dict[str, Any] | None, ) -> dict[str, Any] | None: normalized_query = requested_theme.strip().lower() if not normalized_query: return None # Direct exact match by id or name. for theme in available_themes: theme_id = str(theme.get("id") or "").strip().lower() theme_name = str(theme.get("name") or "").strip().lower() if normalized_query in {theme_id, theme_name}: return theme current_theme_id = str((current_theme or {}).get("id") or "").strip().lower() query_tokens = [token for token in re.split(r"[\s_-]+", normalized_query) if token] if "dark" in query_tokens or any( token in normalized_query for token in ("night", "black") ): for preferred in ("professional-dark", "edge-yellow"): theme = PresentationChatMemoryLayer._find_theme_by_id( available_themes, preferred ) if theme: return theme if "light" in query_tokens or any( token in normalized_query for token in ("bright", "white") ): for preferred in ("professional-blue", "mint-blue", "light-rose"): theme = PresentationChatMemoryLayer._find_theme_by_id( available_themes, preferred ) if theme: return theme if any(token in normalized_query for token in ("another", "different", "change")): opposite = ( not PresentationChatMemoryLayer._is_dark_theme(current_theme) if current_theme else True ) candidates = [ theme for theme in available_themes if str(theme.get("id") or "").strip().lower() != current_theme_id ] for theme in candidates: if PresentationChatMemoryLayer._is_dark_theme(theme) == opposite: return theme if candidates: return candidates[0] # Fuzzy contains match over id/name/description. for theme in available_themes: haystack = " ".join( [ str(theme.get("id") or "").strip().lower(), str(theme.get("name") or "").strip().lower(), str(theme.get("description") or "").strip().lower(), ] ) if normalized_query in haystack: return theme if query_tokens and all(token in haystack for token in query_tokens): return theme return None @staticmethod def _find_theme_by_id( themes: list[dict[str, Any]], theme_id: str ) -> dict[str, Any] | None: normalized_theme_id = theme_id.strip().lower() for theme in themes: current_id = str(theme.get("id") or "").strip().lower() if current_id == normalized_theme_id: return theme return None @staticmethod def _extract_theme_name(theme: dict[str, Any] | None) -> str | None: if not isinstance(theme, dict): return None name = theme.get("name") if isinstance(name, str) and name.strip(): return name.strip() theme_id = theme.get("id") if isinstance(theme_id, str) and theme_id.strip(): return theme_id.strip() return None @staticmethod def _is_dark_theme(theme: dict[str, Any] | None) -> bool: if not isinstance(theme, dict): return False data = theme.get("data") if not isinstance(data, dict): return False colors = data.get("colors") if not isinstance(colors, dict): return False background = colors.get("background") if not isinstance(background, str): return False return PresentationChatMemoryLayer._is_dark_hex(background) @staticmethod def _is_dark_hex(hex_color: str) -> bool: normalized = hex_color.strip().lstrip("#") if len(normalized) != 6: return False try: red = int(normalized[0:2], 16) green = int(normalized[2:4], 16) blue = int(normalized[4:6], 16) except ValueError: return False # Relative luminance approximation. luma = (0.299 * red + 0.587 * green + 0.114 * blue) / 255 return luma < 0.5