New to Kestra?
Use blueprints to kickstart your first workflows.
Podcast Episode Publisher that Transcribes Locally, Writes Show Notes and Chapters with AI, and Publishes a Validated RSS Feed
Transcribe podcast episodes locally, draft show notes and chapters with AI, approve them, and publish a validated RSS feed to S3.
Categories
AIBusiness
id: podcast-episode-publisher
namespace: company.team
description: |
Turn a finished podcast recording into a published episode: transcribe it
locally, draft show notes and chapters with an LLM, check them against the
transcript, get approval, and publish the audio and a validated RSS feed.
labels:
team: creator
use-case: podcast-publishing
variables:
claude_model: claude-sonnet-5-5
whisper_model: base
bucket: my-podcast
prefix: podcast/
s3_endpoint: https://s3.us-east-1.amazonaws.com
region: us-east-1
public_base_url: https://media.example.com
show:
title: Northwind Tech Talk
author: Alex Rivera
email: podcast@northwind.example
description: A weekly show about running servers at home without losing your weekends.
link: https://northwind.example/podcast
artwork: https://northwind.example/podcast/artwork.jpg
category: Technology
language: en
triggers:
- id: new_episode
type: io.kestra.plugin.aws.s3.Trigger
description: An .mp3 or .m4a dropped into podcast/inbox/ starts a run and is
moved to podcast/processing/. Shipped disabled; set the bucket variables
first.
interval: PT1M
disabled: true
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
prefix: "{{ vars.prefix }}inbox/"
regexp: ".*\\.(mp3|m4a)$"
maxKeys: 1
action: MOVE
moveTo:
key: "{{ vars.prefix }}processing/"
inputs:
- id: audio
type: FILE
displayName: Episode audio (.mp3 or .m4a, manual runs)
required: false
- id: episode_title
type: STRING
displayName: Episode title (optional, otherwise suggested by the model)
required: false
- id: dry_run
type: BOOL
displayName: Dry run
description: Transcribe and draft the notes only. No approval form, no upload,
no feed change.
defaults: false
tasks:
- id: require_audio
type: io.kestra.plugin.core.execution.Fail
condition: "{{ (inputs.audio ?? trigger.objects[0].uri ?? '') == '' }}"
errorMessage: Upload an episode file, or drop one into podcast/inbox/ with the
new_episode trigger enabled.
- id: probe
type: io.kestra.plugin.scripts.python.Script
description: Check that the file is real audio and long enough, and fingerprint
it so the same recording is never published twice.
dependencies:
- kestra
- mutagen
inputFiles:
audio.bin: "{{ inputs.audio ?? trigger.objects[0].uri }}"
name.b64: "{{ (trigger.objects[0].key ?? inputs.audio ?? 'episode') |
base64encode }}"
script: |
import base64, hashlib, os, re, sys
import mutagen
from kestra import Kestra
name = base64.b64decode(open("name.b64").read()).decode().rsplit("/", 1)[-1]
audio = mutagen.File("audio.bin")
if audio is None or not getattr(audio, "info", None) or not getattr(audio.info, "length", 0):
sys.exit(f"{name} is not a readable audio file")
duration = int(audio.info.length)
if duration < 20:
sys.exit(f"{name} is only {duration} seconds long")
kind = type(audio).__name__
ext, mime = ("m4a", "audio/x-m4a") if kind in ("MP4", "M4A") else ("mp3", "audio/mpeg")
Kestra.outputs({
"duration": duration, "size": os.path.getsize("audio.bin"), "ext": ext, "mime": mime,
"fingerprint": hashlib.sha256(open("audio.bin", "rb").read()).hexdigest()[:16],
"file_name": re.sub(r"[^\w. -]", "", name)[:120],
})
- id: already_published
type: io.kestra.plugin.core.flow.If
condition: "{{ kv('podcast-episode-' ~ outputs.probe.vars.fingerprint,
errorOnMissing=false) is not null }}"
then:
- id: skip_duplicate
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.probe.vars.file_name }} was already published as {{
kv('podcast-episode-' ~ outputs.probe.vars.fingerprint) }}"
- id: stop
type: io.kestra.plugin.core.execution.Exit
state: SUCCESS
- id: transcribe
type: io.kestra.plugin.scripts.python.Script
description: Transcribe locally with faster-whisper on CPU. The audio never
leaves your infrastructure; only the transcript text goes to the model.
dependencies:
- kestra
- faster-whisper
inputFiles:
audio.bin: "{{ inputs.audio ?? trigger.objects[0].uri }}"
outputFiles:
- transcript.txt
- transcript.vtt
- timed.txt
script: |
import av
from faster_whisper import WhisperModel
from kestra import Kestra
# faster-whisper 1.2 still passes metadata_errors, which current PyAV no longer accepts.
_open = av.open
av.open = lambda *args, metadata_errors=None, **kwargs: _open(*args, **kwargs)
model = WhisperModel("{{ vars.whisper_model }}", device="cpu", compute_type="int8")
segments, info = model.transcribe("audio.bin", vad_filter=True)
stamp = lambda s, sep=".": f"{int(s // 3600):02d}:{int(s % 3600 // 60):02d}:{int(s % 60):02d}{sep}{int(s % 1 * 1000):03d}"
text, vtt, timed = [], ["WEBVTT", ""], []
for seg in segments:
line = seg.text.strip()
text.append(line)
vtt += [f"{stamp(seg.start)} --> {stamp(seg.end)}", line, ""]
timed.append(f"[{int(seg.start // 60):02d}:{int(seg.start % 60):02d}] {line}")
open("transcript.txt", "w").write(" ".join(text))
open("transcript.vtt", "w").write("\n".join(vtt))
open("timed.txt", "w").write("\n".join(timed))
Kestra.outputs({"language": info.language, "words": len(" ".join(text).split()), "segments": len(timed)})
- id: write_notes
type: io.kestra.plugin.ai.completion.ChatCompletion
description: Draft titles, summary, show notes, chapters and quotes from the
timestamped transcript. Everything is checked against the transcript next.
provider:
type: io.kestra.plugin.ai.provider.Anthropic
apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
modelName: "{{ vars.claude_model }}"
configuration:
maxToken: 4000
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 3
messages:
- type: SYSTEM
content: |
You write podcast show notes from a timestamped transcript. Use only what is said in the episode; never invent facts, guests or links.
Respond with one JSON object and nothing else:
{"titles": ["...", "...", "..."], "summary": "...", "show_notes": ["...", "..."], "chapters": [{"start": "00:00", "title": "..."}], "quotes": ["..."], "keywords": ["..."]}
- titles: three episode title options, under 70 characters each.
- summary: two sentences for podcast apps.
- show_notes: 3 to 6 short bullet points of what listeners learn.
- chapters: start at 00:00, in order, each starting at a [mm:ss] timestamp from the transcript where a new topic begins, at least 30 seconds apart.
- quotes: up to 3 sentences copied word for word from the transcript.
- type: USER
content: "Episode transcript:
{{ read(outputs.transcribe.outputFiles['timed.txt']) }}"
- id: check_notes
type: io.kestra.plugin.scripts.python.Script
description: Validate the chapters against the episode length and keep only
quotes that really appear in the transcript.
dependencies:
- kestra
inputFiles:
transcript.txt: "{{ outputs.transcribe.outputFiles['transcript.txt'] }}"
answer.b64: "{{ outputs.write_notes.textOutput | base64encode }}"
title.b64: "{{ (inputs.episode_title ?? '') | base64encode }}"
outputFiles:
- notes.json
- chapters.json
script: |
import base64, html, json, re
from datetime import date
from difflib import SequenceMatcher
from kestra import Kestra
duration = int({{ outputs.probe.vars.duration }})
text = base64.b64decode(open("answer.b64").read()).decode()
match = re.search(r"\{.*\}", text, re.S)
notes = json.loads(match.group(0)) if match else {}
clean = lambda s, n=300: re.sub(r"[{}%#`|<>]", "", str(s or "")).strip()[:n]
norm = lambda s: re.sub(r"[^a-z0-9 ]", "", s.lower()).split()
transcript = norm(open("transcript.txt").read())
def in_transcript(quote):
# Allow a misheard word or two: the quote must match a stretch of the transcript at 85% or more.
words = norm(quote)
if len(words) < 5:
return False
return any(SequenceMatcher(None, words, transcript[i:i + len(words)]).ratio() >= 0.85
for i in range(0, max(len(transcript) - len(words) + 1, 1)))
issues = []
def seconds(stamp):
parts = [int(p) for p in str(stamp).split(":") if p.isdigit()]
return sum(p * 60 ** i for i, p in enumerate(reversed(parts))) if parts else -1
gap = max(5, min(30, duration // 10)) # minimum seconds between chapters, smaller for short episodes
chapters, last = [], -60
for c in notes.get("chapters") or []:
start = seconds(c.get("start"))
if 0 <= start < duration and start - last >= gap and clean(c.get("title"), 80):
chapters.append({"startTime": start, "title": clean(c.get("title"), 80)}); last = start
else:
issues.append(f"dropped chapter '{clean(c.get('title'), 60)}' at {c.get('start')}")
if not chapters or chapters[0]["startTime"] != 0:
chapters.insert(0, {"startTime": 0, "title": "Introduction"})
quotes = []
for q in notes.get("quotes") or []:
if in_transcript(str(q)):
quotes.append(clean(q, 300))
else:
issues.append(f"dropped a quote that is not in the transcript: '{clean(q, 80)}'")
title = base64.b64decode(open("title.b64").read()).decode().strip() or clean((notes.get("titles") or ["Untitled episode"])[0], 100)
mmss = lambda s: f"{s // 60:02d}:{s % 60:02d}"
bullets = [clean(b, 200) for b in notes.get("show_notes") or []][:6]
description = (f"<p>{html.escape(clean(notes.get('summary'), 600))}</p>"
+ ("<ul>" + "".join(f"<li>{html.escape(b)}</li>" for b in bullets) + "</ul>" if bullets else "")
+ "<p>Chapters:</p><ul>" + "".join(f"<li>{mmss(c['startTime'])} {html.escape(c['title'])}</li>" for c in chapters) + "</ul>"
+ "".join(f"<blockquote>{html.escape(q)}</blockquote>" for q in quotes))
json.dump({"version": "1.2.0", "chapters": chapters}, open("chapters.json", "w"))
json.dump({"title": title, "summary": clean(notes.get("summary"), 600), "description": description,
"keywords": [clean(k, 30) for k in notes.get("keywords") or []][:10]}, open("notes.json", "w"))
# The audio fingerprint keeps file names unique, so a new episode can never overwrite a published one.
slug = f"{date.today()}-" + (re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")[:60] or "episode") + "-{{ outputs.probe.vars.fingerprint }}"[:7]
Kestra.outputs({
"title": title, "slug": slug, "chapters": len(chapters), "quotes": len(quotes), "issues": issues,
"preview": f"🎙️ *{title}* ({duration // 60}:{duration % 60:02d}, {{ outputs.transcribe.vars.words }} words, {{ outputs.transcribe.vars.language }})\n"
f"{clean(notes.get('summary'), 600)}\n" + "\n".join(f"• {mmss(c['startTime'])} {c['title']}" for c in chapters)
+ (f"\nOther titles: {' / '.join(clean(t, 80) for t in (notes.get('titles') or [])[1:3])}" if len(notes.get("titles") or []) > 1 else "")
+ (f"\n⚠️ {'; '.join(issues)}" if issues else ""),
})
- id: post_preview
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "{{ (inputs.dry_run ?? false) ? '[dry run] ' : '' }}{{
outputs.check_notes.vars.preview }}{{ (inputs.dry_run ?? false) ? '' : '
Approve: ' ~ (kestra.url ?? '') ~ '/ui/main/executions/' ~ flow.namespace
~ '/' ~ flow.id ~ '/' ~ execution.id }}"
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
- id: review
type: io.kestra.plugin.core.flow.If
condition: "{{ not (inputs.dry_run ?? false) }}"
then:
- id: approve
type: io.kestra.plugin.core.flow.Pause
description: Listen-check the notes, then publish or reject. Without a decision
within three days the run is cancelled and nothing is published.
pauseDuration: P3D
behavior: CANCEL
onResume:
- id: decision
type: SELECT
displayName: Decision
values:
- Publish
- Reject
defaults: Publish
- id: title
type: STRING
displayName: Title override (optional)
- id: episode_number
type: INT
displayName: Episode number (optional)
- id: explicit
type: BOOL
displayName: Explicit content
defaults: false
- id: publish
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approve.onResume.decision == 'Publish' }}"
then:
- id: upload_audio
type: io.kestra.plugin.aws.s3.Upload
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}episodes/{{ outputs.check_notes.vars.slug }}.{{
outputs.probe.vars.ext }}"
from: "{{ inputs.audio ?? trigger.objects[0].uri }}"
contentType: "{{ outputs.probe.vars.mime }}"
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 4
- id: upload_extras
type: io.kestra.plugin.core.flow.Parallel
description: Chapters and transcript files that podcast apps link to from the
feed.
tasks:
- id: upload_chapters
type: io.kestra.plugin.aws.s3.Upload
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}episodes/{{ outputs.check_notes.vars.slug
}}.chapters.json"
from: "{{ outputs.check_notes.outputFiles['chapters.json'] }}"
contentType: application/json+chapters
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 4
- id: upload_transcript
type: io.kestra.plugin.aws.s3.Upload
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}episodes/{{ outputs.check_notes.vars.slug }}.vtt"
from: "{{ outputs.transcribe.outputFiles['transcript.vtt'] }}"
contentType: text/vtt
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 4
- id: find_feed
type: io.kestra.plugin.aws.s3.List
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
prefix: "{{ vars.prefix }}feed.xml"
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
- id: has_feed
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.find_feed.objects ?? []) | length > 0 }}"
then:
- id: fetch_feed
type: io.kestra.plugin.aws.s3.Download
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}feed.xml"
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
- id: build_feed
type: io.kestra.plugin.scripts.python.Script
description: Add the episode to the feed (or create the feed), then validate it
before anything is uploaded.
dependencies:
- kestra
inputFiles:
old_feed.xml: "{{ outputs.fetch_feed.uri ?? '' }}"
notes.json: "{{ outputs.check_notes.outputFiles['notes.json'] }}"
answer.b64: "{{ outputs.approve.onResume | toJson | base64encode }}"
outputFiles:
- feed.xml
script: |
import base64, json, os, re
import xml.etree.ElementTree as ET
from email.utils import formatdate
from kestra import Kestra
NS = {"itunes": "http://www.itunes.com/dtds/podcast-1.0.dtd", "podcast": "https://podcastindex.org/namespace/1.0",
"atom": "http://www.w3.org/2005/Atom", "content": "http://purl.org/rss/1.0/modules/content/"}
for prefix, uri in NS.items():
ET.register_namespace(prefix, uri)
qn = lambda ns, tag: "{" + NS[ns] + "}" + tag # ElementTree's namespaced tag form
it = lambda tag: qn("itunes", tag)
pc = lambda tag: qn("podcast", tag)
show = {{ vars.show | toJson }}
base = "{{ vars.public_base_url }}".rstrip("/") + "/{{ vars.prefix }}"
notes = json.load(open("notes.json"))
answer = json.loads(base64.b64decode(open("answer.b64").read()))
slug, ext = "{{ outputs.check_notes.vars.slug }}", "{{ outputs.probe.vars.ext }}"
def sub(parent, tag, value=None, **attrs):
el = ET.SubElement(parent, tag, {k: str(v) for k, v in attrs.items()})
if value is not None:
el.text = str(value)
return el
if os.path.getsize("old_feed.xml"):
tree = ET.parse("old_feed.xml"); channel = tree.getroot().find("channel")
else:
rss = ET.Element("rss", {"version": "2.0"}); channel = sub(rss, "channel"); tree = ET.ElementTree(rss)
sub(channel, "title", show["title"]); sub(channel, "link", show["link"]); sub(channel, "language", show["language"])
sub(channel, "description", show["description"]); sub(channel, it("author"), show["author"])
owner = sub(channel, it("owner")); sub(owner, it("name"), show["author"]); sub(owner, it("email"), show["email"])
sub(channel, it("image"), href=show["artwork"]); sub(channel, it("category"), text=show["category"]) # Apple expects the category as an attribute
sub(channel, it("explicit"), "false")
sub(channel, qn("atom", "link"), href=base + "feed.xml", rel="self", type="application/rss+xml")
guid = f"{show['link'].rstrip('/')}/{slug}"
if any((i.findtext("guid") or "") == guid for i in channel.findall("item")):
raise SystemExit(f"an episode with guid {guid} is already in the feed")
item = ET.Element("item")
title = (answer.get("title") or "").strip() or notes["title"]
sub(item, "title", re.sub(r"[{}%#`<>]", "", title)[:120]); sub(item, "description", notes["description"])
sub(item, qn("content", "encoded"), notes["description"])
sub(item, "guid", guid, isPermaLink="false"); sub(item, "pubDate", formatdate(usegmt=True))
sub(item, "enclosure", url=f"{base}episodes/{slug}.{ext}", length=int({{ outputs.probe.vars.size }}), type="{{ outputs.probe.vars.mime }}")
d = int({{ outputs.probe.vars.duration }}); sub(item, it("duration"), f"{d // 3600:02d}:{d % 3600 // 60:02d}:{d % 60:02d}")
sub(item, it("summary"), notes["summary"]); sub(item, it("explicit"), "true" if answer.get("explicit") else "false")
if answer.get("episode_number"):
sub(item, it("episode"), int(answer["episode_number"]))
if notes["keywords"]:
sub(item, it("keywords"), ",".join(notes["keywords"]))
sub(item, pc("chapters"), url=f"{base}episodes/{slug}.chapters.json", type="application/json+chapters")
sub(item, pc("transcript"), url=f"{base}episodes/{slug}.vtt", type="text/vtt")
items = channel.findall("item"); position = list(channel).index(items[0]) if items else len(channel)
channel.insert(position, item)
ET.indent(tree); tree.write("feed.xml", encoding="utf-8", xml_declaration=True)
# Validate the result before it replaces the live feed.
check = ET.parse("feed.xml").getroot().find("channel")
problems = [t for t in ("title", "link", "description", "language") if not check.findtext(t)]
guids = [i.findtext("guid") for i in check.findall("item")]
for i in check.findall("item"):
enc = i.find("enclosure")
if not i.findtext("title") or enc is None or not enc.get("url") or not (enc.get("length") or "").isdigit() or not enc.get("type"):
problems.append(f"item '{i.findtext('title')}' misses a title or a complete enclosure")
if len(guids) != len(set(guids)):
problems.append("duplicate guids")
if problems:
raise SystemExit("feed validation failed: " + "; ".join(problems))
Kestra.outputs({"episodes": len(guids), "guid": guid, "title": title, "audio_url": f"{base}episodes/{slug}.{ext}", "feed_url": f"{base}feed.xml"})
- id: backup_feed
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.find_feed.objects ?? []) | length > 0 }}"
then:
- id: upload_backup
type: io.kestra.plugin.aws.s3.Upload
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}backups/feed-{{ execution.startDate |
date('yyyyMMdd-HHmmss') }}.xml"
from: "{{ outputs.fetch_feed.uri }}"
contentType: application/rss+xml
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
- id: upload_feed
type: io.kestra.plugin.aws.s3.Upload
description: Replace the live feed only now that the new one validated and every
file it links to is uploaded.
endpointOverride: "{{ vars.s3_endpoint }}"
forcePathStyle: true
compatibilityMode: true
region: "{{ vars.region }}"
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ vars.bucket }}"
key: "{{ vars.prefix }}feed.xml"
from: "{{ outputs.build_feed.outputFiles['feed.xml'] }}"
contentType: application/rss+xml
cacheControl: max-age=300
retry:
type: exponential
interval: PT10S
maxInterval: PT2M
maxAttempts: 4
- id: remember
type: io.kestra.plugin.core.kv.Set
key: "podcast-episode-{{ outputs.probe.vars.fingerprint }}"
value: "{{ outputs.build_feed.vars.guid }}"
- id: post_published
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "✅ Published *{{ outputs.build_feed.vars.title }}* ({{
outputs.build_feed.vars.episodes }} episodes in the feed). Audio:
{{ outputs.build_feed.vars.audio_url }} · Feed: {{
outputs.build_feed.vars.feed_url }}"
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
else:
- id: post_rejected
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "↩️ Not published: {{ outputs.check_notes.vars.title }}. The
recording stays in {{ vars.prefix }}processing/."
retry:
type: exponential
interval: PT10S
maxInterval: PT1M
maxAttempts: 3
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert with the failed task, the error and a link. The live feed is
only replaced after the new one validates, so a failure never breaks it.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
❌ *Podcast episode publisher FAILED* (flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}).
Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
{{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
outputs:
- id: episode_title
type: STRING
displayName: Episode title
value: "{{ outputs.build_feed.vars.title ?? outputs.check_notes.vars.title ??
'none' }}"
- id: audio_url
type: STRING
displayName: Public audio URL
value: "{{ outputs.build_feed.vars.audio_url ?? 'not published' }}"
- id: feed_url
type: STRING
displayName: Feed URL
value: "{{ outputs.build_feed.vars.feed_url ?? 'not published' }}"
- id: transcript
type: STRING
displayName: Transcript (internal storage URI)
value: "{{ outputs.transcribe.outputFiles['transcript.txt'] ?? 'none' }}"
Independent podcasters spend hours per episode on transcripts, show notes and chapter timestamps, then hand-edit an RSS feed where one wrong tag or file size stops Apple and Spotify from updating. This flow turns a finished recording into a published episode with a human approval step.
How it works
new_episode(aws.s3.Trigger) picks up an.mp3or.m4ainpodcast/inbox/and moves it toprocessing/; manual runs take anaudioupload.probe(scripts.python.Script) rejects files that aren't audio and fingerprints the recording; an episode already published ends atskip_duplicate.transcriberuns faster-whisper on CPU (whisper_model, defaultbase), so the audio never leaves your infrastructure.write_notes(ai.completion.ChatCompletion) drafts titles, summary, show notes, chapters and quotes;check_noteskeeps only chapters inside the episode and quotes found word for word in the transcript.approve(Pause) publishes or rejects, with an optional title, episode number and explicit flag; no answer in 3 days cancels the run.aws.s3.Uploadstores the audio,podcast:chaptersJSON andpodcast:transcriptVTT;build_feedadds the episode (or creates the feed), validates it, backs up the old feed and only then replaces it.
What you get
- A transcript, show notes, chapters and quotes in about the length of the episode, on CPU.
- An RSS feed with exact enclosure sizes, durations and Podcasting 2.0 chapters and transcripts.
- Unique file names per recording, so a new episode never overwrites a published one.
- Outputs
{{ outputs.episode_title }},{{ outputs.audio_url }},{{ outputs.feed_url }}and{{ outputs.transcript }}.
Who it's for
- Independent podcasters, and teams publishing a company podcast.
Why orchestrate this with Kestra
- Transcription, AI, approval and uploads in one YAML flow with retries and a failure alert.
- The live feed is replaced only after the new one validates, so a failed run never breaks it.
Prerequisites
- An S3 bucket (or S3-compatible store) served publicly at
public_base_url; set the bucket, prefix and theshowdetails undervariables. - Docker on the worker; the first run downloads the Whisper model.
Secrets
S3_ACCESS_KEY_IDandS3_SECRET_KEY_ID: read the inbox and upload episodes and the feed.ANTHROPIC_API_KEY: drafts the notes from the transcript (model invariables.claude_model).SLACK_WEBHOOK_URL: previews, published and rejected notices, and failure alerts.
Quick start
- Set the variables and secrets.
- Execute with
audio(FILE),episode_title(STRING, optional) anddry_run(BOOL, defaultfalse) set totrue; read the preview. - Execute without the dry run, approve, and open the feed URL in a podcast validator.
- Submit the feed URL to podcast directories once, then enable
new_episode.
How to extend
- Use a larger
whisper_modelfor accents or noisy audio, or a GPU worker for long episodes. - Post the episode to social media or a newsletter after
post_published.
Links
- AWS S3 plugin
- AI plugin
- Pause task
- faster-whisper
- Podcasting 2.0 namespace
- Apple Podcasts RSS requirements
- This blueprint has been created by Biplab Bera
Tasks
Create a Blueprint
Go to the developer platformMore AI Blueprints
More Business Blueprints
