"""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) @property def fingerprint(self) -> tuple: """Identidad del dato, no de su redacción — para no contar dos veces. Una cifra se identifica por su número y su unidad CANÓNICA: "35,000 FT" dibujado en pantalla y "35,000 feet" dicho en la narración son el mismo dato, y con la narración esa coincidencia pasa a ser lo normal, no la excepción. Contarlos por separado inflaría justo el informe del que depende la revisión humana. Lo demás se identifica por su forma normalizada: una cita reformulada NO es la misma cita, y ahí la literalidad es el criterio correcto. """ if self.kind == "figure": number = normalize(self.text.split()[0]) if self.text.split() else "" return ("figure", number, self.unit) return (self.kind, self.norm) _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\d{1,3}(?:[.,]\d{3})+(?:\.\d+)?|\d+(?:\.\d+)?)\s*" r"(?P%|°|[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. `narration` SÍ entra, y es de lo más importante que entra: lo que se dibuja en pantalla son etiquetas cortas que el modelo copia, pero la narración es prosa que redacta — el sitio natural para deslizar una cifra de más. Se trata como cualquier prosa del spec: aporta sus citas, cifras y fechas. No aporta nombres propios, por lo mismo que no los aportan `headline` o `caption`: una frase entera no es una etiqueta identificadora, y sacar nombres de dentro de la prosa exigiría adivinar por mayúsculas y llenaría el informe de ruido. La cifra y la cita, que son lo que se fabrica, están cubiertas. """ 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) narration = shot.get("narration") if isinstance(narration, str): shot_claims.extend( _claims_from_text(narration, f"shots.{i}.narration", "narration")) claims.extend(shot_claims) seen: set[tuple] = set() unique: list[Claim] = [] for claim in claims: fingerprint = claim.fingerprint 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"(? bool: return re.search(rf"(? 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"(? 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