feat(short): generación y render de Shorts vía shortsmith
Build & Deploy ResearchOwl / build-and-push (push) Successful in 9s

Añade /generate short_en y /short_spec. El pipeline genera un shot spec
con Haiku, verifica cada cifra, fecha y cita contra los chunks de la
sesión, lo renderiza en shortsmith y entrega el MP4 por Telegram junto
a un informe de claims.

- ShortsmithClient con sondeo y fallback al spec JSON si el render falla
- Contrato de plantillas obtenido de GET /templates, no codificado
- Comprobación de fundamento determinista, sin LLM
- outputs.published_url para enlazar el artículo de Ghost
- Normalización de comillas rectas a tipográficas (ver KNOWN-ISSUES.md)

Lo que no aparece en los chunks se contrasta contra el ejemplo del
prompt: si casa ahí es fuga, no invención, y se informa como tal. El
purgado de sesiones se lleva también su MP4.

La subida a YouTube queda fuera a propósito: fase 3.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ChemaVX
2026-08-01 21:55:42 +00:00
co-authored by Claude Opus 5
parent 8b81ef87e4
commit 20c8d03aa7
27 changed files with 4350 additions and 23 deletions
+533
View File
@@ -0,0 +1,533 @@
"""Comprobador de fundamento: ¿cada dato del spec sale de las fuentes?
Un spec de Short no es prosa que un humano juzgue al leerla: es un contrato
tipado que se convierte en vídeo publicado. La premisa entera del canal es que
sus cifras vienen de fuentes primarias, así que una cifra de radar inventada en
un vídeo de 40 segundos es peor que no publicar vídeo.
Esto extrae del spec todo lo que afirma un hecho — citas, cifras, fechas y
nombres propios — y confirma que aparece en al menos un chunk de la sesión.
**Aquí no entra ningún LLM.** Tiene que ser determinista y gratis: si el
comprobador alucinara, no comprobaría nada. Todo es normalización + substring.
El sesgo es deliberado: se esperan falsos positivos (una cita reformulada no
casa aunque el hecho esté en la fuente). Un falso positivo cuesta un vistazo;
un falso negativo cuesta la credibilidad del canal.
"""
from __future__ import annotations
import json
import re
import unicodedata
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
from typing import Any, Iterable, Optional
__all__ = [
"Claim",
"GroundingReport",
"check_grounding",
"extract_claims",
"normalize",
]
#: El ejemplo trabajado que viaja en el prompt. Se contrasta contra él para
#: poder distinguir dos diagnósticos que no piden lo mismo (ver
#: `check_grounding`).
EXAMPLE_PATH = Path(__file__).parent / "examples" / "jal1628.json"
# --- normalización ----------------------------------------------------------
#: Glifos de comilla que hay que unificar antes de comparar: el spec lleva
#: tipográficas («“ ”») y las fuentes scrapeadas, cualquier cosa.
_QUOTE_GLYPHS = "“”„‟«»″ʺ"
_APOSTROPHES = "‘’ʼ′"
_DASHES = "‐‑‒–—―−"
_SEPARATORS = "·•∙|\t "
#: Separador de millares dentro de un número: 35,000 y 35.000 -> 35000. El
#: lookahead exige exactamente tres dígitos, así que 1.5 y 61.22 se quedan
#: como están. "1.234" en el sentido decimal se convertiría en 1234, que es un
#: precio asumido: en este dominio los millares son mucho más frecuentes.
_THOUSANDS = re.compile(r"(?<=\d)[.,](?=\d{3}(?!\d))")
def normalize(text: str) -> str:
"""Forma canónica para comparar los dos lados. Idempotente."""
if not text:
return ""
out = unicodedata.normalize("NFKC", text)
out = out.translate({ord(c): '"' for c in _QUOTE_GLYPHS})
out = out.translate({ord(c): "'" for c in _APOSTROPHES})
out = out.translate({ord(c): "-" for c in _DASHES})
out = out.translate({ord(c): " " for c in _SEPARATORS})
out = _THOUSANDS.sub("", out)
out = out.casefold()
return " ".join(out.split())
# --- vocabulario ------------------------------------------------------------
#: Unidades reconocidas -> alias equivalentes. Sirven para dos cosas: decidir
#: si un número "lleva unidad" (y por tanto afirma algo) y para casar "35,000
#: FT" con una fuente que escribe "35,000 feet".
_UNIT_ALIASES: dict[str, set[str]] = {
"ft": {"ft", "feet", "foot", "pies", "pie"},
"m": {"m", "meter", "meters", "metre", "metres", "metro", "metros"},
"km": {"km", "kilometer", "kilometers", "kilometre", "kilometres",
"kilometro", "kilometros"},
"mi": {"mi", "mile", "miles", "milla", "millas"},
"nm": {"nm", "nmi", "nautical", "naut"},
"mph": {"mph"},
"kt": {"kt", "kts", "knot", "knots", "nudo", "nudos"},
"kph": {"kph", "kmh"},
"%": {"%", "percent", "pct", "porciento"},
"deg": {"deg", "degree", "degrees", "grado", "grados", "°"},
"sec": {"sec", "secs", "second", "seconds", "segundo", "segundos"},
"min": {"min", "mins", "minute", "minutes", "minuto", "minutos"},
"hour": {"hour", "hours", "hr", "hrs", "hora", "horas"},
"day": {"day", "days", "dia", "dias"},
"week": {"week", "weeks", "semana", "semanas"},
"month": {"month", "months", "mes", "meses"},
"year": {"year", "years", "yr", "yrs", "ano", "anos"},
"page": {"page", "pages", "pagina", "paginas"},
"kg": {"kg", "kilo", "kilos", "kilogram", "kilograms"},
"lb": {"lb", "lbs", "pound", "pounds", "libra", "libras"},
"ton": {"ton", "tons", "tonne", "tonnes", "tonelada", "toneladas"},
"mhz": {"mhz"},
"ghz": {"ghz"},
"km2": {"km2"},
}
_UNIT_LOOKUP: dict[str, str] = {
alias: canon for canon, aliases in _UNIT_ALIASES.items() for alias in aliases
}
_MONTHS: dict[str, int] = {}
for _i, _names in enumerate([
("january", "jan", "enero", "ene"),
("february", "feb", "febrero"),
("march", "mar", "marzo"),
("april", "apr", "abril", "abr"),
("may", "mayo"),
("june", "jun", "junio"),
("july", "jul", "julio"),
("august", "aug", "agosto", "ago"),
("september", "sep", "sept", "septiembre", "setiembre"),
("october", "oct", "octubre"),
("november", "nov", "noviembre"),
("december", "dec", "diciembre", "dic"),
], start=1):
for _n in _names:
_MONTHS[_n] = _i
#: Palabras función que no identifican nada. Se usan sólo para el respaldo por
#: tokens de los nombres: un nombre cuyos tokens significativos aparecen todos
#: en las fuentes se da por fundamentado aunque la frase entera no case.
_STOPWORDS = {
"the", "a", "an", "and", "or", "of", "in", "on", "at", "to", "for", "with",
"by", "from", "as", "is", "was", "were", "are", "be", "been", "that",
"this", "these", "those", "it", "its", "his", "her", "their", "no", "not",
"el", "la", "los", "las", "un", "una", "unos", "unas", "de", "del", "y",
"o", "en", "con", "por", "para", "que", "se", "su", "sus", "al", "es",
"son", "fue", "fueron", "lo",
}
#: Claves cuyo valor de texto es una etiqueta identificadora — donde viven los
#: nombres propios ("ELMENDORF ROCC", "CAPT. KENJU TERAUCHI"). La prosa
#: (headline, caption, footnote…) no entra entera: de ella se sacan citas,
#: cifras y fechas, que es lo que afirma hechos.
_NAME_KEYS = {"label", "key", "card_title", "source", "attribution",
"sublabel", "count_label"}
#: Claves numéricas que AFIRMAN un dato. El resto de números del spec son
#: geometría o tiempo de render (lat, lon, duration, sweeps, turn_deg,
#: contact_bearing_deg, markers, bounds…) y no se comprueban: no dicen nada
#: sobre el mundo. Si una plantilla nueva añade un número que sí afirma algo,
#: se añade aquí.
_FACT_NUMBER_KEYS = {"value", "count_to"}
#: Listas de texto que forman UNA frase continua al dibujarse (el renderizador
#: no envuelve: el caller parte la cita en líneas). Se unen antes de comprobar.
_JOINED_LIST_KEYS = {"quote"}
# --- extracción -------------------------------------------------------------
@dataclass(frozen=True)
class Claim:
"""Un dato afirmado por el spec, con dónde vive."""
text: str # tal cual aparece en el spec, para enseñárselo a un humano
kind: str # quote | figure | date | name
path: str # shots.3.scale_bars.props.bars.1.value_label
unit: Optional[str] = None # canónica, cuando la cifra la lleva
_date: Optional[tuple] = None # (dia|None, mes, año) para el respaldo de fechas
@property
def norm(self) -> str:
return normalize(self.text)
_QUOTED = re.compile(r'["“„‟«]([^"“”„‟«»]{3,})'
r'["”„‟»]')
_DATE_DMY = re.compile(r"\b(\d{1,2})\s+([A-Za-zÀ-ž]{3,12})\.?,?\s+(\d{4})\b")
_DATE_MDY = re.compile(r"\b([A-Za-zÀ-ž]{3,12})\.?\s+(\d{1,2})(?:st|nd|rd|th)?,?\s+(\d{4})\b")
_DATE_ISO = re.compile(r"\b(\d{4})-(\d{2})-(\d{2})\b")
_DATE_SLASH = re.compile(r"\b(\d{1,2})[/](\d{1,2})[/](\d{2,4})\b")
_YEAR = re.compile(r"\b(1[5-9]\d{2}|20\d{2})\b")
_NUMBER = re.compile(
r"(?P<num>\d{1,3}(?:[.,]\d{3})+(?:\.\d+)?|\d+(?:\.\d+)?)\s*"
r"(?P<unit>%|°|[A-Za-zÀ-ž]{1,10})?")
_TOKEN = re.compile(r"[A-Za-zÀ-ž][A-Za-zÀ-ž'.\-]*")
def _mask(text: str, start: int, end: int) -> str:
"""Tapa un tramo ya extraído para que no lo vuelva a coger otra regla."""
return text[:start] + " " * (end - start) + text[end:]
def _month_number(token: str) -> Optional[int]:
return _MONTHS.get(normalize(token).strip(". "))
def _claims_from_text(text: str, path: str, key: str) -> list[Claim]:
"""Todo lo que afirma un hecho dentro de una cadena del spec.
El orden importa: las fechas se extraen y se tapan antes que los números,
porque si no "17 NOV 1986" produciría además la cifra suelta 1986.
"""
if not text or not text.strip():
return []
claims: list[Claim] = []
rest = text
# 1. Citas entrecomilladas. Son verbatim por definición: se comprueban tal cual.
for m in _QUOTED.finditer(text):
inner = m.group(1).strip()
if inner:
claims.append(Claim(inner, "quote", path))
rest = _mask(rest, m.start(), m.end())
# 2. Fechas, con sus componentes para el respaldo (día+mes+año en el mismo chunk).
for regex, order in ((_DATE_DMY, "dmy"), (_DATE_MDY, "mdy"),
(_DATE_ISO, "ymd"), (_DATE_SLASH, "dmy_num")):
for m in regex.finditer(rest):
if order == "dmy":
day, month, year = m.group(1), _month_number(m.group(2)), m.group(3)
elif order == "mdy":
month, day, year = _month_number(m.group(1)), m.group(2), m.group(3)
elif order == "ymd":
year, month, day = m.group(1), int(m.group(2)), m.group(3)
else:
day, month, year = m.group(1), int(m.group(2)), m.group(3)
if not 1 <= month <= 12:
continue
if not month:
continue # "5 RADARS 1986" no es una fecha: el token no es un mes
claims.append(Claim(m.group(0).strip(), "date", path,
_date=(int(day), int(month), int(year))))
rest = _mask(rest, m.start(), m.end())
# 3. Cifras: las que llevan unidad reconocida, o las de magnitud (separador
# de millares o >= 1000). "3 RADARS" no afirma una medida y se deja pasar;
# "35,000 FT", "1,500" y "50 minutes" sí.
for m in _NUMBER.finditer(rest):
raw_num, raw_unit = m.group("num"), m.group("unit")
unit = _UNIT_LOOKUP.get(normalize(raw_unit or ""))
had_separator = bool(re.search(r"\d[.,]\d{3}", raw_num))
try:
magnitude = float(normalize(raw_num))
except ValueError:
continue
if not unit and not had_separator and magnitude < 1000:
continue
if not unit or not raw_unit:
shown = raw_num
elif raw_unit in "%°":
shown = f"{raw_num}{raw_unit}" # 360°, no 360 °
else:
shown = f"{raw_num} {raw_unit}"
claims.append(Claim(shown.strip(), "figure", path, unit=unit))
rest = _mask(rest, m.start(), m.end("num") if not unit else m.end())
# 4. Años sueltos que hayan sobrevivido ("40 YEARS" no; "SINCE 1986" sí).
for m in _YEAR.finditer(rest):
claims.append(Claim(m.group(0), "date", path,
_date=(None, None, int(m.group(0)))))
rest = _mask(rest, m.start(), m.end())
# 5. Nombres propios: sólo en posiciones de etiqueta, y sólo si queda algo
# que identifique. "10,000+" (sin letras) y "29 YEARS" (cuya única
# palabra es una unidad) ya viajaron como cifra; repetirlos como nombre
# sólo alarga el informe.
if key in _NAME_KEYS and _name_worth_checking(text):
claims.append(Claim(text.strip(), "name", path))
return claims
def _name_worth_checking(text: str) -> bool:
"""¿Queda algún token que identifique a alguien o algo? Las unidades y los
meses no cuentan: ya viajan dentro de la cifra o de la fecha."""
return any(t not in _UNIT_LOOKUP and t not in _MONTHS
for t in _significant_tokens(text))
def _walk(node: Any, path: str, key: str, claims: list[Claim]) -> None:
if isinstance(node, dict):
for k, v in node.items():
_walk(v, f"{path}.{k}", k, claims)
elif isinstance(node, list):
if key in _JOINED_LIST_KEYS and all(isinstance(x, str) for x in node):
# Una cita partida en líneas es UNA cita.
_walk(" ".join(node), path, key, claims)
return
for i, v in enumerate(node):
_walk(v, f"{path}.{i}", key, claims)
elif isinstance(node, str):
claims.extend(_claims_from_text(node, path, key))
elif isinstance(node, bool):
return
elif isinstance(node, (int, float)) and key in _FACT_NUMBER_KEYS:
claims.append(Claim(_pretty_number(node), "figure", path))
def _pretty_number(value: float) -> str:
return str(int(value)) if float(value).is_integer() else str(value)
def _attach_units(props: dict, claims: list[Claim], base_path: str) -> list[Claim]:
"""Un dict con `value` numérico y `unit` de texto (una barra de escala) dibuja
los dos juntos: "232 FT". Se detecta por forma, no por plantilla."""
out = []
for claim in claims:
if claim.kind == "figure" and claim.path == f"{base_path}.value" and props.get("unit"):
out.append(Claim(f"{claim.text} {props['unit']}", "figure", claim.path,
unit=_UNIT_LOOKUP.get(normalize(str(props["unit"])))))
else:
out.append(claim)
return out
def extract_claims(spec: dict) -> list[Claim]:
"""Todos los datos afirmados por los shots del spec, sin duplicados.
`meta` queda fuera a propósito: el título del spec no se dibuja en ningún
fotograma, es el nombre del fichero.
"""
claims: list[Claim] = []
for i, shot in enumerate(spec.get("shots") or []):
if not isinstance(shot, dict):
continue
template = shot.get("template", "?")
props = shot.get("props") or {}
base = f"shots.{i}.{template}.props"
shot_claims: list[Claim] = []
_walk(props, base, "props", shot_claims)
# Reconstruye "232 FT" a partir de {value: 232, unit: "FT"}.
for path_prefix, sub in _dicts_with_value_and_unit(props, base):
shot_claims = _attach_units(sub, shot_claims, path_prefix)
claims.extend(shot_claims)
seen: set[tuple[str, str]] = set()
unique: list[Claim] = []
for claim in claims:
fingerprint = (claim.kind, claim.norm)
if not claim.norm or fingerprint in seen:
continue
seen.add(fingerprint)
unique.append(claim)
return unique
def _dicts_with_value_and_unit(node: Any, path: str) -> Iterable[tuple[str, dict]]:
if isinstance(node, dict):
if isinstance(node.get("value"), (int, float)) and node.get("unit"):
yield path, node
for k, v in node.items():
yield from _dicts_with_value_and_unit(v, f"{path}.{k}")
elif isinstance(node, list):
for i, v in enumerate(node):
yield from _dicts_with_value_and_unit(v, f"{path}.{i}")
# --- comprobación -----------------------------------------------------------
@dataclass
class GroundingReport:
grounded: list[Claim] = field(default_factory=list)
ungrounded: list[Claim] = field(default_factory=list)
#: Ni en las fuentes ni inventado: copiado del ejemplo del prompt.
contaminated: list[Claim] = field(default_factory=list)
chunk_count: int = 0
url_count: int = 0
@property
def total(self) -> int:
return len(self.grounded) + len(self.ungrounded) + len(self.contaminated)
@property
def unsupported(self) -> list[Claim]:
"""Todo lo que no se apoya en las fuentes, sea cual sea el diagnóstico."""
return self.ungrounded + self.contaminated
@property
def clean(self) -> bool:
return not self.unsupported
def summary(self) -> str:
"""Texto del informe de claims (§8 del spec de fase 2).
Con cero claims sin fundamento TAMBIÉN se informa: un éxito silencioso
enseña al lector a dejar de mirar.
"""
lines = [f"{len(self.grounded)} claims casados con las fuentes"]
if self.ungrounded:
lines.append(f"⚠️ {len(self.ungrounded)} sin encontrar:")
for claim in self.ungrounded:
lines.append(f" • [{claim.kind}] \"{claim.text}\"")
else:
lines.append("✅ 0 sin encontrar")
if self.contaminated:
lines.append("")
lines.append(f"🧪 {len(self.contaminated)} copiados del EJEMPLO del prompt "
"(fuga, no invención — bórralos o sustitúyelos por datos "
"de esta sesión):")
for claim in self.contaminated:
lines.append(f" • [{claim.kind}] \"{claim.text}\"")
lines.append("")
lines.append(f"Fuentes: {self.chunk_count} chunks de {self.url_count} URLs")
return "\n".join(lines)
def _unit_near(haystack: str, number: str, unit: str, window: int = 40) -> bool:
"""¿Aparece la unidad (o un alias) cerca de esa cifra en el texto?"""
aliases = _UNIT_ALIASES.get(unit, {unit})
for m in re.finditer(rf"(?<!\d){re.escape(number)}(?!\d)", haystack):
tail = haystack[m.end():m.end() + window]
if any(re.search(rf"\b{re.escape(a)}", tail) for a in aliases):
return True
return False
def _number_present(haystack: str, number: str) -> bool:
return re.search(rf"(?<![\d.,]){re.escape(number)}(?![\d])", haystack) is not None
def _date_present(haystack: str, parts: tuple) -> bool:
"""Respaldo de fechas: "17 NOV 1986" contra una fuente que escribe
"November 17, 1986". Los tres componentes en el mismo chunk bastan."""
day, month, year = parts
if str(year) not in haystack:
return False
if month is None:
return True
names = [n for n, num in _MONTHS.items() if num == month]
if not any(re.search(rf"\b{n}", haystack) for n in names) and \
not re.search(rf"(?<!\d){month:02d}(?!\d)", haystack) and \
not re.search(rf"(?<!\d){month}(?!\d)", haystack):
return False
return day is None or _number_present(haystack, str(day))
def _significant_tokens(text: str) -> list[str]:
"""Tokens que identifican algo. La puntuación de cola se cae para que
"CAPT." case por substring contra "captain"."""
tokens = (t.strip(".-'") for t in _TOKEN.findall(normalize(text)))
return [t for t in tokens if len(t) >= 3 and t not in _STOPWORDS]
def _is_grounded(claim: Claim, haystacks: list[str]) -> bool:
needle = claim.norm
if any(needle in h for h in haystacks):
return True
if claim.kind == "quote":
return False # una cita o es verbatim o no es una cita
if claim.kind == "figure":
number = normalize(claim.text.split()[0]) if claim.text else ""
if not number:
return False
if claim.unit:
return any(_unit_near(h, number, claim.unit) for h in haystacks)
return any(_number_present(h, number) for h in haystacks)
if claim.kind == "date":
return any(_date_present(h, claim._date) for h in haystacks) if claim._date else False
# name: cada token significativo tiene que aparecer en las fuentes. Casa
# "CAPT. KENJU TERAUCHI" con "Captain Kenju Terauchi" sin dejar pasar un
# "ELMENDORF ROCC" donde ninguna fuente menciona ROCC.
tokens = _significant_tokens(claim.text)
if not tokens:
return False
return all(any(t in h for h in haystacks) for t in tokens)
@lru_cache(maxsize=1)
def _example_haystacks() -> tuple[str, ...]:
"""El ejemplo del prompt, troceado para poder buscar dentro.
Se junta lo que el ejemplo DIBUJA (sus claims, con la cita reconstruida) y
sus cadenas sueltas. **Un trozo por campo, no un texto único**: el respaldo
de fechas casa día, mes y año dentro del MISMO pajar, y en un ejemplo con
"17 NOV 1986" y "5 MARCH 1987" pegados, un "17 NOV 1987" inventado parecería
venir de ahí.
"""
try:
example = json.loads(EXAMPLE_PATH.read_text(encoding="utf-8"))
except Exception:
return ()
pieces = [c.text for c in extract_claims(example)]
for shot in example.get("shots", []):
_collect_strings(shot.get("props") or {}, pieces)
return tuple(sorted({normalize(p) for p in pieces if normalize(p)}))
def _collect_strings(node: Any, out: list[str]) -> None:
if isinstance(node, dict):
for v in node.values():
_collect_strings(v, out)
elif isinstance(node, list):
for v in node:
_collect_strings(v, out)
elif isinstance(node, str) and node.strip():
out.append(node)
def check_grounding(spec: dict, chunks: list[dict],
example_haystacks: Optional[tuple[str, ...]] = None
) -> GroundingReport:
"""Comprueba el spec contra el material de la sesión.
`chunks` son filas de la tabla `chunks` (con `content` y, si viene del join
con `sources`, `url`). No se descarta ningún shot ni se reintenta a ciegas:
se devuelve el informe y decide un humano si es una fabricación real o un
artefacto de formato.
Lo que no aparece en los chunks se contrasta ADEMÁS contra el ejemplo
trabajado que viaja en el prompt. Si casa ahí, no es una invención: es una
fuga del prompt — el modelo copió una cifra del ejemplo en vez de sacarla
de la sesión (medido el 2026-08-01: "232 FT", el largo de un 747, en cero
de 126 chunks). Son dos diagnósticos distintos y piden acciones distintas:
una invención hay que verificarla, una fuga hay que borrarla.
El ejemplo se queda como está — con cifras reales — a propósito: uno
sintético enseña peor la forma, y la defensa estructural es esto.
"""
haystacks = [normalize(c.get("content") or "") for c in chunks]
haystacks = [h for h in haystacks if h]
urls = {c.get("url") for c in chunks if c.get("url")}
from_example = _example_haystacks() if example_haystacks is None else example_haystacks
report = GroundingReport(chunk_count=len(chunks), url_count=len(urls))
for claim in extract_claims(spec):
if _is_grounded(claim, haystacks):
report.grounded.append(claim)
elif from_example and _is_grounded(claim, list(from_example)):
report.contaminated.append(claim)
else:
report.ungrounded.append(claim)
return report