Privacy-Safe PII Redaction with a Local Ollama Model
A local Ollama model redacts names, emails, and phone numbers from a document and stores the sanitized version as an artifact before anything leaves the network.
Transcribe podcast episodes locally, draft show notes and chapters with AI, approve them, and publish a validated RSS feed to S3.
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.
new_episode (aws.s3.Trigger) picks up an .mp3 or .m4a in podcast/inbox/ and moves it to processing/; manual runs take an audio upload.probe (scripts.python.Script) rejects files that aren't audio and fingerprints the recording; an episode already published ends at skip_duplicate.transcribe runs faster-whisper on CPU (whisper_model, default base), so the audio never leaves your infrastructure.write_notes (ai.completion.ChatCompletion) drafts titles, summary, show notes, chapters and quotes; check_notes keeps 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.Upload stores the audio, podcast:chapters JSON and podcast:transcript VTT; build_feed adds the episode (or creates the feed), validates it, backs up the old feed and only then replaces it.{{ outputs.episode_title }}, {{ outputs.audio_url }}, {{ outputs.feed_url }} and {{ outputs.transcript }}.public_base_url; set the bucket, prefix and the show details under variables.S3_ACCESS_KEY_ID and S3_SECRET_KEY_ID: read the inbox and upload episodes and the feed.ANTHROPIC_API_KEY: drafts the notes from the transcript (model in variables.claude_model).SLACK_WEBHOOK_URL: previews, published and rejected notices, and failure alerts.audio (FILE), episode_title (STRING, optional) and dry_run (BOOL, default false) set to true; read the preview.new_episode.whisper_model for accents or noisy audio, or a GPU worker for long episodes.post_published.A local Ollama model redacts names, emails, and phone numbers from a document and stores the sanitized version as an artifact before anything leaves the network.
Audit running Ollama models, track GPU VRAM allocation vs RAM spilling, and trigger automated idle model eviction when memory headroom breaches thresholds.
A local Ollama model extracts typed JSON fields from unstructured text, a Python task validates them into scalar outputs, and Discord gets the verdict.
DuckDB computes daily metrics, the stored result flows into a local Ollama model that writes a plain-English standup summary, and Discord gets the pointer.
Audit local Ollama model storage footprint, flag capacity breaches, alert Slack, and safely prune stale weights.
A shell loop inside one OllamaCLI task classifies every line of an input file with a local model, producing a CSV artifact and a Slack count report.
Weekly pharmacy stock check in DuckDB that flags expired and soon-to-expire batches, bad stock records, and the money at risk, with optional Slack alerts.
Read-only check of an Orthanc DICOM server's jobs on a schedule, paging Slack when studies fail to reach the PACS, a CAD or triage service, or an archive, or transfers get stuck.
Check whether a PayFit payroll period has completed, alert Slack only once it is overdue past a deadline day, and log a governance audit trail.
A webhook catches each signup form submission, creates the person and a linked deal in Pipedrive, and confirms the new lead in Slack.
Verify OpenPGP digital signatures and MDC integrity on partner data payloads and gate unauthenticated ingestion.
A monthly flow lists Pipedrive persons, counts records missing emails or owners in Python, and posts the hygiene metrics to Discord.
