Build
The capstone project
Every node works in isolation. Today they run as one chain, unattended, from a raw topic to a video sitting on YouTube (or in a local folder if you're not ready to publish) — 3 times, producing 3 real videos.
Uploading with real metadata, not placeholder text
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
def upload_to_youtube(video_path: str, title: str, description: str, tags: list[str]) -> str:
youtube = build("youtube", "v3", credentials=get_refreshed_credentials())
request = youtube.videos().insert(
part="snippet,status",
body={
"snippet": {"title": title, "description": description, "tags": tags, "categoryId": "22"},
"status": {"privacyStatus": "private"}, # start private, publish manually after review
},
media_body=MediaFileUpload(video_path),
)
response = request.execute()
return response["id"]
privacyStatus: "private" is deliberate for a course context — every uploaded video starts private so you review it before anyone else can see it, rather than trusting the pipeline to have produced something publish-ready on the first automated run. Title/description/tags should be generated by Claude from the script content (a small additional prompt), not hardcoded placeholders.
OAuth token refresh: the part everyone finds annoying
YouTube OAuth tokens expire -- an unattended pipeline needs a refresh strategy, not a one-time login
A token you authorize once in a browser expires; without handling refresh, your pipeline works today and mysteriously fails in a week with an auth error that looks like something else broke. Store the refresh token (not just the access token) from your initial OAuth flow, and use it to silently get a new access token before each upload — this is annoying to set up once and then invisible forever after.
The orchestrator: the actual product
def run_full_pipeline(niche_keyword: str):
topics = find_topics(count=1)
if not topics:
print("No new topics found")
return
topic = topics[0]
script_result = generate_script(topic["title"], anthropic_client)
if not script_result["valid"]:
print(f"Script out of duration range, skipping: {topic['title']}")
return
voiceover = generate_voiceover(script_result["script"], "voiceover.mp3")
broll_clips = fetch_broll_for_script(script_result["script"], niche_keyword)
video_path = assemble_full_video(broll_clips, voiceover["output_path"], "final_video.mp4")
metadata = generate_video_metadata(script_result["script"], anthropic_client)
video_id = upload_to_youtube(video_path, **metadata)
print(f"Published (private): https://youtube.com/watch?v={video_id}")
Notice each stage checks its own success before continuing (if not topics, if not script_result["valid"]) -- this is the same "fail explicitly, don't cascade a bad input forward" instinct from Course 03's factory.py, applied to a 7-node pipeline instead of a 6-generator one.
You run the full pipeline 3 times to produce your 3 required videos. The 2nd run fails at the B-roll fetch step (Pexels rate limit). What should happen to that run?
This is the same principle from every earlier "handle failure explicitly" lesson in this curriculum: a failure partway through the pipeline should stop that run and report clearly, not silently produce and upload a broken or incomplete video. The topic can go back in the queue for a later, successful run instead.
Ship it
Run the full pipeline 3 times. Review all 3 videos before publishing (or share as private-link previews). Post one to Discord.
Ship Prompt: Your faceless channel is live. Post a link (or preview) of one video. What niche did you choose? How long did the pipeline take to produce one video?