"""Create editable content JSON schemas from template v2 slide layouts.""" from __future__ import annotations import copy import re from typing import Any from .models.layouts import RawSlideLayout CONTENT_TYPES = {"text", "image", "text-list", "table", "chart"} REPEATED_NAME_SUFFIX_RE = re.compile(r"_\d+$") JSON_SCHEMA_URI = "https://json-schema.org/draft/2020-12/schema" COMPONENT_REPEATED_NAME_TOKEN_RE = re.compile(r"_\d+(?=_|$)") COMPONENT_SCHEMA_METADATA_KEYS = { "$schema", "title", "description", "x-element-type", "x-element-path", } def _is_editable_element(element: dict[str, Any]) -> bool: return element.get("decorative") is False def extract_slide_schema_from_layout(layout: RawSlideLayout) -> dict[str, Any]: """ Take slide layout and return content schema from slide layout. """ return _object_schema(_properties_schema(layout.elements)) def get_component_schema(component: Any | dict[str, Any]) -> dict[str, Any] | None: """ Return an editable content schema for a generated template component. """ component_data = _component_data(component) elements = component_data.get("elements") if not isinstance(elements, list): raise ValueError("component must contain an elements array") properties = _component_schema_properties(elements) if not properties: return None return { "$schema": JSON_SCHEMA_URI, "type": "object", "title": component_data.get("id", "component_content"), "description": component_data.get("description"), "additionalProperties": False, "properties": properties, "required": list(properties), } def get_template_schema( template_json: Any | dict[str, Any], *, source_file: str = "template.json", ) -> dict[str, Any]: """ Return editable content schemas for component-based template layouts. """ template_data = _template_data(template_json) layouts = template_data.get("layouts") if not isinstance(layouts, list): raise ValueError("template JSON must contain a layouts array") generated_layouts = [ _template_layout_schema(layout, index) for index, layout in enumerate(layouts, start=1) if isinstance(layout, dict) ] return { "source_file": source_file, "layout_count": len(generated_layouts), "layouts": generated_layouts, } def _properties_schema(elements: list[Any]) -> dict[str, Any]: properties: dict[str, Any] = {} for name, schema in _nodes_for_elements(elements): _add_property(properties, name, schema) return properties def _nodes_for_elements(elements: list[Any]) -> list[tuple[str, dict[str, Any]]]: nodes: list[tuple[str, dict[str, Any]]] = [] for value in elements: node = _node_for_element_value(value) if node is not None: nodes.append(node) return nodes def _node_for_element_value(value: Any) -> tuple[str, dict[str, Any]] | None: element = _element_dict(value) if element is None: return None return _node_for_element(element) def _node_for_element(element: dict[str, Any]) -> tuple[str, dict[str, Any]] | None: element_type = element.get("type") if element_type == "container": return _node_for_element_value(element.get("child")) if element_type in {"flex", "grid", "group"}: children = element.get("children", []) if not isinstance(children, list): return None nodes = _nodes_for_elements(children) if not nodes: return None name = _element_name(element) if name is None: return None if element_type in {"flex", "grid"}: array_schema = _array_schema_for_repeated_children(element, children, nodes) if array_schema is not None: return name, array_schema properties: dict[str, Any] = {} for child_name, child_schema in nodes: _add_property(properties, child_name, child_schema) schema = _object_schema(properties) if element_type in {"flex", "grid"}: schema.update( _compact( { "minProperties": element.get("min_children"), "maxProperties": element.get("max_children"), } ) ) return name, schema if element_type not in CONTENT_TYPES or not _is_editable_element(element): return None name = _element_name(element) if name is None: return None return name, _content_schema_for_element(element) def _content_schema_for_element(element: dict[str, Any]) -> dict[str, Any]: element_type = element["type"] if element_type == "text": return _compact( { "type": "string", "minLength": element.get("min_length"), "maxLength": element.get("max_length"), } ) if element_type == "image": key = "query" if element.get("is_icon") is True else "prompt" return _object_schema({key: {"type": "string"}}) if element_type == "text-list": return _compact( { "type": "array", "minItems": element.get("min_items"), "maxItems": element.get("max_items"), "items": _compact( { "type": "string", "minLength": element.get("min_item_length"), "maxLength": element.get("max_item_length"), } ), } ) if element_type == "table": return _compact( { "type": "array", "minItems": element.get("min_rows"), "maxItems": element.get("max_rows"), "items": _compact( { "type": "array", "minItems": element.get("min_columns"), "maxItems": element.get("max_columns"), "items": {"type": "string"}, } ), } ) if element_type == "chart": return _chart_content_schema() raise ValueError(f"unsupported content element type: {element_type}") def _array_schema_for_repeated_children( element: dict[str, Any], children: list[Any], nodes: list[tuple[str, dict[str, Any]]], ) -> dict[str, Any] | None: if len(nodes) < 2 or len(nodes) != _element_count(children): return None item_schemas = [ _schema_without_repeated_name_suffix(schema, _repeated_name_suffix(name)) for name, schema in nodes ] first_schema = item_schemas[0] if any(schema != first_schema for schema in item_schemas[1:]): return None return _compact( { "type": "array", "minItems": element.get("min_children"), "maxItems": element.get("max_children"), "items": first_schema, } ) def _object_schema( properties: dict[str, Any], *, required: list[str] | None = None, ) -> dict[str, Any]: return { "type": "object", "properties": properties, "required": list(properties) if required is None else required, "additionalProperties": False, } def _element_name(element: dict[str, Any]) -> str | None: name = element.get("name") if not isinstance(name, str): return None stripped = name.strip() return stripped or None def _add_property( properties: dict[str, Any], name: str, schema: dict[str, Any], ) -> None: key = name suffix = 2 while key in properties: key = f"{name}_{suffix}" suffix += 1 properties[key] = schema def _schema_without_repeated_name_suffix( schema: dict[str, Any], suffix: str | None, ) -> dict[str, Any]: if schema.get("type") != "object": return { key: _normalize_schema_value(value, suffix) for key, value in schema.items() } properties = schema.get("properties") if not isinstance(properties, dict): return { key: _normalize_schema_value(value, suffix) for key, value in schema.items() } normalized_properties: dict[str, Any] = {} name_map: dict[str, str] = {} for key, value in properties.items(): normalized_key = _strip_repeated_suffix(key, suffix) name_map[key] = normalized_key normalized_properties[normalized_key] = _normalize_schema_value(value, suffix) normalized_schema = { key: _normalize_schema_value(value, suffix) for key, value in schema.items() if key not in {"properties", "required"} } normalized_schema["properties"] = normalized_properties required = schema.get("required") if isinstance(required, list): normalized_schema["required"] = [ name_map.get(item, _strip_repeated_suffix(item, suffix)) for item in required if isinstance(item, str) ] return normalized_schema def _normalize_schema_value(value: Any, suffix: str | None) -> Any: if isinstance(value, dict): return _schema_without_repeated_name_suffix(value, suffix) if isinstance(value, list): return [_normalize_schema_value(item, suffix) for item in value] return value def _repeated_name_suffix(value: str) -> str | None: match = REPEATED_NAME_SUFFIX_RE.search(value) return match.group(0) if match else None def _strip_repeated_suffix(value: str, suffix: str | None) -> str: if suffix and value.endswith(suffix): return value[: -len(suffix)] return value def _element_count(values: list[Any]) -> int: return sum(1 for value in values if _element_dict(value) is not None) def _compact(value: dict[str, Any]) -> dict[str, Any]: return {key: item for key, item in value.items() if item is not None} def _element_dict(value: Any) -> dict[str, Any] | None: if isinstance(value, dict): return value model_dump = getattr(value, "model_dump", None) if not callable(model_dump): return None dumped = model_dump(mode="json") if isinstance(dumped, dict): return dumped return None def _template_layout_schema(layout: dict[str, Any], slide_index: int) -> dict[str, Any]: properties: dict[str, Any] = {} required: list[str] = [] components = layout.get("components") if not isinstance(components, list): components = [] component_entries: list[tuple[str, dict[str, Any]]] = [] for component in components: component_data = _component_data_or_none(component) if component_data is None: continue component_schema = get_component_schema(component_data) if component_schema is None: continue component_entries.append((_component_id(component_data), component_schema)) component_counts: dict[str, int] = {} for component_id, _schema in component_entries: component_counts[component_id] = component_counts.get(component_id, 0) + 1 component_indexes: dict[str, int] = {} for component_id, component_schema in component_entries: component_index = component_indexes.get(component_id, 0) component_indexes[component_id] = component_index + 1 key = _template_component_key( component_id, occurrence_index=component_index, occurrence_count=component_counts[component_id], properties=properties, ) properties[key] = _component_schema_for_template(component_schema) required.append(key) schema = None if properties: schema = { "$schema": JSON_SCHEMA_URI, "type": "object", "title": layout.get("id") or f"slide_{slide_index}", "description": layout.get("description"), "additionalProperties": False, "properties": properties, "required": required, } return { "slide": slide_index, "layout_id": layout.get("id"), "schema": schema, } def _component_schema_for_template(component_schema: dict[str, Any]) -> dict[str, Any]: schema = _strip_component_schema_metadata(copy.deepcopy(component_schema)) return schema if isinstance(schema, dict) else component_schema def _strip_component_schema_metadata(value: Any) -> Any: if isinstance(value, list): return [_strip_component_schema_metadata(item) for item in value] if not isinstance(value, dict): return value stripped: dict[str, Any] = {} for key, nested in value.items(): if key in COMPONENT_SCHEMA_METADATA_KEYS: continue if key == "properties" and isinstance(nested, dict): stripped[key] = { property_name: _strip_component_schema_metadata(property_schema) for property_name, property_schema in nested.items() } continue stripped[key] = _strip_component_schema_metadata(nested) return stripped def _template_component_key( component_id: str, *, occurrence_index: int, occurrence_count: int, properties: dict[str, Any], ) -> str: key = ( f"{component_id}_{occurrence_index}" if occurrence_count > 1 else component_id ) suffix = 1 unique_key = key while unique_key in properties: unique_key = f"{key}_{suffix}" suffix += 1 return unique_key def _component_data(component: Any | dict[str, Any]) -> dict[str, Any]: component_data = _component_data_or_none(component) if component_data is not None: return component_data raise ValueError("component must be a Component or JSON object") def _component_data_or_none(value: Any) -> dict[str, Any] | None: if isinstance(value, dict): return copy.deepcopy(value) model_dump = getattr(value, "model_dump", None) if callable(model_dump): dumped = model_dump(mode="json") if isinstance(dumped, dict): return dumped return None def _template_data(template_json: Any | dict[str, Any]) -> dict[str, Any]: if isinstance(template_json, dict): return copy.deepcopy(template_json) model_dump = getattr(template_json, "model_dump", None) if callable(model_dump): dumped = model_dump(mode="json") if isinstance(dumped, dict): return dumped raise ValueError("template JSON must be a JSON object") def _component_id(component_data: dict[str, Any]) -> str: component_id = component_data.get("id") if isinstance(component_id, str): return component_id raise ValueError("component must include a string id") def _component_schema_properties(elements: list[Any]) -> dict[str, Any]: properties: dict[str, Any] = {} for name, schema in _component_schema_nodes_for_elements(elements): _component_add_schema_property(properties, name, schema) return properties def _component_schema_nodes_for_elements( elements: list[Any], *, path: str = "elements", ) -> list[tuple[str, dict[str, Any]]]: nodes: list[tuple[str, dict[str, Any]]] = [] for index, element in enumerate(elements): if isinstance(element, dict): nodes.extend( _component_schema_nodes_for_element( element, path=f"{path}.{index}", ) ) return nodes def _component_schema_nodes_for_element( element: dict[str, Any], *, path: str, ) -> list[tuple[str, dict[str, Any]]]: element_type = element.get("type") name = _component_schema_element_name(element) if ( element_type in CONTENT_TYPES and _is_editable_element(element) and name is not None ): return [ ( name, _component_content_field_schema( {"name": name, "path": path, "element": element} ), ) ] if element_type == "container": child = element.get("child") child_nodes = ( _component_schema_nodes_for_element(child, path=f"{path}.child") if isinstance(child, dict) else [] ) if name is None or not child_nodes: return child_nodes return [(name, _component_object_schema_from_nodes(child_nodes))] if element_type in {"flex", "grid", "group"}: children = element.get("children") if not isinstance(children, list): return [] child_node_sets = [ _component_schema_nodes_for_element(child, path=f"{path}.children.{index}") if isinstance(child, dict) else [] for index, child in enumerate(children) ] child_nodes = [ node for node_set in child_node_sets for node in node_set ] if name is None or not child_nodes: return child_nodes if element_type in {"flex", "grid"}: array_schema = _component_array_schema_for_repeated_children( element, child_node_sets, ) if array_schema is not None: return [(name, array_schema)] return [(name, _component_object_schema_from_nodes(child_nodes))] return [] def _component_schema_element_name(element: dict[str, Any]) -> str | None: name = element.get("name") if not isinstance(name, str): return None stripped = name.strip() return stripped or None def _component_object_schema_from_nodes( nodes: list[tuple[str, dict[str, Any]]], ) -> dict[str, Any]: properties: dict[str, Any] = {} for name, schema in nodes: _component_add_schema_property(properties, name, schema) return { "type": "object", "additionalProperties": False, "properties": properties, "required": list(properties), } def _component_add_schema_property( properties: dict[str, Any], name: str, schema: dict[str, Any], ) -> None: key = name suffix = 2 while key in properties: key = f"{name}_{suffix}" suffix += 1 properties[key] = schema def _component_array_schema_for_repeated_children( element: dict[str, Any], child_node_sets: list[list[tuple[str, dict[str, Any]]]], ) -> dict[str, Any] | None: populated_node_sets = [node_set for node_set in child_node_sets if node_set] if len(populated_node_sets) < 2 or len(populated_node_sets) != len(child_node_sets): return None for strategy in ("numeric", "none", "prefix"): normalized_item_schemas = [ _component_normalized_repeated_item_schema(node_set, strategy=strategy) for node_set in populated_node_sets ] merged_item_schema = _component_merge_repeated_schemas( normalized_item_schemas ) if merged_item_schema is not None: return _without_none_values( { "type": "array", "minItems": element.get("min_children"), "maxItems": element.get("max_children"), "items": merged_item_schema, } ) return None def _component_normalized_repeated_item_schema( nodes: list[tuple[str, dict[str, Any]]], *, strategy: str, ) -> dict[str, Any]: token = _component_normalization_token_for_nodes(nodes, strategy=strategy) item_schema = ( nodes[0][1] if len(nodes) == 1 and nodes[0][1].get("type") == "object" else _component_object_schema_from_nodes(nodes) ) return _component_schema_without_repeated_name_suffix(item_schema, token) def _component_normalization_token_for_nodes( nodes: list[tuple[str, dict[str, Any]]], *, strategy: str, ) -> str | None: if strategy == "none": return None token_getter = ( _component_numeric_name_token if strategy == "numeric" else _component_prefix_name_token ) tokens = [token_getter(name) for name, _schema in nodes] tokens = [token for token in tokens if token is not None] if not tokens: return None first_token = tokens[0] if all(token == first_token for token in tokens): return first_token return None def _component_numeric_name_token(value: str) -> str | None: match = COMPONENT_REPEATED_NAME_TOKEN_RE.search(value) return match.group(0) if match else None def _component_prefix_name_token(value: str) -> str | None: token, separator, _rest = value.partition("_") if not separator or not token: return None return f"{token}_" def _component_schema_without_repeated_name_suffix( schema: dict[str, Any], suffix: str | None, ) -> dict[str, Any]: normalized = _component_normalize_schema_value(schema, suffix) return normalized if isinstance(normalized, dict) else schema def _component_normalize_schema_value(value: Any, suffix: str | None) -> Any: if isinstance(value, list): return [_component_normalize_schema_value(item, suffix) for item in value] if not isinstance(value, dict): return value normalized: dict[str, Any] = {} for key, nested in value.items(): if key == "x-element-path": continue if key == "properties" and isinstance(nested, dict): properties: dict[str, Any] = {} for property_name, property_schema in nested.items(): normalized_name = _component_strip_repeated_suffix( property_name, suffix, ) normalized_schema = _component_normalize_schema_value( property_schema, suffix, ) if isinstance(normalized_schema, dict) and "title" in normalized_schema: normalized_schema["title"] = _component_content_field_title( normalized_name ) properties[normalized_name] = normalized_schema normalized[key] = properties continue if key == "required" and isinstance(nested, list): normalized[key] = [ _component_strip_repeated_suffix(item, suffix) for item in nested if isinstance(item, str) ] continue normalized[key] = _component_normalize_schema_value(nested, suffix) return normalized def _component_strip_repeated_suffix(value: str, suffix: str | None) -> str: if suffix and suffix in value: return value.replace(suffix, "", 1) return value def _component_merge_repeated_schemas( schemas: list[dict[str, Any]], ) -> dict[str, Any] | None: if not schemas: return None schema_types = [schema.get("type") for schema in schemas] first_type = schema_types[0] if any(schema_type != first_type for schema_type in schema_types): return None if first_type == "object": property_sets = [ schema.get("properties") for schema in schemas ] if not all(isinstance(properties, dict) for properties in property_sets): return None property_keys = [set(properties) for properties in property_sets] first_keys = property_keys[0] if any(keys != first_keys for keys in property_keys): return None merged_properties: dict[str, Any] = {} for key in sorted(first_keys): merged_property = _component_merge_repeated_schemas( [properties[key] for properties in property_sets] ) if merged_property is None: return None merged_properties[key] = merged_property return { "type": "object", "additionalProperties": False, "properties": merged_properties, "required": [ key for key in schemas[0].get("required", list(merged_properties)) if key in merged_properties ], } if first_type == "array": item_schemas = [ schema.get("items") for schema in schemas if isinstance(schema.get("items"), dict) ] if len(item_schemas) != len(schemas): return None merged_items = _component_merge_repeated_schemas(item_schemas) if merged_items is None: return None return _without_none_values( { "type": "array", "minItems": _component_min_numeric_schema_value(schemas, "minItems"), "maxItems": _component_max_numeric_schema_value(schemas, "maxItems"), "items": merged_items, } ) merged = copy.deepcopy(schemas[0]) merged.pop("x-element-path", None) if first_type == "string": min_length = _component_min_numeric_schema_value(schemas, "minLength") max_length = _component_max_numeric_schema_value(schemas, "maxLength") if min_length is not None: merged["minLength"] = min_length if max_length is not None: merged["maxLength"] = max_length return _without_none_values(merged) def _component_min_numeric_schema_value( schemas: list[dict[str, Any]], key: str, ) -> Any: values = [ schema.get(key) for schema in schemas if isinstance(schema.get(key), (int, float)) ] return min(values) if values else None def _component_max_numeric_schema_value( schemas: list[dict[str, Any]], key: str, ) -> Any: values = [ schema.get(key) for schema in schemas if isinstance(schema.get(key), (int, float)) ] return max(values) if values else None def _component_content_field_schema(field: dict[str, Any]) -> dict[str, Any]: element = field["element"] element_type = element.get("type") schema: dict[str, Any] if element_type == "text": schema = { "type": "string", "minLength": element.get("min_length"), "maxLength": element.get("max_length"), } elif element_type == "image": prompt_key = _component_image_prompt_key(element) schema = { "type": "object", "additionalProperties": False, "properties": { prompt_key: { "type": "string", "description": _component_image_prompt_description(element), } }, "required": [prompt_key], } elif element_type == "text-list": schema = { "type": "array", "items": { "type": "string", "minLength": element.get("min_item_length"), "maxLength": element.get("max_item_length"), }, "minItems": element.get("min_items"), "maxItems": element.get("max_items"), } elif element_type == "table": schema = { "type": "object", "additionalProperties": False, "properties": { "columns": { "type": "array", "items": {"type": "string"}, "minItems": element.get("min_columns"), "maxItems": element.get("max_columns"), }, "rows": { "type": "array", "items": { "type": "array", "items": {"type": "string"}, "minItems": element.get("min_columns"), "maxItems": element.get("max_columns"), }, "minItems": element.get("min_rows"), "maxItems": element.get("max_rows"), }, }, "required": ["columns", "rows"], } elif element_type == "chart": schema = _chart_content_schema() else: schema = {} return { **_without_none_values(schema), "title": _component_content_field_title(field["name"]), "x-element-type": element_type, "x-element-path": field["path"], } def _without_none_values(value: Any) -> Any: if isinstance(value, dict): return { key: _without_none_values(nested) for key, nested in value.items() if nested is not None } if isinstance(value, list): return [_without_none_values(item) for item in value] return value def _chart_content_schema() -> dict[str, Any]: return { "type": "object", "additionalProperties": False, "properties": { "title": {"type": ["string", "null"]}, "title_color": {"type": ["string", "null"]}, "categories": { "type": "array", "items": {"type": "string"}, "maxItems": 24, }, "series": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "name": {"type": "string"}, "values": { "type": "array", "items": {"type": "number"}, "maxItems": 24, }, }, "required": ["name", "values"], }, "maxItems": 12, }, }, "required": ["categories", "series"], } def _component_content_field_title(name: str) -> str: return " ".join(part.capitalize() for part in name.split("_") if part) or name def _component_image_prompt_key(element: dict[str, Any]) -> str: return "icon_query" if element.get("is_icon") is True else "image_prompt" def _component_image_prompt_description(element: dict[str, Any]) -> str: if element.get("is_icon") is True: return "Search query for the replacement icon." return "Prompt for the replacement image."