Files
researchowl/src/generator/spec_contract.py
T
ChemaVXandClaude Opus 5 6e4b3e1379
Build & Deploy ResearchOwl / build-and-push (push) Successful in 8s
fix(short): dos citas en el campo de una son una frase que nadie dijo
El comprobador de fundamento une las líneas de `quote` antes de buscarlas, y eso
cerró la forma con la que falló Socorro en agosto (`366ded1`): `“LIKE ALUMINUM` +
`SMOOTH, NO WINDOWS”` unidas son una sola frase, no aparece en ninguna fuente, y
se rechaza.

Pero la unión se derrota poniéndole a cada línea su propio par de comillas.
Entonces son DOS citas, cada una fundamentada por su lado, y pasa en silencio
mientras el fotograma dibuja la frase de nadie. Medido sobre los cinco Shorts
generados hoy: le pasaba a dos.

Se comprueba la FORMA, no el contenido — dos aperturas son dos citas, diga lo
que diga la fuente — y se comprueba en `validate_spec`, o sea que cuesta un
reintento y no una revisión humana. Es regla del canal, no de shortsmith: allí
renderiza igual. Va como error duro y no como nota editorial porque el fallo que
evita es el peor que tiene este sistema: una cita fabricada con material
auténtico y atribuida a alguien con nombre y apellidos.

La comprobación es universal por nombre de prop, no por plantilla: una plantilla
nueva con un `quote` de varias líneas queda cubierta sin tocar nada.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-13 21:47:07 +00:00

698 lines
30 KiB
Python

