Recover real episode metadata on --resume instead of rebuilding from the slug

A --resume run rebuilt the episode title from the Castopod URL slug, which is
lowercased and stripped of punctuation, so episode 58 went to YouTube as
"Episode 58: Rayfield S Nephew The Marfa Lights And Why Nobody Believes Concho"
with a placeholder description. save_chapters() only persisted chapters, so
resume had no real metadata to fall back on.

Metadata is now written to <audio>.metadata.json during a normal run and read
back on resume. If that file is missing (episodes published before this change)
the title and description are read from Castopod instead. Slug reconstruction
survives only as a last resort: it warns loudly, flags the result, and anchors
the episode prefix strip so it no longer eats the phrase mid-title.

Reading back from Castopod needed its own fix — TO_BASE64 wraps every 76 chars
and mysql renders those breaks as a literal backslash-n, which made b64decode
throw and the lookup silently return None.

Also includes two changes that were already sitting in the working tree: the
YouTube tag budget fix that episode 58's invalidTags failure prompted (with its
test), and a metadata model bump to claude-haiku-4.5.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-14 02:54:29 -05:00
co-authored by Claude Opus 5
parent 1c7ac334b4
commit 00ddc2d1b4
3 changed files with 363 additions and 13 deletions
+144 -13
View File
@@ -321,7 +321,7 @@ Respond with ONLY valid JSON, no markdown or explanation."""
"Content-Type": "application/json" "Content-Type": "application/json"
}, },
json={ json={
"model": "anthropic/claude-3.5-haiku", "model": "anthropic/claude-haiku-4.5",
"messages": [{"role": "user", "content": prompt}], "messages": [{"role": "user", "content": prompt}],
"temperature": 0.7 "temperature": 0.7
}, },
@@ -571,6 +571,107 @@ def save_chapters(metadata: dict, output_path: str):
print(f" Chapters saved to: {output_path}") print(f" Chapters saved to: {output_path}")
def save_metadata(metadata: dict, output_path: str):
"""Persist generated metadata so a --resume run can recover it losslessly.
The slug is not a safe source to rebuild a title from: it is lowercased and
stripped of punctuation, so apostrophes and commas cannot be recovered.
"""
keep = {k: metadata.get(k) for k in
("title", "description", "thumbnail_text", "chapters")}
with open(output_path, "w") as f:
json.dump(keep, f, indent=2)
print(f" Metadata saved to: {output_path}")
def _decode_db_row(output: str) -> dict | None:
"""Decode a TO_BASE64 payload from mysql batch output.
TO_BASE64 wraps every 76 characters and mysql renders those breaks as a
literal backslash-n, so the raw output is not directly decodable.
"""
cleaned = re.sub(r"\\n|\s", "", output or "")
if not cleaned:
return None
try:
return json.loads(base64.b64decode(cleaned).decode())
except Exception:
return None
def _fetch_episode_metadata_from_db(episode_number: int) -> dict | None:
"""Read back the title/description Castopod already stored for this episode.
Base64 keeps the round trip safe — descriptions contain newlines and quotes
that mysql's batch output would otherwise escape.
"""
cmd = (f'{DOCKER_PATH} exec {MARIADB_CONTAINER} mysql --defaults-extra-file=/tmp/.my.cnf '
f'-u {DB_USER} {DB_NAME} -N -e '
f'''"SELECT TO_BASE64(JSON_OBJECT('title', title, 'description', description_markdown)) '''
f'FROM cp_episodes WHERE number = {episode_number} LIMIT 1;"')
success, output = run_ssh_command(cmd)
if not success:
return None
row = _decode_db_row(output)
return row if row and row.get("title") else None
def recover_metadata(episode_number: int, slug: str, metadata_path, chapters: list,
db_lookup=None) -> dict:
"""Rebuild metadata for a --resume run, preferring lossless sources.
1. the metadata file written by the original run
2. the title/description Castopod already stored
3. the slug — which cannot round-trip punctuation or capitalization
"""
metadata_path = Path(metadata_path)
title = description = thumbnail_text = None
reconstructed = False
if metadata_path.exists():
try:
saved = json.loads(metadata_path.read_text())
title = saved.get("title")
description = saved.get("description")
thumbnail_text = saved.get("thumbnail_text")
if title:
print(f" Recovered metadata from {metadata_path.name}")
except (json.JSONDecodeError, OSError) as e:
print(f" Warning: could not read {metadata_path.name}: {e}")
if not title:
lookup = db_lookup or _fetch_episode_metadata_from_db
row = lookup(episode_number)
if row and row.get("title"):
title = row["title"]
description = description or row.get("description")
print(" Recovered title/description from Castopod")
if not title:
title_part = re.sub(rf"^episode-{episode_number}-", "", slug).replace("-", " ").title()
title = f"Episode {episode_number}: {title_part}"
reconstructed = True
print(" WARNING: rebuilding the title from the slug is lossy — "
"apostrophes, commas and capitalization cannot be recovered.")
print(f' WARNING: got "{title}" — pass --title to override.')
if not description:
description = f"Episode {episode_number} of Luke at the Roost."
if not thumbnail_text:
thumbnail_text = re.sub(rf"^Episode {episode_number}:\s*", "", title).upper()[:30]
meta = {
"title": title,
"description": description,
"chapters": chapters,
"thumbnail_text": thumbnail_text,
}
if reconstructed:
meta["title_is_reconstructed"] = True
return meta
def run_ssh_command(command: str, timeout: int = 30) -> tuple[bool, str]: def run_ssh_command(command: str, timeout: int = 30) -> tuple[bool, str]:
"""Run a command on the NAS via SSH.""" """Run a command on the NAS via SSH."""
ssh_cmd = [ ssh_cmd = [
@@ -1272,8 +1373,24 @@ def _check_youtube_duplicate(youtube, title: str) -> str | None:
return None return None
# YouTube rejects the entire upload with `invalidTags` if the combined tag
# string exceeds 500 chars. Tags containing a space are quoted by YouTube and
# those quotes count, so budget conservatively and leave a small margin.
YOUTUBE_TAG_BUDGET = 480
YOUTUBE_MAX_TAGS = 25
def _youtube_tag_cost(tag: str) -> int:
return len(tag) + (2 if " " in tag else 0)
def _extract_youtube_tags(metadata: dict) -> list[str]: def _extract_youtube_tags(metadata: dict) -> list[str]:
"""Extract dynamic tags from episode metadata for YouTube SEO.""" """Extract dynamic tags from episode metadata for YouTube SEO.
Base tags come first so they survive trimming; chapter titles fill whatever
budget is left. Anything that would bust the budget is skipped rather than
truncated, so no tag is ever emitted half-formed.
"""
base_tags = ["podcast", "Luke at the Roost", "talk radio", "call-in show", base_tags = ["podcast", "Luke at the Roost", "talk radio", "call-in show",
"talk show", "comedy", "AI podcast", "late night radio", "advice"] "talk show", "comedy", "AI podcast", "late night radio", "advice"]
skip = {"intro", "outro", "opening", "closing", "wrap up", "wrap-up"} skip = {"intro", "outro", "opening", "closing", "wrap up", "wrap-up"}
@@ -1284,7 +1401,21 @@ def _extract_youtube_tags(metadata: dict) -> list[str]:
continue continue
if len(title) <= 50: if len(title) <= 50:
dynamic.append(title) dynamic.append(title)
return (base_tags + dynamic)[:25]
tags: list[str] = []
used = 0
for tag in base_tags + dynamic:
tag = tag.replace("<", "").replace(">", "").strip()
if len(tag) < 3 or tag in tags:
continue
cost = _youtube_tag_cost(tag) + (1 if tags else 0) # +1 for the comma
if used + cost > YOUTUBE_TAG_BUDGET:
continue
tags.append(tag)
used += cost
if len(tags) >= YOUTUBE_MAX_TAGS:
break
return tags
def upload_to_youtube(audio_path: str, metadata: dict, chapters: list, def upload_to_youtube(audio_path: str, metadata: dict, chapters: list,
@@ -1513,6 +1644,7 @@ def main():
# Load existing metadata from files saved during first run # Load existing metadata from files saved during first run
transcript_path = audio_path.with_suffix(".transcript.txt") transcript_path = audio_path.with_suffix(".transcript.txt")
chapters_path = audio_path.with_suffix(".chapters.json") chapters_path = audio_path.with_suffix(".chapters.json")
metadata_path = audio_path.with_suffix(".metadata.json")
srt_path = audio_path.with_suffix(".srt") srt_path = audio_path.with_suffix(".srt")
# Build metadata from existing files; re-transcribe + gen only if missing # Build metadata from existing files; re-transcribe + gen only if missing
@@ -1520,21 +1652,16 @@ def main():
print(" Loading existing chapters and metadata files") print(" Loading existing chapters and metadata files")
with open(chapters_path) as f: with open(chapters_path) as f:
chapters_data = json.load(f) chapters_data = json.load(f)
# Reconstruct metadata from slug and chapters chapters = (chapters_data if isinstance(chapters_data, list)
slug = episode["slug"] else chapters_data.get("chapters", []))
# Convert slug to title: "episode-49-foo-bar" -> "Episode 49: Foo Bar" metadata = recover_metadata(episode_number, episode["slug"],
title_part = slug.replace(f"episode-{episode_number}-", "").replace("-", " ").title() metadata_path, chapters)
metadata = {
"title": f"Episode {episode_number}: {title_part}",
"description": f"Episode {episode_number} of Luke at the Roost.",
"chapters": chapters_data if isinstance(chapters_data, list) else chapters_data.get("chapters", []),
"thumbnail_text": title_part.upper()[:30],
}
else: else:
print("[1/5] Re-transcribing audio for metadata...") print("[1/5] Re-transcribing audio for metadata...")
transcript = transcribe_audio(str(audio_path)) transcript = transcribe_audio(str(audio_path))
metadata = generate_metadata(transcript, episode_number) metadata = generate_metadata(transcript, episode_number)
save_chapters(metadata, str(chapters_path)) save_chapters(metadata, str(chapters_path))
save_metadata(metadata, str(metadata_path))
if not transcript_path.exists(): if not transcript_path.exists():
labeled_text = label_transcript_speakers(transcript["full_text"]) labeled_text = label_transcript_speakers(transcript["full_text"])
with open(transcript_path, "w") as f: with open(transcript_path, "w") as f:
@@ -1608,6 +1735,10 @@ def main():
chapters_path = audio_path.with_suffix(".chapters.json") chapters_path = audio_path.with_suffix(".chapters.json")
save_chapters(metadata, str(chapters_path)) save_chapters(metadata, str(chapters_path))
# Save metadata so --resume recovers the real title, not a slug rebuild
metadata_path = audio_path.with_suffix(".metadata.json")
save_metadata(metadata, str(metadata_path))
# Save transcript text file with LUKE:/CALLER: speaker labels # Save transcript text file with LUKE:/CALLER: speaker labels
transcript_path = audio_path.with_suffix(".transcript.txt") transcript_path = audio_path.with_suffix(".transcript.txt")
raw_text = transcript["full_text"] raw_text = transcript["full_text"]
+140
View File
@@ -0,0 +1,140 @@
"""A --resume run must not invent the episode title.
Episode 58's first run published to Castopod, then died on the YouTube upload
(see test_youtube_tags.py). The re-run with --resume rebuilt the title from the
URL slug, which is lowercase and punctuation-free, so
"Episode 58: Rayfield's Nephew, the Marfa Lights, and Why Nobody Believes Concho"
went to YouTube as
"Episode 58: Rayfield S Nephew The Marfa Lights And Why Nobody Believes Concho"
with the description replaced by a placeholder.
"""
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from publish_episode import recover_metadata, save_metadata, _decode_db_row
EP58_TITLE = "Episode 58: Rayfield's Nephew, the Marfa Lights, and Why Nobody Believes Concho"
EP58_SLUG = "episode-58-rayfield-s-nephew-the-marfa-lights-and-why-nobody-believes-concho"
EP58_DESC = "Rayfield calls about his nephew and the catalytic converters."
CHAPTERS = [{"startTime": 0, "title": "Intro"}]
def test_metadata_file_roundtrips_losslessly(tmp_path):
path = tmp_path / "ep58.metadata.json"
save_metadata({"title": EP58_TITLE, "description": EP58_DESC,
"chapters": CHAPTERS, "thumbnail_text": "RAYFIELD"}, str(path))
meta = recover_metadata(58, EP58_SLUG, path, CHAPTERS)
assert meta["title"] == EP58_TITLE
assert meta["description"] == EP58_DESC
def test_metadata_file_wins_over_slug(tmp_path):
"""The exact ep58 regression: apostrophes, commas and case must survive."""
path = tmp_path / "ep58.metadata.json"
save_metadata({"title": EP58_TITLE, "description": EP58_DESC,
"chapters": CHAPTERS, "thumbnail_text": "RAYFIELD"}, str(path))
meta = recover_metadata(58, EP58_SLUG, path, CHAPTERS)
assert "Rayfield S Nephew" not in meta["title"]
assert "Rayfield's Nephew" in meta["title"]
assert "the Marfa Lights" in meta["title"]
assert meta["description"] != "Episode 58 of Luke at the Roost."
def test_falls_back_to_castopod_db_when_file_missing(tmp_path):
path = tmp_path / "missing.metadata.json"
calls = []
def db_lookup(episode_number):
calls.append(episode_number)
return {"title": EP58_TITLE, "description": EP58_DESC}
meta = recover_metadata(58, EP58_SLUG, path, CHAPTERS, db_lookup=db_lookup)
assert calls == [58]
assert meta["title"] == EP58_TITLE
assert meta["description"] == EP58_DESC
def test_slug_fallback_only_when_file_and_db_unavailable(tmp_path, capsys):
path = tmp_path / "missing.metadata.json"
meta = recover_metadata(58, EP58_SLUG, path, CHAPTERS,
db_lookup=lambda n: None)
assert meta["title"].startswith("Episode 58: ")
warning = capsys.readouterr().out
assert "lossy" in warning.lower() or "warning" in warning.lower()
def test_slug_fallback_does_not_title_case_away_real_words(tmp_path):
"""Even degraded, the fallback should not be silently trusted."""
meta = recover_metadata(58, EP58_SLUG, tmp_path / "nope.json", CHAPTERS,
db_lookup=lambda n: None)
assert meta.get("title_is_reconstructed") is True
def test_episode_prefix_stripped_only_at_start(tmp_path):
"""A global replace would also eat the phrase inside the title."""
slug = "episode-5-the-episode-5-mixup"
meta = recover_metadata(5, slug, tmp_path / "nope.json", CHAPTERS,
db_lookup=lambda n: None)
assert meta["title"].lower().count("episode 5") == 2
def test_chapters_come_from_chapters_json_not_metadata_file(tmp_path):
path = tmp_path / "ep58.metadata.json"
save_metadata({"title": EP58_TITLE, "description": EP58_DESC,
"chapters": [{"startTime": 999, "title": "Stale"}],
"thumbnail_text": "X"}, str(path))
meta = recover_metadata(58, EP58_SLUG, path, CHAPTERS)
assert meta["chapters"] == CHAPTERS
def test_decodes_base64_wrapped_by_mariadb():
"""TO_BASE64 wraps at 76 chars and mysql renders those breaks as a literal
backslash-n, which b64decode rejects outright."""
import base64 as b64
payload = json.dumps({"title": EP58_TITLE, "description": EP58_DESC})
encoded = b64.b64encode(payload.encode()).decode()
wrapped = "\\n".join(encoded[i:i + 76] for i in range(0, len(encoded), 76))
assert "\\n" in wrapped, "test needs a payload long enough to wrap"
row = _decode_db_row(wrapped)
assert row["title"] == EP58_TITLE
assert row["description"] == EP58_DESC
def test_decodes_base64_with_real_newlines():
import base64 as b64
payload = json.dumps({"title": EP58_TITLE, "description": EP58_DESC})
encoded = b64.b64encode(payload.encode()).decode()
wrapped = "\n".join(encoded[i:i + 76] for i in range(0, len(encoded), 76))
assert _decode_db_row(wrapped)["title"] == EP58_TITLE
def test_decode_db_row_returns_none_on_garbage():
assert _decode_db_row("") is None
assert _decode_db_row("not base64 at all !!!") is None
def test_save_metadata_keeps_only_publishable_fields(tmp_path):
path = tmp_path / "m.json"
save_metadata({"title": EP58_TITLE, "description": EP58_DESC,
"chapters": CHAPTERS, "thumbnail_text": "RAYFIELD",
"transcript": "huge blob that should not be persisted"}, str(path))
saved = json.loads(path.read_text())
assert "transcript" not in saved
assert saved["title"] == EP58_TITLE
+79
View File
@@ -0,0 +1,79 @@
"""YouTube rejects the whole upload with `invalidTags` if the tag list busts
its 500-character budget. Any tag containing a space gets wrapped in quotes and
those quotes count, so the naive sum of tag lengths understates the real cost.
Episode 58 failed here after a 284 MB upload had already completed.
"""
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from publish_episode import _extract_youtube_tags, YOUTUBE_TAG_BUDGET
def tag_cost(tags):
"""Mirror YouTube's accounting: quotes around multi-word tags, plus commas."""
return sum(len(t) + (2 if " " in t else 0) for t in tags) + max(0, len(tags) - 1)
# The real chapter titles from episode 58, which produced a 534-char tag list.
EP58_CHAPTERS = [
{"title": "Intro & Election Fraud Voicemail"},
{"title": "Rayfield: Nephew Stealing Catalytic Converters"},
{"title": "Suki & the Marfa Lights"},
{"title": "Aurora Toothbrush Sponsor"},
{"title": "Gus: Partner's Unlogged Stop at Meth House"},
{"title": "Merritt: Son Stealing from Family Store"},
{"title": "Concho: Mysterious Water Source Surveyors"},
{"title": "Desmond: Colleague's Research Misconduct"},
{"title": "Iron Heart Survival School Sponsor"},
{"title": "Fern: Donald Judd's Aluminum Boxes in Marfa"},
{"title": "Outro & Show Reflection"},
]
def test_episode_58_tags_fit_the_budget():
"""The exact input that broke the ep58 upload."""
tags = _extract_youtube_tags({"chapters": EP58_CHAPTERS})
assert tag_cost(tags) <= YOUTUBE_TAG_BUDGET, (
f"{tag_cost(tags)} chars > {YOUTUBE_TAG_BUDGET} budget: {tags}"
)
def test_pathological_long_titles_still_fit():
chapters = [{"title": "A" * 49 + f" {i}"} for i in range(40)]
tags = _extract_youtube_tags({"chapters": chapters})
assert tag_cost(tags) <= YOUTUBE_TAG_BUDGET, tag_cost(tags)
assert len(tags) <= 25
def test_no_chapters_still_returns_base_tags():
tags = _extract_youtube_tags({"chapters": []})
assert tags, "should still return the base SEO tags"
assert tag_cost(tags) <= YOUTUBE_TAG_BUDGET
def test_base_tags_are_prioritised_over_chapter_titles():
"""Base SEO tags matter more than chapter titles; they must survive trimming."""
chapters = [{"title": "B" * 48 + f" {i}"} for i in range(30)]
tags = _extract_youtube_tags({"chapters": chapters})
assert "podcast" in tags
assert "Luke at the Roost" in tags
def test_angle_brackets_are_stripped():
"""YouTube rejects tags containing < or >."""
chapters = [{"title": "Weird <script> Chapter"}]
tags = _extract_youtube_tags({"chapters": chapters})
assert not any("<" in t or ">" in t for t in tags), tags
def test_skips_intro_outro_and_short_titles():
chapters = [{"title": "Intro"}, {"title": "Outro"}, {"title": "ab"},
{"title": "A Real Chapter Title"}]
tags = _extract_youtube_tags({"chapters": chapters})
assert "Intro" not in tags and "Outro" not in tags and "ab" not in tags
assert "A Real Chapter Title" in tags