Author SHA1 Message Date
Renovate Bot a38c2e1eca chore(deps): update dependency numpy to v2 2026-08-09 18:00:21 +00:00
11 changed files with 46 additions and 802 deletions
-29
View File
@@ -86,35 +86,6 @@ What actually landed, and where it differs from the plan below:
thing that order revealed: with the voice repeating on-screen figures, claims had to
be de-duplicated by *canonical unit* ("35,000 FT" and "35,000 feet" are one claim) or
every narrated Short would double its own review report.
- **researchowl's estimate of the voice had to be measured, not assumed** (2026-08-12).
It shipped with 14.2 characters per second, taken from a single line, and that
overshot every narration by about a fifth — four to six seconds on a whole Short,
enough to make the spec writer rewrite videos that were already inside the target.
Every generation since narration shipped had spent all three attempts on it.
Synthesizing the 28 narration lines the bot had actually written gave 18.5 char/s
**plus 0.25 s at every full stop**, which is the term that matters: Piper's
`SENTENCE_SILENCE` is per sentence, so "Witness identities. Sensor details. Locations
redacted." costs three quarters of a second that a characters-only model gives away.
Estimates now land within half a second of the three rendered MP4s. The lesson is the
older one restated: a constant taken from one sample is a guess with a decimal point.
- **The spec writer was under-declaring because the prompt asked for three things that
could not all be true** (2026-08-13). Every narrated spec the bot had written declared
less time than its own narration needed on three or four shots out of five or six —
3.8 to 8.2 seconds of drift per Short. The video still came out the right length,
because `plan()` grows the shot; what was wrong was that the spec described a visual
rhythm that never rendered. The cause was not a lazy model. The prompt asked at once
for lines of up to 18 words, shots of at most 6 seconds, and enough declared time for
the line — and 18 words need 7.3 s, so the set is unsatisfiable. The model broke the
only one of the three that nothing checked. Three fixes, all deterministic and none
costing a generation to find: the taught rule got the per-sentence term it was missing
(`words/2.75 + 0.5` fell short on 14 of the 28 measured lines, by up to 2.27 s — so
even perfect obedience under-declared); the word cap is now *derived* from the shot
cap rather than written by hand, so the contradiction cannot come back; and the worked
example, which violated its own rule on two of its six lines, was cut to obey it. That
last one is the lesson worth keeping: **the example is the strongest signal in the
prompt, so an example that breaks a rule teaches the breakage**, whatever the prose
says. It is the same finding as "el ejemplo del prompt habla, y por eso los specs
vuelven a hablar", arriving a second time.
Original plan, kept for the record:
+1 -1
View File
@@ -23,7 +23,7 @@ aiosqlite==0.22.1
# Processing
tiktoken==0.7.0
numpy==1.26.4
numpy==2.5.2
scikit-learn==1.5.1
# Claude API (scoring)
+1 -15
View File
@@ -854,13 +854,7 @@ def _upload_message(video, metadata: dict, article_url: Optional[str]) -> str:
lines = [f"🎬 {video.title}", "", f"Revisar y publicar: {video.studio_url}",
f"Enlace del vídeo: {video.watch_url}", ""]
if video.visibility_contradiction:
# Primera línea del mensaje, no una nota al pie: si esto pasa, el vídeo
# ya está en la calle mientras lees el informe de fundamento.
lines.insert(0, "🚨 EL VÍDEO SE VE SIN INICIAR SESIÓN, aunque YouTube "
"dijo que lo subía en privado. Ocúltalo en Studio antes "
"de nada — el enlace de abajo lleva ahí.\n")
elif video.privacy_status == "private":
if video.privacy_status == "private":
lines.append(
"🔒 Está PRIVADO. Los vídeos subidos por API desde un proyecto sin "
"auditar se quedan así: el candado es del proyecto, no del vídeo, y "
@@ -869,14 +863,6 @@ def _upload_message(video, metadata: dict, article_url: Optional[str]) -> str:
else:
lines.append(f"👁 Visibilidad: {video.privacy_status}")
# Lo comprobado, aparte de lo que dijo la API: son dos cosas distintas y el
# 2026-08-12 se demostró que conviene no confundirlas.
if video.reachable is False:
lines.append("✔ Comprobado desde fuera: no se ve sin sesión.")
elif video.reachable is None:
lines.append("⚠️ No se pudo comprobar la visibilidad desde fuera; me "
"queda sólo lo que dijo la API. Míralo en Studio.")
if video.forced_private:
lines.append("⚠️ Pediste otra visibilidad y YouTube la forzó a privada. "
"Es exactamente la firma de ese candado.")
+2 -2
View File
@@ -139,7 +139,7 @@
"caption": "CONTACT HOLDS RELATIVE POSITION",
"turn_deg": 360
},
"narration": "He tried to shake it. Full circle, steep descent, and it stayed there."
"narration": "He tried to shake it. Full circle, steep descent, and it was still there."
},
{
"template": "signal_strips",
@@ -207,7 +207,7 @@
"url": "THEEXCLUSIONZONE.COM",
"show_mark": true
},
"narration": "The file was never closed. It was left where anyone can read it."
"narration": "The file was never closed. It was filed, and left where anyone can read it."
}
]
}
+19 -131
View File
@@ -22,9 +22,7 @@ from typing import Any, Awaitable, Callable, Optional
import structlog
from src.generator.spec_contract import (
SpecInvalid, describe_templates, editorial_notes, estimated_duration,
max_words_in, validate_spec, NARRATION_ROUNDED_PAD,
NARRATION_SENTENCE_SILENCE, NARRATION_WORDS_PER_SECOND,
SpecInvalid, describe_templates, editorial_notes, validate_spec,
TARGET_MAX_DURATION, TARGET_MIN_DURATION,
)
@@ -34,42 +32,6 @@ logger = structlog.get_logger()
#: que arreglar es el prompt, no este número.
MAX_ATTEMPTS = 3
#: Reescrituras que se gastan en una nota editorial, no en un fallo de contrato.
#: UNA. Un spec que ya cumple el contrato y sólo se pasa de duración es
#: renderizable: la segunda reescritura no compraba un Short mejor, compraba una
#: generación más. Medido sobre las sesiones 166, 167 y 168 — las tres gastaron
#: los tres intentos por duración y las tres acabaron renderizando un spec que
#: seguía pasándose. Los intentos que quedan son para el contrato, que sí es
#: binario. Ver `_how_to_trim` en `spec_contract`: si la nota no se obedece a la
#: primera, lo que hay que arreglar es la nota.
NOTE_ATTEMPTS = 1
#: Cuánta narración cabe en un Short entero. Comprobación cruzada de la regla
#: de arriba, en la unidad que el modelo escribe: 80 palabras son unos 29 s de
#: voz, y con los respiros y algún plano mudo eso deja el vídeo cerca de 40 s.
#: El ejemplo de referencia habla 74. La sesión 167 habló 97 y salió a 47,5 s.
NARRATION_WORD_BUDGET = 80
#: Lo que dura un plano como mucho. Del ejemplo (6,0 s el más largo, 5,6 de
#: media). Sin este tope el modelo declaraba 42 s en seis planos de siete
#: segundos y LUEGO les colgaba la narración encima: el primer borrador salía a
#: 50 s las tres veces que se midió, y hacía falta una reescritura entera para
#: bajarlo.
MAX_SHOT_DURATION = 6.0
#: Lo que mide una línea. No es preferencia de estilo: es la media del ejemplo.
NARRATION_WORDS_PER_LINE = 12
#: Y el tope duro NO se escribe a mano: es cuántas palabras caben en el plano
#: más largo que se permite declarar. Escribirlos por separado fue el fallo que
#: hizo que el modelo infradeclarase casi todos sus planos, y no por pereza: el
#: prompt le pedía a la vez líneas de hasta 18 palabras, planos de 6 s como
#: mucho y tiempo declarado suficiente para su propia voz. Las tres a la vez son
#: imposibles — 18 palabras piden 7,3 s — así que el modelo rompía la única que
#: nadie comprobaba, la duración declarada. Derivando el tope de la duración, la
#: contradicción no puede volver.
NARRATION_WORDS_PER_LINE_MAX = max_words_in(MAX_SHOT_DURATION, sentences=2)
EXAMPLE_PATH = Path(__file__).parent / "examples" / "jal1628.json"
__all__ = ["ShortSpecWriter", "SpecResult", "SpecWriteFailed", "NARRATIVE_SHAPES"]
@@ -122,18 +84,10 @@ exists, and a prop name that is not listed is a parse error, not a nuance.
# 3. Rules
- **Total duration {target_min:.0f}-{target_max:.0f} seconds, and the voice is \
what decides it, not the durations you declare.** The contract allows 180; that \
is a ceiling, not a target. A narrated shot runs as long as its line takes to \
say the renderer never cuts the voice off, it grows the shot so the whole \
video is really about {word_budget} words of narration and no more. That is the \
number to hold: **count the words of every `narration` you write, and stop at \
{word_budget}.** Section 3b has the arithmetic behind it.
- Typically 6-9 shots, and **none of them longer than {max_shot:.0f} seconds** \
that is the example's longest, and its average is 5.6. Give a shot the \
seconds its content needs to be read: a card with four rows needs longer than a \
headline. A {max_shot:.0f}-second shot with a short line on it is not a \
generous shot, it is a shot the viewer has already finished reading.
- Total duration 20-45 seconds. The contract allows 180; that is a ceiling, \
not a target. Aim for {target_min:.0f}-{target_max:.0f}.
- Typically 6-9 shots. Give a shot the seconds its content needs to be read: \
a card with four rows needs longer than a headline.
- Every string is drawn as given. Write them the way they should appear: \
SHORT, UPPERCASE, no trailing punctuation. A headline is 2-5 words.
- **"CABE ~N caracteres dibujados" is a width, and it is the one limit nothing \
@@ -162,13 +116,9 @@ aloud by the renderer and burned in as captions. Write it for the ear.
- **The first line is the whole hook.** Two seconds decide whether anyone \
watches the rest, and the opening shot's narration is those two seconds. Lead \
with the strangest true thing you have, not with a preamble.
- **Keep a line to {words_per_line} words, hard stop at {words_per_line_max}.** \
That is the example's own average, and the hard stop is not a style preference \
either it is exactly as much as fits in the longest shot you are allowed to \
declare. {words_per_line_max} words is {max_shot:.0f} seconds; the same limit, \
written twice. Go past it and the shot has to grow, because the renderer will \
not cut your voice off it makes the shot longer instead, and a Short that \
drifts past {target_max:.0f} seconds is a Short people leave.
- Keep a line under 25 words. Long sentences lose the listener and stretch the \
shot; the renderer will not cut your voice off, it will make the shot longer \
instead, and a Short that drifts past 45 seconds is a Short people leave.
- **Do not read the screen aloud.** The captions already show your words and \
the template already shows its own. If the shot draws "35,000 FT", the voice \
says what that altitude meant, not the number again.
@@ -182,34 +132,10 @@ silent exactly the two that draw a quotation, where the voice would only be \
competing with words already on the frame. Chosen silence is an edit; a spec \
with one narrated shot out of eight is not a Short with a voice, it is a Short \
that forgot to speak.
- **Give every narrated shot enough time for its own line, and work it out \
rather than guessing.** The voice reads about {words_per_second:g} words a \
second and pauses a quarter second at every full stop, so **count the words AND \
count the sentences**:
duration words ÷ {words_per_second:g} + {sentence_pause} × sentences + \
{rounded_pad}
The second term is the one that catches people out. "Witness identities. \
Sensor details. Locations redacted." is six words and three full stops: it is \
not a fast line, it is three quarters of a second of silence on top. Two lines \
of the same length do not take the same time if one of them is chopped.
Worked, on the example below: shot 1 speaks twelve words in one sentence, so \
12 ÷ {words_per_second:g} + {sentence_pause} + {rounded_pad} = 5.1, and it \
declares 5.5. Shot 4 speaks thirteen words in two sentences, so 13 ÷ \
{words_per_second:g} + 0.5 + {rounded_pad} = 5.7, and it declares 6.0. Round \
up, never down.
This is the one rule that makes your own arithmetic true: a shot runs for the LONGER of \
its declared duration and its line never shorter, the voice is never cut off \
so a shot that declares less than its line silently grows, and the video ends \
up longer than the durations you wrote. Hold this rule and the total you \
declare IS the video's length; break it once and nothing you counted means \
anything.
- As a cross-check, all the narration in the spec together should come to about \
{word_budget} words. The example below speaks 74. A spec that spoke 97 rendered \
at 47.5 seconds and had to be cut.
- Narration costs seconds. A shot is never cut short to fit the voice it \
grows instead so a line that needs six seconds in a four-second shot pushes \
your whole total past the target. Write the line, then give the shot the time \
the line actually takes.
- Everything in section 4 applies to narration word for word. It is prose you \
compose rather than a label you copy, which makes it the easiest place to \
slip in a figure no source gave you and it is checked exactly like the rest.
@@ -433,25 +359,6 @@ def _format_notes(notes: list[str]) -> str:
"Return the adjusted JSON object.")
def _off_target(spec: dict) -> float:
"""Segundos fuera de la ventana editorial. 0 = dentro."""
total = estimated_duration(spec)
return max(0.0, TARGET_MIN_DURATION - total, total - TARGET_MAX_DURATION)
def _closer_to_target(a: Optional[SpecResult], b: SpecResult) -> SpecResult:
"""De dos specs válidos, el que menos se sale del objetivo.
Antes se guardaba el PRIMERO válido y punto, con lo que una reescritura que
obedecía la nota a medias 53 s en vez de 58 se tiraba entera y salía el
largo. El empate se lo lleva el anterior: sin razón para cambiar, no se
cambia.
"""
if a is None:
return b
return a if _off_target(a.spec) <= _off_target(b.spec) else b
#: (system, prompt) -> texto del modelo.
LLMCall = Callable[[str, str], Awaitable[str]]
@@ -481,13 +388,6 @@ class ShortSpecWriter:
domain=domain,
target_min=TARGET_MIN_DURATION,
target_max=TARGET_MAX_DURATION,
words_per_second=NARRATION_WORDS_PER_SECOND,
word_budget=NARRATION_WORD_BUDGET,
words_per_line=NARRATION_WORDS_PER_LINE,
words_per_line_max=NARRATION_WORDS_PER_LINE_MAX,
max_shot=MAX_SHOT_DURATION,
sentence_pause=f"{NARRATION_SENTENCE_SILENCE:g}",
rounded_pad=f"{NARRATION_ROUNDED_PAD:g}",
example=_load_example(),
article=article,
context=context,
@@ -504,8 +404,6 @@ class ShortSpecWriter:
#: Un spec que cumple el contrato pero se pasa de duración. Se guarda
#: para que un intento posterior peor no lo tire: es renderizable.
best: Optional[SpecResult] = None
#: Reescrituras ya gastadas en notas editoriales.
note_rounds = 0
for attempt in range(1, MAX_ATTEMPTS + 1):
if on_progress and attempt > 1:
@@ -542,33 +440,23 @@ class ShortSpecWriter:
notes = editorial_notes(spec)
result = SpecResult(spec=spec, attempts=attempt, notes=notes,
history=list(history))
if not notes:
logger.info("short spec válido", attempts=attempt,
shots=len(spec.get("shots", [])), notes=0)
return result
best = _closer_to_target(best, result)
if note_rounds < NOTE_ATTEMPTS and attempt < MAX_ATTEMPTS:
# Nota editorial, no violación del contrato: se comenta y, si
# insiste, se renderiza el intento que menos se pase.
note_rounds += 1
if notes and attempt < MAX_ATTEMPTS:
# Nota editorial, no violación del contrato: se comenta una vez
# y, si insiste, se renderiza igual.
best = best or result
history.append(notes)
feedback = _format_notes(notes)
continue
logger.info("short spec válido pero fuera de objetivo",
attempts=attempt, shots=len(best.spec.get("shots", [])),
off_target=round(_off_target(best.spec), 1))
best.attempts = attempt
best.history = history
return best
logger.info("short spec válido", attempts=attempt,
shots=len(spec.get("shots", [])), notes=len(notes))
return result
if best is not None:
# Un intento anterior sí cumplía el contrato. Vale más un Short
# largo que ningún Short.
logger.info("short spec: se recupera el intento válido anterior",
attempts=MAX_ATTEMPTS, notes=best.notes)
best.attempts = MAX_ATTEMPTS
best.history = history
return best
+11 -134
View File
@@ -28,10 +28,6 @@ __all__ = [
"validate_spec",
"editorial_notes",
"estimated_duration",
"spoken_seconds",
"sentence_count",
"teachable_seconds",
"max_words_in",
"describe_templates",
"TARGET_MIN_DURATION",
"TARGET_MAX_DURATION",
@@ -51,15 +47,6 @@ META_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
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."""
@@ -334,90 +321,12 @@ def validate_spec(spec: Any, templates: dict[str, dict],
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
#: Caracteres por segundo de la voz (Piper `en_US-lessac-medium` a length_scale
#: 1.0). Medido el 2026-08-06: 82 caracteres en 5.78 s. Sirve para ESTIMAR aquí
#: lo que shortsmith sabrá exacto al sintetizar.
NARRATION_CHARS_PER_SECOND = 14.2
#: 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 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:
@@ -427,9 +336,6 @@ def estimated_duration(spec: dict) -> float:
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 []:
@@ -439,7 +345,8 @@ def estimated_duration(spec: dict) -> float:
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)
spoken = len(narration.strip()) / NARRATION_CHARS_PER_SECOND + NARRATION_PAD
declared = max(declared, spoken)
total += declared
return total
@@ -459,48 +366,18 @@ def editorial_notes(spec: dict) -> list[str]:
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:
if total < TARGET_MIN_DURATION:
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:
elif total > TARGET_MAX_DURATION:
fix = ("recorta narración: la voz manda sobre la duración declarada"
if stretched else "recorta shots o acorta duraciones")
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))
f"{TARGET_MIN_DURATION:.0f}-{TARGET_MAX_DURATION:.0f}s: {fix}")
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,
+1 -71
View File
@@ -45,15 +45,6 @@ UPLOAD_URL = "https://www.googleapis.com/upload/youtube/v3/videos"
#: del canal: si el token se filtra, lo peor que se puede hacer con él es subir.
SCOPE = "https://www.googleapis.com/auth/youtube.upload"
#: Con `youtube.upload` no se le puede PREGUNTAR a la API por el estado de un
#: vídeo, así que la visibilidad se comprueba desde fuera y sin credenciales:
#: oEmbed contesta 200 a un vídeo que se ve sin sesión y 401/404 a uno que no.
#: Es la única forma de contrastar lo que dice la respuesta de la subida sin
#: cambiar un token que sólo sabe subir por uno que puede vaciar el canal.
OEMBED_URL = "https://www.youtube.com/oembed"
#: Segunda pasada por si YouTube aún no había indexado el vídeo recién subido.
_VISIBILITY_RECHECK_DELAY = 3.0
#: Márgen antes de que caduque el token de acceso (dura 3600 s).
_TOKEN_MARGIN = 120.0
#: Tokens de acceso en memoria por client_id. El bot crea un uploader nuevo en
@@ -106,16 +97,6 @@ class UploadedVideo:
upload_status: str = ""
#: Por qué YouTube marcó el vídeo como no reproducible, si lo hizo.
rejection_reason: str = ""
#: Si el vídeo se ve sin iniciar sesión, comprobado desde fuera en vez de
#: creerle a la respuesta de la subida. None = no se pudo comprobar.
reachable: Optional[bool] = None
@property
def visibility_contradiction(self) -> bool:
"""La API dice privado y el vídeo se ve. Es el caso que hay que gritar:
todo el flujo de revisión informe de fundamento primero, publicar
después descansa en que subir NO publica."""
return self.reachable is True and self.privacy_status == "private"
@property
def watch_url(self) -> str:
@@ -355,62 +336,11 @@ class YouTubeUploader:
rejection_reason=(status.get("rejectionReason")
or status.get("failureReason") or ""),
)
await _report(on_progress, "🔎 Comprobando la visibilidad…")
result.reachable = await self.reachable(result.video_id)
logger.info("Short subido a YouTube", video_id=result.video_id,
privacy=result.privacy_status,
forced_private=result.forced_private,
reachable=result.reachable)
if result.visibility_contradiction:
logger.error("El vídeo se ve sin sesión y la API lo dio por privado",
video_id=result.video_id)
forced_private=result.forced_private)
return result
async def reachable(self, video_id: str) -> Optional[bool]:
"""¿Se ve este vídeo sin haber iniciado sesión?
True = cualquiera con el enlace lo ve. False = no. None = no se pudo
averiguar.
**La asimetría es deliberada.** Un 200 PRUEBA que el vídeo es accesible;
un 404 no prueba que sea privado, porque también lo devuelve un vídeo
que YouTube todavía no ha terminado de indexar segundos después de
subirlo. Por eso sólo el 200 dispara un aviso, y por eso el negativo se
reintenta una vez antes de darlo por bueno.
Nunca levanta: esto contrasta un dato, no lo produce. Si la red falla, el
vídeo ya está subido y lo que toca es decir que no se pudo comprobar
no convertir una comprobación en el motivo de que la subida parezca
haber fallado.
"""
if not video_id:
return None
params = {"url": f"https://www.youtube.com/watch?v={video_id}",
"format": "json"}
seen: Optional[bool] = None
for attempt in (1, 2):
try:
async with self._session(20) as sess:
async with sess.get(OEMBED_URL, params=params) as resp:
status = resp.status
except (aiohttp.ClientError, OSError) as e:
logger.warning("No se pudo comprobar la visibilidad",
video_id=video_id, error=str(e))
return seen
if status == 200:
return True
if status in (401, 403, 404):
seen = False
else:
logger.warning("oEmbed contestó algo inesperado",
video_id=video_id, status=status)
return seen
if attempt == 1:
import asyncio
await asyncio.sleep(_VISIBILITY_RECHECK_DELAY)
return seen
async def _start(self, token: str, metadata: dict, size: int) -> str:
"""Paso 1: los metadatos. Devuelve la URL de subida (cabecera Location)."""
headers = {
-34
View File
@@ -103,40 +103,6 @@ def test_upload_message_warns_when_the_description_has_no_article():
assert "force" in text
def test_upload_message_shouts_when_the_video_is_already_watchable():
"""El caso serio, y va PRIMERO en el mensaje.
Si subir publica, el informe de fundamento se lee cuando el vídeo ya está en
la calle el orden entero del flujo deja de significar nada. Enterarse
tiene que costar cero atención: en la primera línea o no sirve.
"""
from src.bot.bot import _upload_message
text = _upload_message(_uploaded(reachable=True), {}, "https://x.test/")
assert "SE VE SIN INICIAR SESIÓN" in text.split("\n")[0]
assert "studio.youtube.com/video/abc123/edit" in text
# Y no se cuenta a la vez el cuento tranquilizador del candado.
assert "auditoría" not in text
def test_upload_message_confirms_a_video_nobody_can_watch():
from src.bot.bot import _upload_message
text = _upload_message(_uploaded(reachable=False), {}, "https://x.test/")
assert "no se ve sin sesión" in text
assert "SE VE SIN INICIAR SESIÓN" not in text
def test_upload_message_admits_when_it_could_not_check():
"""No haber podido comprobar no es haber comprobado que no. Decir "privado"
a secas aquí sería dar por garantía lo que sólo es la palabra de la API."""
from src.bot.bot import _upload_message
text = _upload_message(_uploaded(reachable=None), {}, "https://x.test/")
assert "No se pudo comprobar" in text
assert "Míralo en Studio" in text
def test_upload_message_is_plain_text():
"""Va sin parse_mode: lleva el título del modelo, y un Markdown roto haría
que Telegram rechazara justo el mensaje que trae el enlace."""
+6 -169
View File
@@ -134,128 +134,15 @@ def test_the_worked_example_narrates_most_of_its_shots():
def test_the_worked_example_declares_time_for_its_own_narration():
"""Un plano que se queda corto para su propia voz enseña a infradeclarar: el
render no corta la voz, alarga el plano, y el total se va del objetivo."""
from src.generator.spec_contract import (
NARRATION_PAD, sentence_count, spoken_seconds, teachable_seconds,
)
from src.generator.spec_contract import NARRATION_CHARS_PER_SECOND
example = json.loads(EXAMPLE.read_text())
for i, shot in enumerate(example["shots"]):
narration = shot.get("narration", "")
if not narration:
continue
# La cuenta que el prompt le pide al modelo, aplicada al ejemplo que le
# pone delante. Si no cuadran, la regla en prosa pierde: el ejemplo es
# la señal más fuerte. Dos de estas seis líneas NO cumplían — y ese es
# exactamente el defecto que el modelo copiaba.
rule = teachable_seconds(len(narration.split()), sentence_count(narration))
assert shot["duration"] >= rule - 1e-9, \
f"shot {i} declara menos de lo que su propia regla pide"
# Y contra la voz medida, no sólo contra la regla que la aproxima.
assert shot["duration"] >= spoken_seconds(narration) + NARRATION_PAD - 1e-9, \
f"shot {i} se quedaría corto para su propia voz"
def test_the_prompt_gives_a_budget_the_model_can_count():
""""20-45 segundos" no es accionable: la duración real no está escrita en el
spec, sale de sumar el mayor entre lo declarado y lo que tarda la voz. El
modelo puede contar sus `duration` y sus palabras, así que el encargo se
le da en esas dos unidades."""
from src.generator.shortspec import NARRATION_WORD_BUDGET
from src.generator.spec_contract import NARRATION_WORDS_PER_SECOND
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", None, "X.TEST")
# La constante exacta, no redondeada: el prompt trae una cuenta trabajada, y
# con "2.8" el divisor mostrado no reproduce el resultado mostrado.
assert f"{NARRATION_WORDS_PER_SECOND:g} words a second" in prompt, \
"sin el ritmo de la voz no hay cuenta que el modelo pueda hacer"
assert f"words ÷ {NARRATION_WORDS_PER_SECOND:g}" in prompt
assert f"{NARRATION_WORD_BUDGET} words" in prompt
def test_the_worked_example_obeys_the_budget_it_preaches():
"""El ejemplo es la señal más fuerte del prompt — más que cualquier regla en
prosa. Uno que hablara de más enseñaría a hablar de más, dijera lo que
dijera la sección 3b."""
from src.generator.shortspec import (
MAX_SHOT_DURATION, NARRATION_WORDS_PER_LINE,
NARRATION_WORDS_PER_LINE_MAX, NARRATION_WORD_BUDGET,
)
example = json.loads(EXAMPLE.read_text())
lines = [len(s["narration"].split()) for s in example["shots"] if s.get("narration")]
assert max(s["duration"] for s in example["shots"]) == MAX_SHOT_DURATION
assert sum(lines) <= NARRATION_WORD_BUDGET
assert max(lines) <= NARRATION_WORDS_PER_LINE_MAX
# El tope corto se anuncia como "la media del ejemplo": si deja de serlo, la
# regla en prosa se convierte en un número inventado y el modelo la nota.
assert round(sum(lines) / len(lines)) == NARRATION_WORDS_PER_LINE
def test_the_longest_line_allowed_fits_in_the_longest_shot_allowed():
"""La contradicción que hacía infradeclarar, convertida en test.
El prompt pedía a la vez líneas de hasta 18 palabras, planos de 6 s como
mucho, y tiempo declarado suficiente para la propia voz. Las tres juntas son
imposibles 18 palabras piden 7,3 s y el modelo rompía la única que nadie
comprobaba. Si alguien vuelve a subir el tope de palabras a mano, esto salta.
"""
from src.generator.shortspec import (
MAX_SHOT_DURATION, NARRATION_WORDS_PER_LINE_MAX,
)
from src.generator.spec_contract import teachable_seconds
# En el caso malo: una línea al tope, partida en dos frases (dos pausas).
assert teachable_seconds(NARRATION_WORDS_PER_LINE_MAX, 2) <= MAX_SHOT_DURATION
# Y el tope es apretado, no una holgura cómoda que esconda otra vez el fallo:
# una palabra más ya no cabría.
assert teachable_seconds(NARRATION_WORDS_PER_LINE_MAX + 1, 2) > MAX_SHOT_DURATION
def test_the_prompt_rule_counts_the_pauses_and_not_only_the_words():
"""Palabras por segundo a secas es el mismo error que tenía el estimador.
Medido contra las 28 líneas que el bot ha narrado de verdad, `words/2.75 +
0.5` se quedaba corta en 14 y hasta 2,27 s: obedecerla al pie de la letra
seguía infradeclarando media docena de planos. La regla del prompt tiene que
llevar el término por frase, y tiene que ser LA MISMA que aplican los tests.
"""
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", None, "X.TEST")
assert "× sentences" in prompt
assert "count the sentences" in prompt
# Una línea troceada cuesta más que una seguida con las mismas palabras.
from src.generator.spec_contract import teachable_seconds
assert teachable_seconds(12, 3) > teachable_seconds(12, 1)
def test_the_worked_arithmetic_in_the_prompt_is_actually_right():
"""Un ejemplo numérico equivocado enseña la cuenta equivocada, y se lee antes
que la fórmula."""
import re as _re
from src.generator.spec_contract import sentence_count, teachable_seconds
w, _ = writer("{}")
prompt = w.build_prompt("Caso X", "material", None, "X.TEST")
example = json.loads(EXAMPLE.read_text())
worked = _re.findall(
r"[Ss]hot (\d+) speaks \w+ words in \w+ sentences?, so [^=]+= ([\d.]+)",
prompt)
# Sin esto el test pasa en vacío si alguien reescribe el párrafo.
assert len(worked) == 2, f"no se encontraron las cuentas trabajadas: {worked}"
for index, claimed in worked:
narration = example["shots"][int(index)]["narration"]
real = teachable_seconds(len(narration.split()), sentence_count(narration))
assert abs(real - float(claimed)) < 0.05, \
f"el prompt dice {claimed}s para shots.{index}, la regla da {real:.2f}s"
needs = len(narration) / NARRATION_CHARS_PER_SECOND
assert shot["duration"] >= needs, f"shot {i} declara menos de lo que habla"
def test_prompt_says_out_loud_that_there_is_no_article_yet():
@@ -333,20 +220,15 @@ async def test_three_failures_raise_but_keep_the_last_attempt():
@pytest.mark.asyncio
async def test_an_off_target_duration_is_commented_once_then_accepted():
"""70 s cumple el contrato pero no el encargo: se comenta UNA vez y, si el
modelo insiste, se renderiza igual antes que tirar la generación.
Una y no dos. El tercer intento se reserva para el contrato, que es
binario: un spec largo se ve, uno malformado no se puede ni renderizar.
"""
"""70 s cumple el contrato pero no el encargo: se comenta y, si el modelo
insiste, se renderiza igual antes que tirar la generación."""
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0 # 70 s en total
w, llm = writer(json.dumps(long_spec))
result = await w.write("Caso X", "material")
assert result.attempts == 2, "una nota no vale dos reescrituras"
assert len(llm.prompts) == 2
assert result.attempts == MAX_ATTEMPTS
assert result.notes and "recorta" in result.notes[0]
assert "off-brief" in llm.prompts[1]
@@ -363,51 +245,6 @@ async def test_a_valid_attempt_is_not_thrown_away_by_a_worse_one():
assert result.notes
@pytest.mark.asyncio
async def test_the_rewrite_is_kept_when_it_obeys_the_note_only_halfway():
"""Obedecer a medias es obedecer. Antes se guardaba el PRIMER intento válido
y se descartaba la reescritura entera, así que un spec que había bajado de
70 s a 50 s salía a 70."""
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0 # 70 s
better = json.loads(json.dumps(GOOD))
better["shots"][0]["duration"] = 35.0 # 50 s: sigue pasándose, pero menos
w, _ = writer(json.dumps(long_spec), json.dumps(better))
result = await w.write("Caso X", "material")
assert result.spec["shots"][0]["duration"] == 35.0
assert result.attempts == 2
@pytest.mark.asyncio
async def test_a_rewrite_that_makes_it_worse_is_discarded():
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0 # 70 s
worse = json.loads(json.dumps(GOOD))
worse["shots"][0]["duration"] = 90.0 # 105 s
w, _ = writer(json.dumps(long_spec), json.dumps(worse))
result = await w.write("Caso X", "material")
assert result.spec["shots"][0]["duration"] == 55.0
assert result.attempts == 2, "se pagaron dos generaciones aunque valga la primera"
@pytest.mark.asyncio
async def test_a_note_does_not_eat_the_attempt_the_contract_needs():
"""Si la reescritura sale malformada, aún queda un intento para arreglarla."""
long_spec = json.loads(json.dumps(GOOD))
long_spec["shots"][0]["duration"] = 55.0
w, llm = writer(json.dumps(long_spec), "esto no es JSON", json.dumps(GOOD))
result = await w.write("Caso X", "material")
assert result.attempts == 3 and result.notes == []
assert "off-brief" in llm.prompts[1]
assert "not a valid JSON" in llm.prompts[2] or "no es un objeto JSON" in llm.prompts[2]
@pytest.mark.asyncio
async def test_the_contract_is_refetched_after_a_validation_failure():
"""Si el renderizador se actualizó a mitad de la run, la plantilla nueva
+2 -107
View File
@@ -283,73 +283,15 @@ def test_an_unknown_shot_key_still_names_the_valid_ones():
assert any("narration" in e for e in errors_of(doc))
#: Líneas de narración de specs que se renderizaron de verdad, con lo que tarda
#: Piper en decirlas. Medido el 2026-08-12 con el binario, el modelo y las
#: banderas de shortsmith (`en_US-lessac-medium`, length_scale 1.0,
#: --noise_scale 0 --noise_w 0), que son deterministas: estos segundos se
#: reproducen. Se eligieron los extremos del muestreo de 28 líneas — la más
#: rápida, la más lenta y las dos más largas — porque son las que rompen un
#: modelo mal calibrado; la media la aguanta cualquiera.
MEASURED = [
("Eight FBI witness interviews. Five digital renderings. All describe the "
"same shape flying across America for twenty-four years.", 7.809),
("The files are public now, but sections remain blacked out. Witness "
"identities. Sensor details. Locations redacted.", 8.140),
("Three hundred seventy-eight files released. Hundreds of incidents "
"documented. And the government still cannot explain what those shapes "
"were.", 7.681),
("The files came out. The numbers stayed classified.", 3.310),
("Nothing should have been able to hold station beside them up there.", 3.396),
("The Air Force's own investigators called it unexplained.", 2.990),
]
@pytest.mark.parametrize("line,real", MEASURED)
def test_the_estimate_lands_within_a_second_of_the_voice(line, real):
"""La estimación es lo único que separa un aviso útil de una reescritura
inventada, así que se contrasta contra audio medido, no contra misma.
El margen es un segundo. Más apretado sería falso esto estima, no
sintetiza y más ancho deja de decir nada: el error del modelo anterior
sobre un Short entero era de cuatro a seis segundos, y de ahí salían los
tres intentos que se gastaban en cada generación.
"""
from src.generator.spec_contract import spoken_seconds
assert spoken_seconds(line) == pytest.approx(real, abs=1.0)
def test_a_line_of_short_sentences_is_not_taken_for_fast_prose():
"""Piper calla un cuarto de segundo en cada punto. Cuatro frases cortas son
un segundo de silencio, y contarlas como texto corrido las da por rápidas:
es el caso donde más se equivocaba el modelo de sólo caracteres."""
from src.generator.spec_contract import spoken_seconds
chopped = "The files are public now, but sections remain blacked out. " \
"Witness identities. Sensor details. Locations redacted."
flowing = "The files are public now but sections remain blacked out with " \
"witness identities sensor details and locations redacted"
assert len(chopped) < len(flowing)
assert spoken_seconds(chopped) > spoken_seconds(flowing)
def test_a_decimal_point_is_not_the_end_of_a_sentence():
from src.generator.spec_contract import spoken_seconds
assert spoken_seconds("It climbed to 1.5 miles") == \
pytest.approx(spoken_seconds("It climbed to 155 miles"))
def test_the_estimate_counts_the_voice_not_just_the_declared_seconds():
"""La duración declarada es un suelo: shortsmith estira el shot si la frase
no cabe, y el modelo tiene que enterarse ANTES de pagar el render."""
from src.generator.spec_contract import estimated_duration
doc = spec_with(shot(duration=3.0))
doc["shots"][0]["narration"] = MEASURED[0][0] # 7,81 s de voz medidos
doc["shots"][0]["narration"] = "A" * 142 # ~10 s de voz
assert estimated_duration(doc) > 8.0
assert estimated_duration(doc) > 10.0
def test_a_shot_with_room_for_its_line_is_estimated_as_declared():
@@ -378,53 +320,6 @@ def test_narration_that_overshoots_the_target_is_flagged_as_narration():
assert "narración" in note and "estimada" in note
def test_a_second_over_the_target_is_not_worth_a_rewrite():
"""El objetivo sigue siendo 45 s, pero la estimación tiene un segundo de
error por línea: avisar por medio segundo es avisar del estimador. Caso
real la sesión 168 salió a 45,4 s y se pagó una generación por ello."""
from src.generator.spec_contract import TARGET_GRACE, TARGET_MAX_DURATION
justo = spec_with(shot(duration=TARGET_MAX_DURATION + TARGET_GRACE - 0.1))
pasado = spec_with(shot(duration=TARGET_MAX_DURATION + TARGET_GRACE + 0.1))
assert editorial_notes(justo) == []
assert editorial_notes(pasado)
# Y el consejo se mide contra el objetivo, no contra el margen: se pide
# bajar hasta 45, no hasta 46,5.
assert "sobran 1.6s" in editorial_notes(pasado)[0]
def test_the_grace_works_at_both_ends():
from src.generator.spec_contract import TARGET_GRACE, TARGET_MIN_DURATION
assert editorial_notes(spec_with(shot(duration=TARGET_MIN_DURATION
- TARGET_GRACE + 0.1))) == []
assert editorial_notes(spec_with(shot(duration=TARGET_MIN_DURATION
- TARGET_GRACE - 0.1)))
def test_the_advice_says_how_much_to_cut_and_from_where():
""""Recorta narración" no dice cuánta, y las tres veces que saltó este aviso
el modelo devolvió un spec que seguía pasándose. El exceso va en palabras
porque es lo que el modelo escribe, y señalando el plano que más habla."""
doc = spec_with(shot(duration=4.0), shot(duration=4.0))
doc["shots"][0]["narration"] = "Short line."
doc["shots"][1]["narration"] = " ".join(["word"] * 200)
note = editorial_notes(doc)[0]
assert "palabras de narración" in note
assert "shots.1" in note and "shots.0" not in note
def test_the_advice_for_a_silent_spec_never_mentions_narration():
"""Sin voz, pedir que recorte narración es mandarlo a arreglar algo que no
existe: lo que sobra son duraciones declaradas."""
note = editorial_notes(spec_with(*[shot(duration=10.0) for _ in range(6)]))[0]
assert "narración" not in note and "duraciones declaradas" in note
def test_a_spec_without_narration_keeps_the_old_wording():
note = editorial_notes(spec_with(*[shot(duration=10.0) for _ in range(6)]))[0]
assert "duración total" in note and "estimada" not in note
+3 -109
View File
@@ -6,7 +6,6 @@ sube un vídeo a un canal de verdad, y eso no es algo que deba pasar por teclear
"""
import json
import aiohttp
import pytest
from src.generator import youtube as yt
@@ -76,9 +75,6 @@ class FakeSession:
def put(self, url, **kw):
return self._next("PUT", url)
def get(self, url, **kw):
return self._next("GET", url)
@pytest.fixture(autouse=True)
def clean_token_cache():
@@ -87,33 +83,14 @@ def clean_token_cache():
yt._token_cache.clear()
@pytest.fixture(autouse=True)
def no_recheck_delay(monkeypatch):
"""La segunda pasada de la comprobación espera 3 s en producción, que es lo
que tarda YouTube en indexar. Aquí no se espera a nada."""
monkeypatch.setattr(yt, "_VISIBILITY_RECHECK_DELAY", 0)
@pytest.fixture
def uploader():
return YouTubeUploader(client_id="cid", client_secret="secret",
refresh_token="refresh")
#: Lo que oEmbed contesta de un vídeo que no se ve sin sesión.
OEMBED_HIDDEN = FakeResp(404, body="Not Found")
#: Y de uno que sí.
OEMBED_VISIBLE = FakeResp(200, {"title": "JAL 1628", "type": "video"})
def patch(client, routes):
"""El servidor falso, con la comprobación de visibilidad ya enrutada.
`upload()` la hace siempre, así que todo test que suba pasa por oEmbed. Por
defecto contesta "no se ve", que es lo que se espera de un vídeo privado; el
test que quiera el caso malo pone su propia ruta `/oembed`.
"""
session = FakeSession({"/oembed": OEMBED_HIDDEN, **routes})
session = FakeSession(routes)
client._session = lambda total: session
return session
@@ -193,10 +170,8 @@ async def test_upload_does_metadata_then_bytes(uploader, video):
assert result.video_id == "abc123"
assert result.watch_url == "https://youtube.com/shorts/abc123"
assert result.studio_url.endswith("/abc123/edit")
# Token, metadatos, bytes, y la comprobación de visibilidad — que se
# reintenta porque el primer 404 puede ser YouTube todavía indexando.
assert [c[0] for c in session.calls] == ["POST", "POST", "PUT", "GET", "GET"]
assert len(seen) == 4, "cada etapa avisa: autenticar, abrir, subir, comprobar"
assert [c[0] for c in session.calls] == ["POST", "POST", "PUT"]
assert len(seen) == 3, "cada etapa avisa: autenticar, abrir, subir"
@pytest.mark.asyncio
@@ -364,84 +339,3 @@ def test_uploaded_video_urls():
video = UploadedVideo(video_id="xyz", title="t", privacy_status="private")
assert video.watch_url == "https://youtube.com/shorts/xyz"
assert video.studio_url == "https://studio.youtube.com/video/xyz/edit"
# --- la visibilidad, comprobada en vez de creída ----------------------------
@pytest.mark.asyncio
async def test_a_video_anyone_can_watch_is_detected(uploader, video):
"""El caso que existe para pillar: la API dice privado y el vídeo se ve.
Todo el flujo de revisión informe de fundamento primero, publicar después
descansa en que subir NO publique. Si eso deja de ser cierto hay que
enterarse por el parte de la subida, no por una visita al canal.
"""
patch(uploader, {
"/token": TOKEN_OK,
"/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/x"}),
"https://up/x": FakeResp(200, VIDEO_OK),
"/oembed": OEMBED_VISIBLE,
})
result = await uploader.upload(video, build_metadata(SPEC, "x"))
assert result.privacy_status == "private", "la API sigue diciendo privado"
assert result.reachable is True
assert result.visibility_contradiction
@pytest.mark.asyncio
async def test_a_private_video_reports_no_contradiction(uploader, video):
patch(uploader, {
"/token": TOKEN_OK,
"/upload/youtube": FakeResp(200, {}, headers={"Location": "https://up/x"}),
"https://up/x": FakeResp(200, VIDEO_OK),
})
result = await uploader.upload(video, build_metadata(SPEC, "x"))
assert result.reachable is False
assert not result.visibility_contradiction
@pytest.mark.asyncio
async def test_a_video_visible_only_on_the_second_look_still_counts(uploader):
"""Segundos después de subirlo, oEmbed devuelve 404 de un vídeo que sí se
ve: aún no está indexado. Un solo vistazo daría por privado justo el vídeo
que hay que gritar."""
session = patch(uploader, {"/oembed": [OEMBED_HIDDEN, OEMBED_VISIBLE]})
assert await uploader.reachable("abc123") is True
assert len(session.calls) == 2
@pytest.mark.asyncio
async def test_two_hidden_looks_are_enough_to_stop_asking(uploader):
session = patch(uploader, {"/oembed": OEMBED_HIDDEN})
assert await uploader.reachable("abc123") is False
assert len(session.calls) == 2
@pytest.mark.asyncio
async def test_a_network_failure_is_not_knowing_rather_than_privacy(uploader):
"""No se pudo comprobar NO es lo mismo que no se ve. Devolver False aquí
sería inventarse una garantía a partir de un fallo de red."""
class Broken:
async def __aenter__(self): return self
async def __aexit__(self, *a): return False
def get(self, url, **kw):
raise aiohttp.ClientError("sin red")
uploader._session = lambda total: Broken()
assert await uploader.reachable("abc123") is None
@pytest.mark.asyncio
async def test_an_upload_without_an_id_is_not_checked(uploader):
session = patch(uploader, {"/oembed": OEMBED_VISIBLE})
assert await uploader.reachable("") is None
assert session.calls == []