"""El contrato del spec, leído — no copiado — de shortsmith.
Dos cosas, las dos guiadas por lo que publica `GET /templates`:
* `describe_templates()` — el contrato en prosa compacta, para meterlo en el
prompt. Añadir una plantilla en shortsmith la deja descrita aquí sola.
* `validate_spec()` — validación local ANTES de renderizar, con las mismas
rutas de error que devolvería el servidor
(`shots.0.radar_sweep.props.sweeeps`). Hace falta que sea local porque el
comprobador de fundamento va entre la validación y el render: mandar el spec
a `POST /render` para validarlo ya encolaría el render.
La mitad de props del contrato NO vive aquí: se valida contra el esquema
recibido. Lo único escrito a mano es el sobre (version/meta/audio/shots), que
es pequeño, estable, y está anotado con la regla equivalente de
`shortsmith/src/shortsmith/spec.py`. Las reglas de pydantic que cruzan campos
(los límites de MapBounds, "3 barras no dejan sitio para una cita") NO se
replican: las coge el 422 del servidor al enviar, y ese camino también está
cubierto.
"""
from __future__ import annotations
import re
from typing import Any, Iterable, Optional
__all__ = [
"SpecInvalid",
"validate_spec",
"editorial_notes",
"estimated_duration",
"spoken_seconds",
"sentence_count",
"teachable_seconds",
"max_words_in",
"describe_templates",
"TARGET_MIN_DURATION",
"TARGET_MAX_DURATION",
]
# Límites del sobre — espejo de shortsmith/spec.py.
RESOLUTIONS = {(1080, 1920), (1920, 1080)}
FPS_VALUES = {24, 25, 30, 60}
MIN_SHOT_DURATION = 0.5
MIN_TOTAL_DURATION = 5.0
MAX_TOTAL_DURATION = 180.0 # límite duro de YouTube Shorts
MAX_SHOTS = 64
META_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
#: El objetivo editorial, que NO es el techo del contrato. 180 s es lo que el
#: renderizador acepta; 20-45 s es lo que se ve entero.
TARGET_MIN_DURATION = 20.0
TARGET_MAX_DURATION = 45.0
#: Lo que se le perdona al objetivo antes de gastar una reescritura. La
#: estimación de la voz acierta dentro de un segundo por línea, así que un
#: exceso de medio segundo puede ser del estimador y no del spec — y una
#: reescritura cuesta cuatro céntimos y un minuto para ahorrar un segundo que
#: nadie ve. El objetivo sigue siendo 20-45: esto sólo decide cuándo vale la
#: pena decirlo. Sin este margen, la sesión 168 (45,8 s estimados) se llevaba
#: una generación entera por ochocientas milésimas.
TARGET_GRACE = 1.5
class SpecInvalid(Exception):
"""El spec no cumple el contrato. `errors` son rutas + motivo, verbatim."""
def __init__(self, errors: list[str]):
self.errors = errors
super().__init__("; ".join(errors[:5]) or "spec inválido")
# --- validación contra el esquema publicado ---------------------------------
def _resolve(schema: dict, defs: dict) -> dict:
ref = schema.get("$ref")
if not ref:
return schema
name = ref.rsplit("/", 1)[-1]
return defs.get(name, {})
def _type_ok(value: Any, expected: str) -> bool:
if expected == "object":
return isinstance(value, dict)
if expected == "array":
return isinstance(value, list)
if expected == "string":
return isinstance(value, str)
if expected == "integer":
return isinstance(value, int) and not isinstance(value, bool)
if expected == "number":
return isinstance(value, (int, float)) and not isinstance(value, bool)
if expected == "boolean":
return isinstance(value, bool)
if expected == "null":
return value is None
return True
def _check(value: Any, schema: dict, path: str, defs: dict) -> list[str]:
"""Subconjunto de JSON Schema que emite pydantic. Devuelve rutas de error."""
schema = _resolve(schema, defs)
if not schema:
return []
if "anyOf" in schema:
for branch in schema["anyOf"]:
if not _check(value, branch, path, defs):
return []
kinds = [_resolve(b, defs).get("type", "?") for b in schema["anyOf"]]
return [f"{path}: no casa con ninguna alternativa ({', '.join(kinds)})"]
errors: list[str] = []
expected = schema.get("type")
if expected and not _type_ok(value, expected):
return [f"{path}: se esperaba {expected}, llegó {type(value).__name__}"]
if "enum" in schema and value not in schema["enum"]:
allowed = ", ".join(repr(v) for v in schema["enum"])
return [f"{path}: {value!r} no es un valor permitido ({allowed})"]
if isinstance(value, str):
if len(value) < schema.get("minLength", 0):
errors.append(f"{path}: cadena vacía o más corta que "
f"{schema['minLength']} caracteres")
if "maxLength" in schema and len(value) > schema["maxLength"]:
errors.append(f"{path}: {len(value)} caracteres, el máximo es "
f"{schema['maxLength']}")
if isinstance(value, (int, float)) and not isinstance(value, bool):
for key, ok, text in (
("minimum", lambda v, lim: v >= lim, ">="),
("maximum", lambda v, lim: v <= lim, "<="),
("exclusiveMinimum", lambda v, lim: v > lim, ">"),
("exclusiveMaximum", lambda v, lim: v < lim, "<"),
):
if key in schema and not ok(value, schema[key]):
errors.append(f"{path}: {value} debe ser {text} {schema[key]}")
if isinstance(value, list):
if "minItems" in schema and len(value) < schema["minItems"]:
errors.append(f"{path}: {len(value)} elementos, el mínimo es "
f"{schema['minItems']}")
if "maxItems" in schema and len(value) > schema["maxItems"]:
errors.append(f"{path}: {len(value)} elementos, el máximo es "
f"{schema['maxItems']}")
item_schema = schema.get("items")
if item_schema:
for i, item in enumerate(value):
errors.extend(_check(item, item_schema, f"{path}.{i}", defs))
if isinstance(value, dict):
properties = schema.get("properties", {})
for required in schema.get("required", []):
if required not in value:
errors.append(f"{path}.{required}: falta y es obligatorio")
if schema.get("additionalProperties") is False:
for key in value:
if key not in properties:
allowed = ", ".join(sorted(properties)) or "ninguna"
errors.append(f"{path}.{key}: campo no permitido "
f"(las válidas son: {allowed})")
for key, sub in properties.items():
if key in value:
errors.extend(_check(value[key], sub, f"{path}.{key}", defs))
return errors
def _check_props(props: Any, schema: dict, path: str) -> list[str]:
return _check(props, schema, path, schema.get("$defs", {}))
# --- reglas que cruzan campos -----------------------------------------------
# Espejo a mano de los `@model_validator` de shortsmith/spec.py, porque NO salen
# en el JSON Schema publicado: pydantic no los serializa. Antes se dejaban al 422
# del servidor, y eso costaba una generación entera — el 422 llega al RENDERIZAR,
# cuando el bucle de reintentos ya ha terminado, así que el spec no se reescribe:
# se devuelve a mano. La sesión 162 (Trans-en-Provence) se perdió justo así el
# 2026-08-13. Comprobadas aquí, son un reintento normal.
#
# El texto del error es el de shortsmith palabra por palabra: al modelo se le
# devuelve verbatim, y dos redacciones distintas del mismo fallo según dónde se
# cace es exactamente el tipo de detalle que hace inútil un mensaje de error.
def _scale_bars_quote_needs_room(props: dict, path: str) -> list[str]:
bars = props.get("bars")
quote = props.get("quote")
if isinstance(bars, list) and isinstance(quote, list) and len(bars) > 2 and quote:
return [f"{path}: {len(bars)} bars leave no room for a quote — use at "
"most 2 bars with a quote"]
return []
def _track_map_waypoints_inside(props: dict, path: str) -> list[str]:
"""Una ventana fijada a mano tiene que contener la ruta que enmarca.
La proyección de shortsmith es lineal y sin recortar, así que un waypoint
fuera de `bounds` no se dibuja en el borde: se dibuja donde lo ponga la
aritmética, a veces fuera del encuadre. Se rechaza en vez de recortarse
porque un mapa que miente sobre dónde pasó algo es peor que un spec que
falla.
"""
bounds = props.get("bounds")
waypoints = props.get("waypoints")
if not isinstance(bounds, dict) or not isinstance(waypoints, list):
return []
try:
lat_min, lat_max = float(bounds["lat_min"]), float(bounds["lat_max"])
lon_min, lon_max = float(bounds["lon_min"]), float(bounds["lon_max"])
except (KeyError, TypeError, ValueError):
return [] # incompleto o mal tipado: ya lo dijo el esquema
if lat_max <= lat_min or lon_max <= lon_min:
return [f"{path}.bounds: map bounds must have max greater than min on "
"both axes"]
outside = []
for w in waypoints:
if not isinstance(w, dict):
continue
try:
lat, lon = float(w["lat"]), float(w["lon"])
except (KeyError, TypeError, ValueError):
continue
if not (lat_min <= lat <= lat_max and lon_min <= lon <= lon_max):
outside.append(str(w.get("label", "?")))
if outside:
return [f"{path}: waypoints outside the map bounds: "
f"{', '.join(outside)} — widen bounds or omit them to fit the "
"window to the route"]
return []
#: Comillas de apertura. Una cita bien partida abre UNA vez.
_OPENING_QUOTES = "“«„‟"
def _quote_is_one_span(props: dict, path: str) -> list[str]:
"""Una `quote` de varias líneas es UN span partido, no dos citas.
Esta regla NO es de shortsmith: allí renderiza igual. Es del canal, y es de
las duras, porque el fallo que evita es el peor que tiene este sistema —
una cita fabricada con material auténtico y atribuida a una persona con
nombre y apellidos.
El comprobador de fundamento ya une las líneas antes de buscarlas, y eso
cerró la forma con la que falló Socorro en agosto (`“LIKE ALUMINUM` +
`SMOOTH, NO WINDOWS”`): unidas son una sola frase, no aparece en ninguna
fuente, y se rechaza. Pero la unión se puede derrotar poniéndole a cada
línea su propio par de comillas: entonces son DOS citas, cada una
fundamentada por su lado, y pasa en silencio — mientras el fotograma dibuja
la frase de nadie. Le pasó a la sesión 162 el 2026-08-13 con
`“GRAY, LIKE ZINC”` + `“TWO SAUCERS GLUED AT THE RIM”`.
Por eso se mira la FORMA y no el contenido: dos aperturas son dos citas,
diga lo que diga la fuente.
"""
quote = props.get("quote")
if not isinstance(quote, list) or len(quote) < 2:
return []
joined = " ".join(str(line) for line in quote)
openings = sum(joined.count(glyph) for glyph in _OPENING_QUOTES)
if openings < 2:
return []
return [f"{path}.quote: son {openings} citas, y este campo es UNA cita "
"partida en líneas — leídas seguidas forman una frase que nadie "
"dijo. Elige un solo span verbatim y pártelo donde tenga que "
"partirse, o quita las comillas y cuenta el hecho en llano"]
#: Comprobaciones que se aplican a TODAS las plantillas, por nombre de prop. Van
#: aparte de las de abajo para que una plantilla nueva con un campo `quote` de
#: varias líneas quede cubierta sin tocar nada — el mismo pacto que el contrato.
UNIVERSAL_CHECKS = [_quote_is_one_span]
#: template -> comprobaciones extra. Una plantilla sin entrada no tiene reglas
#: cruzadas, que es el caso de casi todas.
CROSS_FIELD_CHECKS = {
"scale_bars": [_scale_bars_quote_needs_room],
"track_map": [_track_map_waypoints_inside],
}
def _check_cross_field(template: str, props: Any, path: str) -> list[str]:
if not isinstance(props, dict):
return []
errors: list[str] = []
for check in (*UNIVERSAL_CHECKS, *CROSS_FIELD_CHECKS.get(template, ())):
errors.extend(check(props, path))
return errors
# --- el sobre ---------------------------------------------------------------
def _check_meta(meta: Any) -> list[str]:
if not isinstance(meta, dict):
return ["meta: se esperaba un objeto"]
errors = []
spec_id = meta.get("id")
if not isinstance(spec_id, str) or not META_ID.match(spec_id):
errors.append("meta.id: minúsculas, dígitos, '_' y '-', empezando por "
f"letra o dígito, hasta 64 caracteres (llegó {spec_id!r})")
if not isinstance(meta.get("title"), str) or not meta.get("title"):
errors.append("meta.title: obligatorio y no vacío")
width = meta.get("width", 1080)
height = meta.get("height", 1920)
if (width, height) not in RESOLUTIONS:
allowed = ", ".join(f"{w}x{h}" for w, h in sorted(RESOLUTIONS))
errors.append(f"meta: {width}x{height} no es una resolución admitida ({allowed})")
if meta.get("fps", 30) not in FPS_VALUES:
errors.append(f"meta.fps: {meta.get('fps')!r} no está entre "
f"{sorted(FPS_VALUES)}")
for key in meta:
if key not in ("id", "title", "width", "height", "fps", "theme"):
errors.append(f"meta.{key}: campo no permitido")
return errors
#: Los presets que existían antes de `GET /audio`. Solo es el default cuando el
#: caller no pasa la paleta viva; con ella, un preset nuevo en shortsmith llega
#: aquí sin tocar este repo — el mismo pacto que las plantillas.
BASELINE_PRESET_NAMES = ("sonar", "none")
def _check_audio(audio: Any, total: float,
presets: Optional[Iterable[str]] = None) -> list[str]:
if audio is None:
return []
if not isinstance(audio, dict):
return ["audio: se esperaba un objeto"]
errors = []
known = tuple(presets) if presets else BASELINE_PRESET_NAMES
if audio.get("preset", "sonar") not in known:
errors.append(f"audio.preset: {audio.get('preset')!r} no existe "
f"(los presets son: {', '.join(sorted(known))})")
silence = audio.get("silence", [])
if not isinstance(silence, list):
return errors + ["audio.silence: se esperaba una lista de pares [inicio, fin]"]
if len(silence) > 16:
errors.append(f"audio.silence: {len(silence)} rangos, el máximo es 16")
for i, rango in enumerate(silence):
if not (isinstance(rango, (list, tuple)) and len(rango) == 2
and all(isinstance(v, (int, float)) for v in rango)):
errors.append(f"audio.silence.{i}: se esperaba [inicio, fin] numérico")
continue
start, end = rango
if start < 0:
errors.append(f"audio.silence.{i}: empieza antes de 0")
if end <= start:
errors.append(f"audio.silence.{i}: el fin no va después del inicio")
if end > total + 1e-9:
errors.append(f"audio.silence.{i}: [{start}, {end}] se sale de la "
f"duración total ({total:.2f}s)")
for key in audio:
if key not in ("preset", "silence"):
errors.append(f"audio.{key}: campo no permitido")
return errors
#: Tope de la narración de un shot, el mismo que aplica shortsmith. Rechazarla
#: aquí cuesta un reintento del modelo; rechazarla allí cuesta el render entero.
MAX_NARRATION_CHARS = 320
def _check_narration(narration: Any, path: str) -> list[str]:
if narration is None:
return []
if not isinstance(narration, str):
return [f"{path}.narration: se esperaba texto"]
if len(narration) > MAX_NARRATION_CHARS:
return [f"{path}.narration: {len(narration)} caracteres, el máximo es "
f"{MAX_NARRATION_CHARS}"]
return []
def _total_duration(spec: dict) -> float:
total = 0.0
for shot in spec.get("shots") or []:
if isinstance(shot, dict) and isinstance(shot.get("duration"), (int, float)):
total += float(shot["duration"])
return total
def validate_spec(spec: Any, templates: dict[str, dict],
presets: Optional[Iterable[str]] = None) -> None:
"""Lanza `SpecInvalid` con TODAS las rutas que fallan.
Se devuelven todos los errores de golpe a propósito: el bucle de reintento
se los da al modelo verbatim y arreglar cinco de una vez sale más barato
que cinco vueltas.
`presets` es la paleta viva de `GET /audio`; sin ella se valida contra la
paleta base, que nunca acepta nada que un shortsmith viejo no renderice.
"""
errors: list[str] = []
if not isinstance(spec, dict):
raise SpecInvalid([f"el spec debe ser un objeto JSON, llegó {type(spec).__name__}"])
if spec.get("version") != 1:
errors.append(f"version: debe ser 1 (llegó {spec.get('version')!r})")
for key in spec:
if key not in ("version", "meta", "audio", "shots"):
errors.append(f"{key}: campo no permitido en la raíz "
"(las válidas son: version, meta, audio, shots)")
errors.extend(_check_meta(spec.get("meta")))
shots = spec.get("shots")
if not isinstance(shots, list) or not shots:
errors.append("shots: hace falta al menos un shot")
raise SpecInvalid(errors)
if len(shots) > MAX_SHOTS:
errors.append(f"shots: {len(shots)} shots, el máximo es {MAX_SHOTS}")
known = ", ".join(sorted(templates))
for i, shot in enumerate(shots):
path = f"shots.{i}"
if not isinstance(shot, dict):
errors.append(f"{path}: se esperaba un objeto")
continue
template = shot.get("template")
if template not in templates:
errors.append(f"{path}.template: {template!r} no existe "
f"(las plantillas son: {known})")
continue
for key in shot:
if key not in ("template", "duration", "props", "narration"):
errors.append(f"{path}.{key}: campo no permitido "
"(las válidas son: template, duration, props, narration)")
errors.extend(_check_narration(shot.get("narration"), path))
duration = shot.get("duration")
if not isinstance(duration, (int, float)) or isinstance(duration, bool):
errors.append(f"{path}.duration: obligatoria y numérica")
elif not MIN_SHOT_DURATION <= duration <= MAX_TOTAL_DURATION:
errors.append(f"{path}.duration: {duration} fuera de "
f"[{MIN_SHOT_DURATION}, {MAX_TOTAL_DURATION}]")
if "props" not in shot:
errors.append(f"{path}.props: falta y es obligatorio")
continue
props_path = f"{path}.{template}.props"
props_errors = _check_props(shot["props"], templates[template], props_path)
errors.extend(props_errors)
# Sólo si el esquema pasó: con props mal tipadas, una regla cruzada
# diría algo que no es el fallo real y taparía el que sí lo es.
if not props_errors:
errors.extend(_check_cross_field(template, shot["props"], props_path))
total = _total_duration(spec)
if total < MIN_TOTAL_DURATION:
errors.append(f"shots: la duración total ({total:.2f}s) no llega al "
f"mínimo de {MIN_TOTAL_DURATION}s")
if total > MAX_TOTAL_DURATION:
errors.append(f"shots: la duración total ({total:.2f}s) pasa del límite "
f"de {MAX_TOTAL_DURATION}s")
errors.extend(_check_audio(spec.get("audio"), total, presets))
if errors:
raise SpecInvalid(errors)
#: Caracteres por segundo de la voz, sin contar las pausas. Medido el
#: 2026-08-12 sintetizando de verdad las 28 líneas de narración que el bot ha
#: escrito hasta hoy con el mismo Piper y las mismas banderas que usa shortsmith
#: (`en_US-lessac-medium`, length_scale 1.0, --noise_scale 0 --noise_w 0):
#: 2429 caracteres en 140,91 s de audio.
NARRATION_CHARS_PER_SECOND = 18.5
#: Piper añade este silencio DESPUÉS DE CADA FRASE, no sólo al final de la
#: línea, y es un valor que shortsmith fija a propósito (`voice.SENTENCE_SILENCE`).
#: Contarlo por separado es lo que arregla el caso raro: "Witness identities.
#: Sensor details. Locations redacted." son tres frases cortas que valen 0,75 s
#: de pausa, y un modelo de caracteres a secas las da por rápidas.
NARRATION_SENTENCE_SILENCE = 0.25
#: El respiro que shortsmith deja tras cada línea antes de permitir el corte.
NARRATION_PAD = 0.45
#: Palabras por segundo de la misma medida (387 palabras en 140,91 s). Sólo se
#: usa para traducir un exceso de segundos a palabras en el aviso: al modelo se
#: le pide que recorte texto, no tiempo.
NARRATION_WORDS_PER_SECOND = 2.75
#: El respiro redondeado hacia arriba, para la regla que se le enseña al modelo.
#: `NARRATION_PAD` son 0,45 s; "medio segundo" se sostiene en la cabeza y va
#: sobrado, que es la dirección correcta en la que equivocarse.
NARRATION_ROUNDED_PAD = 0.5
#: Final de frase: un punto pegado a la palabra y seguido de espacio o de nada.
#: El decimal de "1.5" no cuenta, y por eso mira lo que va detrás.
_SENTENCE_END = re.compile(r"[.!?](?=\s|$)")
def sentence_count(line: str) -> int:
"""Frases de una línea, contadas como las cuenta Piper para sus pausas."""
return max(1, len(_SENTENCE_END.findall(line))) if line.strip() else 0
def teachable_seconds(words: int, sentences: int = 1) -> float:
"""Lo que hay que DECLARAR para una línea, en las unidades que el modelo cuenta.
Es `spoken_seconds` traducido de caracteres a palabras. La traducción hace
falta porque un LLM no cuenta caracteres de fiar, pero sí cuenta palabras y
puntos — y la regla tiene que ser computable por quien debe obedecerla, o no
es una regla, es un deseo.
Los dos términos son los mismos que los de la voz. La versión anterior del
prompt colapsaba el segundo en un "+ medio segundo" fijo, y ese es el mismo
error de clase que tenía el estimador antes del 2026-08-12: sin pausa por
frase, una línea troceada en frases cortas se da por rápida. Medido contra
las 28 líneas reales, aquella regla se quedaba corta en 14 y hasta 2,27 s —
o sea que un modelo que la obedeciera al pie de la letra seguiría
infradeclarando la mitad de sus planos. Con el término por frase el peor
caso baja a 1,27 s y sólo en 5 de 28.
"""
return (words / NARRATION_WORDS_PER_SECOND
+ sentences * NARRATION_SENTENCE_SILENCE
+ NARRATION_ROUNDED_PAD)
def max_words_in(seconds: float, sentences: int = 1) -> int:
"""Cuántas palabras caben en un plano de esa duración, según la regla de arriba.
Existe para que el tope de palabras por línea y el tope de duración de plano
no puedan volver a contradecirse: se deriva uno del otro en vez de escribir
los dos a mano.
"""
room = seconds - sentences * NARRATION_SENTENCE_SILENCE - NARRATION_ROUNDED_PAD
return max(1, int(room * NARRATION_WORDS_PER_SECOND))
def spoken_seconds(line: str) -> float:
"""Lo que tarda la voz en decir una línea, sin el respiro final.
Dos términos porque la voz tiene dos: lee a ritmo casi constante y se calla
un cuarto de segundo en cada punto. La versión anterior sólo tenía el
primero y con un ritmo medido sobre una única frase — 14,2 car/s —, así que
sobreestimaba cada línea alrededor de un 20 %. Sobre un Short entero eso son
de cuatro a seis segundos de duración que no existen, suficientes para que
el bucle de reescritura se disparara con vídeos que estaban dentro del
objetivo.
"""
line = " ".join(line.split())
if not line:
return 0.0
sentences = max(1, len(_SENTENCE_END.findall(line)))
return (len(line) / NARRATION_CHARS_PER_SECOND
+ sentences * NARRATION_SENTENCE_SILENCE)
def estimated_duration(spec: dict) -> float:
"""Lo que durará el vídeo, no lo que suman las duraciones declaradas.
Con narración, la duración declarada es un suelo: shortsmith estira el shot
si la frase no cabe. Sin esta estimación el modelo escribiría 40 s de shots,
les colgaría narración a todos y recibiría un Short de 55 s sin que nada le
hubiera avisado — el aviso llegaría del render, cuando ya está pagado.
Contrastada contra los tres MP4 que hay renderizados (sesiones 166, 167 y
168): 39,42 / 47,19 / 45,81 s estimados contra 39,57 / 47,53 / 45,40 reales.
"""
total = 0.0
for shot in spec.get("shots") or []:
if not isinstance(shot, dict):
continue
declared = shot.get("duration")
declared = float(declared) if isinstance(declared, (int, float)) else 0.0
narration = shot.get("narration")
if isinstance(narration, str) and narration.strip():
declared = max(declared, spoken_seconds(narration) + NARRATION_PAD)
total += declared
return total
def editorial_notes(spec: dict) -> list[str]:
"""Lo que no viola el contrato pero sí el encargo.
Va aparte de `validate_spec` justo porque no impide renderizar: un Short de
70 s se ve, sólo que peor. Se le devuelve al modelo como comentario una vez;
si insiste, se renderiza igual antes que tirar la generación a la basura.
"""
notes = []
declared = _total_duration(spec)
total = estimated_duration(spec)
stretched = total > declared + 0.5
how = (f"la duración estimada son {total:.1f}s con la narración "
f"({declared:.1f}s de shots)" if stretched
else f"la duración total son {total:.1f}s")
if total < TARGET_MIN_DURATION - TARGET_GRACE:
notes.append(f"{how} y el objetivo es "
f"{TARGET_MIN_DURATION:.0f}-{TARGET_MAX_DURATION:.0f}s: "
"queda corto, añade un shot o alarga los que tienes")
elif total > TARGET_MAX_DURATION + TARGET_GRACE:
notes.append(f"{how} y el objetivo es "
f"{TARGET_MIN_DURATION:.0f}-{TARGET_MAX_DURATION:.0f}s: "
+ _how_to_trim(spec, total - TARGET_MAX_DURATION, stretched))
return notes
def _how_to_trim(spec: dict, excess: float, stretched: bool) -> str:
"""El consejo, en la unidad en la que el modelo puede obedecerlo.
"Recorta narración" no dice cuánta, y las tres veces que se ha disparado
esto el modelo devolvió un spec que seguía pasándose. Un exceso en segundos
tampoco le sirve, porque no escribe segundos: escribe frases. Así que el
aviso va en palabras y señala DÓNDE están las más largas.
"""
if not stretched:
return (f"sobran {excess:.1f}s: recorta un shot o baja las duraciones "
"declaradas")
words = max(3, round(excess * NARRATION_WORDS_PER_SECOND))
advice = (f"sobran {excess:.1f}s, unas {words} palabras de narración — la voz "
"manda sobre la duración declarada, así que acortar los shots no "
"quita ni un segundo")
spoken = sorted(
((i, len((s.get("narration") or "").split()))
for i, s in enumerate(spec.get("shots") or []) if isinstance(s, dict)),
key=lambda pair: -pair[1])
spoken = [pair for pair in spoken if pair[1]]
if not spoken:
return advice
# Sólo las que de verdad son largas: señalar una línea de dos palabras al
# lado de una de veinte convierte el consejo en ruido.
named = [f"shots.{i} ({n} palabras)"
for i, n in spoken[:2] if n * 2 >= spoken[0][1]]
return advice + f"; {'las líneas más largas son' if len(named) > 1 else 'la línea más larga es'} {' y '.join(named)}"
# --- el contrato en prosa, para el prompt -----------------------------------
def _describe_field(name: str, schema: dict, required: bool, defs: dict,
indent: str = " ") -> list[str]:
schema = _resolve(schema, defs)
bits: list[str] = []
if "anyOf" in schema:
inner = [b for b in schema["anyOf"] if _resolve(b, defs).get("type") != "null"]
if inner:
return _describe_field(name, inner[0], required, defs, indent) + \
[f"{indent} (opcional, admite null)"]
kind = schema.get("type", "?")
if "enum" in schema:
bits.append("uno de: " + ", ".join(str(v) for v in schema["enum"]))
elif kind == "array":
item = _resolve(schema.get("items", {}), defs)
bits.append("lista")
if "minItems" in schema or "maxItems" in schema:
bits.append(f"{schema.get('minItems', 0)}-{schema.get('maxItems', '∞')} elementos")
else:
bits.append(kind)
if schema.get("minLength"):
bits.append("no vacío")
if "maxLength" in schema:
bits.append(f"máx {schema['maxLength']} caracteres")
# `x-fits` es cuánto texto cabe DIBUJADO al tamaño de diseño, medido por
# shortsmith contra sus propias fuentes. No se valida — los caracteres son
# un proxy de los píxeles — pero es lo único que evita que el modelo escriba
# una cita de 58 caracteres en un hueco de 16 y salga dibujada ilegible.
if "x-fits" in schema:
bits.append(f"CABE ~{schema['x-fits']} caracteres dibujados")
for key, text in (("minimum", "≥"), ("maximum", "≤"),
("exclusiveMinimum", ">"), ("exclusiveMaximum", "<")):
if key in schema:
bits.append(f"{text} {schema[key]}")
bits.append("OBLIGATORIO" if required else f"opcional (por defecto {schema.get('default')!r})")
lines = [f"{indent}{name}: {', '.join(bits)}"]
# Los objetos (sueltos o dentro de una lista) se despliegan: si no, el
# modelo ve "waypoints: lista" y no sabe que cada uno lleva label/lat/lon.
nested = _resolve(schema.get("items", {}), defs) if kind == "array" else schema
if nested.get("type") == "object" and nested.get("properties"):
nested_required = set(nested.get("required", []))
for sub, sub_schema in nested["properties"].items():
lines.extend(_describe_field(sub, sub_schema, sub in nested_required,
defs, indent + " "))
return lines
def describe_templates(templates: dict[str, dict]) -> str:
"""El contrato tal cual lo publica el servicio, en prosa compacta.
Se describe lo recibido, sin lista de plantillas escrita a mano: una
plantilla nueva en shortsmith aparece aquí sin tocar este repo.
"""
blocks = []
for name in sorted(templates):
schema = templates[name] or {}
defs = schema.get("$defs", {})
required = set(schema.get("required", []))
lines = [f"{name}:"]
for field, field_schema in schema.get("properties", {}).items():
lines.extend(_describe_field(field, field_schema, field in required, defs))
blocks.append("\n".join(lines))
return "\n\n".join(blocks)