How one AI agent builds and ships my YouTube Shorts every day

Avatar match (the agent picks by look, not name — VEED labels every tile "Character"):
- Bald presenter
- Blue Design Delight Studio tee with the copper DDS chest logo
- Light background
Keep a screenshot of this tile on disk as your reference thumbnail.

Create a seamless, looping vertical 9:16 (1080x1920) video, about 44 seconds, of green
falling-glyph "code-rain": thin columns of dim cyan-and-green monospace glyphs
(0 1 / \ : ; . + * [ ] $ #) descending on a near-black #04090c background. Keep a
denser brighter band through the middle third. Subtle, even, hypnotic; no text,
no logos, no faces, no UI. If your tool offers model tiers, pick the higher-quality
tier (the fast tier can render an all-black clip).

Create a vertical 9:16 hook card for a YouTube Short. Background: near-black #04090c
with a subtle dark-green falling-glyph code-rain texture. Foreground: a big, bold,
ALL-CAPS headline in bright gold #FFE069 -- three short stacked lines:
"<LINE 1>", "<LINE 2>", "<LINE 3>". Heavy condensed sans, high contrast, slight glow
so it reads on mobile. Add one small glowing gold assistant icon. No watermark, no UI,
no clutter, no extra text. Mobile-legible, premium, cinematic.

google-api-python-client
google-auth-oauthlib
google-auth-httplib2

pip install -r requirements.txt

#!/usr/bin/env python3
"""
get_refresh_token.py - one-time OAuth consent. Writes yt_oauth.json.

Mint your OWN credentials:
  1) In Google Cloud, enable "YouTube Data API v3".
  2) Configure the OAuth consent screen as External and set it to "In production"
     (Testing-mode refresh tokens expire in 7 days and will silently break the
     scheduled uploads after a week).
  3) Create an OAuth client of type "Desktop app" and download its JSON as
     client_secret.json into this folder.
  4) Run this script. A browser opens; sign in and consent AS THE BRAND ACCOUNT
     (the channel you want the Shorts on, not your personal account).

Result: yt_oauth.json, holding the refresh token. Keep it secret.
"""
from google_auth_oauthlib.flow import InstalledAppFlow

SCOPES = [
    "https://www.googleapis.com/auth/youtube.upload",
    "https://www.googleapis.com/auth/youtube.readonly",
]


def main():
    flow = InstalledAppFlow.from_client_secrets_file("client_secret.json", SCOPES)
    # Opens a local browser for consent. Sign in as the BRAND Google account.
    creds = flow.run_local_server(port=0)
    with open("yt_oauth.json", "w", encoding="utf-8") as fh:
        fh.write(creds.to_json())
    print("Wrote yt_oauth.json. Keep it secret.")
    print("If uploads start failing after ~7 days, your consent screen is still in "
          "Testing mode. Move it to 'In production' and re-run this once.")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
yt_upload.py - resumable YouTube upload + native scheduling.

Mint your OWN credentials. Never reuse anyone else's yt_oauth.json or
client_secret.json. Run get_refresh_token.py once to create yt_oauth.json.

What it does:
  - refreshes your token,
  - VERIFIES the token is bound to the channel you expect (the guard),
  - resumable-uploads the file (no browser size cap),
  - sets status.publishAt so the video is PRIVATE until the scheduled time
    (scheduled, never public-now).

The youtube.upload scope can insert + schedule, but cannot set playlists or
thumbnails. Do those in YouTube Studio after the upload.

Usage:
  python3 yt_upload.py SHORT.mp4 \
    --title "My title" --desc-file desc.txt --tags "a,b,c" \
    --category 28 --publish-at 2026-06-20T18:00:00Z \
    --creds yt_oauth.json --channel-id UCxxxxxxxxxxxxxxxxxxxxxx
"""
import argparse
import os
import random
import sys
import time

import googleapiclient.discovery
import googleapiclient.errors
from googleapiclient.http import MediaFileUpload
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request

# youtube.upload inserts + schedules; youtube.readonly lets us read the channel id for the guard.
SCOPES = [
    "https://www.googleapis.com/auth/youtube.upload",
    "https://www.googleapis.com/auth/youtube.readonly",
]
RETRIABLE_STATUS = (500, 502, 503, 504)
MAX_RETRIES = 8


def get_service(creds_path):
    if not os.path.exists(creds_path):
        sys.exit(f"Missing {creds_path}. Run get_refresh_token.py first.")
    creds = Credentials.from_authorized_user_file(creds_path, SCOPES)
    if not creds.valid:
        if creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            sys.exit("Token invalid or expired with no refresh token. Re-run get_refresh_token.py.")
    return googleapiclient.discovery.build("youtube", "v3", credentials=creds)


def verify_channel(youtube, expected_id):
    """The guard: stop the run if the token is bound to the wrong channel."""
    resp = youtube.channels().list(part="id,snippet", mine=True).execute()
    items = resp.get("items", [])
    if not items:
        sys.exit("This token controls no channel. Re-consent as the brand account.")
    cid = items[0]["id"]
    title = items[0]["snippet"]["title"]
    print(f"[channel] {title}  ({cid})")
    if expected_id and cid != expected_id:
        sys.exit(
            f"STOP: token is bound to {cid}, but you expected {expected_id}.\n"
            "Re-run get_refresh_token.py and consent as the CORRECT Google account."
        )
    return cid


def resumable_upload(request, filename):
    response = None
    error = None
    retry = 0
    print(f"Uploading {filename} ...")
    while response is None:
        try:
            status, response = request.next_chunk()
            if status:
                print(f"  {int(status.progress() * 100)}%")
            if response is not None and "id" not in response:
                sys.exit(f"Upload failed, unexpected response: {response}")
        except googleapiclient.errors.HttpError as e:
            if e.resp.status in RETRIABLE_STATUS:
                error = e
            else:
                raise
        except (IOError, OSError) as e:
            error = e
        if error is not None:
            retry += 1
            if retry > MAX_RETRIES:
                sys.exit("Too many retries. Aborting.")
            sleep = min((2 ** retry) + random.random(), 60)
            print(f"  retriable error; retry {retry} in {sleep:.1f}s")
            time.sleep(sleep)
            error = None
    return response


def main():
    ap = argparse.ArgumentParser(description="Resumable YouTube upload + schedule.")
    ap.add_argument("file", help="Path to the finished .mp4")
    ap.add_argument("--title", required=True)
    ap.add_argument("--desc-file", help="UTF-8 text file with the description")
    ap.add_argument("--tags", default="", help="Comma-separated tags")
    ap.add_argument("--category", default="28", help="28 = Science & Technology")
    ap.add_argument("--publish-at", required=True,
                    help="RFC 3339, e.g. 2026-06-20T18:00:00Z (UTC). Makes it SCHEDULED.")
    ap.add_argument("--creds", default="yt_oauth.json")
    ap.add_argument("--channel-id", default=os.environ.get("YT_CHANNEL_ID", ""),
                    help="Your brand channel id; the run stops if the token does not match.")
    args = ap.parse_args()

    if not os.path.exists(args.file):
        sys.exit(f"File not found: {args.file}")

    description = ""
    if args.desc_file:
        if not os.path.exists(args.desc_file):
            sys.exit(f"Description file not found: {args.desc_file}")
        with open(args.desc_file, encoding="utf-8") as fh:
            description = fh.read()

    youtube = get_service(args.creds)
    verify_channel(youtube, args.channel_id)

    body = {
        "snippet": {
            "title": args.title,
            "description": description,
            "tags": [t.strip() for t in args.tags.split(",") if t.strip()],
            "categoryId": args.category,
        },
        "status": {
            "privacyStatus": "private",   # private + publishAt == SCHEDULED
            "publishAt": args.publish_at,  # do NOT also set 'public' anywhere
            "selfDeclaredMadeForKids": False,
        },
    }

    media = MediaFileUpload(args.file, chunksize=8 * 1024 * 1024, resumable=True)
    request = youtube.videos().insert(part="snippet,status", body=body, media_body=media)
    resp = resumable_upload(request, args.file)

    video_id = resp["id"]
    print(f"DONE: https://youtu.be/{video_id}")
    print(f"Scheduled for {args.publish_at}. Set playlists + thumbnail in Studio.")


if __name__ == "__main__":
    main()

# mint your OWN credentials first (get_refresh_token.py). category 28 = Science & Technology.
python3 yt_upload.py rendered_shorts/<SHORT_ID>_<date>.mp4 \
  --title "<title>" --desc-file desc.txt --tags "veed ai,ai video automation" \
  --category 28 --publish-at 2026-06-20T18:00:00Z \
  --creds yt_oauth.json --channel-id <YOUR_CHANNEL_ID>

REM Windows: run the daily build at 1:00 AM. run_daily.bat triggers your agent + ship.sh.
schtasks /Create /TN "ddsboston-veed-auto-build-daily" /SC DAILY /ST 01:00 /F ^
  /TR "C:\YouTube\03_Content_Pipeline\run_daily.bat"

# macOS / Linux equivalent (crontab -e): 1:00 AM daily
0 1 * * * /home/you/youtube/03_Content_Pipeline/run_daily.sh >> ~/veed.log 2>&1

You are the daily Shorts build agent. On each run, with no questions:
1. Read the active build order and production_log.md. Pick the FIRST Short with no successful row.
2. Check the queue depth and the next open publish slot in Studio. Skip long-form days and any booked date.
3. Build the Short in VEED (fresh AI-Avatars project every run; 9:16; avatar by visual match;
   voice = Dan, US English; Eye Contact + Remove Background; code-rain background full length, no black tail;
   auto B-roll; gold karaoke captions; export YouTube 4K, burn subtitles).
4. Verify the matte ON THE EXPORT. If Remove Background reverted, re-export once; if still bad,
   SHIP ANYWAY, log "matte reverted - shipped per override", and create a one-time patch task.
5. Retrieve the .mp4 (curl), generate the Gemini hook card, composite the first GOP with ffmpeg.
6. Upload with yt_upload.py using --publish-at (SCHEDULED, never public now). Verify the channel id.
7. Add playlists in Studio, append production_log.md, and schedule the pinned-comment follow-up.
HARD RULES: never publish immediately (always --publish-at); never change the voice (Dan) or quality (4K);
never invent results in the description.

#!/usr/bin/env bash
# ship.sh - the deterministic half of the pipeline (retrieve -> composite -> upload).
# The VEED build (browser) is driven by the agent; this chains the steps that ARE scriptable.
# Usage: ./ship.sh <SHORT_ID> <YYYY-MM-DD> <MP4_URL> <HOOK_IMAGE> <PUBLISH_AT> <CHANNEL_ID>
set -euo pipefail
ID="$1"; DATE="$2"; MP4_URL="$3"; HOOK="$4"; PUBAT="$5"; CHAN="$6"
BASE="rendered_shorts/${ID}_${DATE}.mp4"
OUT="${ID}_${DATE}.mp4"

# 1) retrieve from the VEED CDN
curl -sL "$MP4_URL" -o "$BASE"
SIZE=$(wc -c < "$BASE")
if [ "$SIZE" -lt 1000000 ]; then echo "Download too small ($SIZE bytes) - URL likely expired"; exit 1; fi

# 2) GOP-aligned hook composite (re-encode first GOP only, stream-copy the tail)
ffmpeg -y -t 8.333333 -i "$BASE" -i "$HOOK" -filter_complex \
  "[1:v]scale=2160:3840:force_original_aspect_ratio=increase,crop=2160:3840,setsar=1[hk];[0:v][hk]overlay=x=0:y=0:enable='lt(t,2)'[v]" \
  -map "[v]" -map 0:a -c:v libx264 -preset veryfast -crf 18 -pix_fmt yuv420p -c:a aac -b:a 192k head.mp4
ffmpeg -y -ss 8.333333 -i "$BASE" -c copy tail.mp4
printf "file 'head.mp4'\nfile 'tail.mp4'\n" > concat.txt
ffmpeg -y -f concat -safe 0 -i concat.txt -c copy "$OUT"

# 3) upload + schedule (verifies channel id, never publishes now)
python3 yt_upload.py "$OUT" --title "$ID" --tags "veed ai,ai video automation" \
  --category 28 --publish-at "$PUBAT" --creds yt_oauth.json --channel-id "$CHAN"
echo "Shipped $OUT (scheduled $PUBAT)."

| short_id        | build_date | publish_at (UTC)      | youtube_id  | matte | notes               |
|-----------------|------------|-----------------------|-------------|-------|---------------------|
| VA_cost_hook_01 | 2026-06-13 | 2026-06-20T18:00:00Z  | t2qWz2M5QTM | ok    | -                   |

Free masterclass (full walkthrough): https://youtu.be/<LONGFORM_ID>
This Short's class page: https://ddsboston.com/pages/<CLASS_SLUG>
All 60 free classes (no signup): https://ddsboston.com/pages/dds-vibe-academy

# Get the direct MP4 URL from the VEED Share panel (download anchor href), or read the
# network requests for a *.mp4 on the VEED CDN, then pull it into the connected folder:
curl -sL "<MP4_URL>" -o "rendered_shorts/<SHORT_ID>_<YYYY-MM-DD>.mp4"

# Verify: a 40-60s 4K Short is ~250-360 MB and ffprobe/file reports an MP4.
# If the download is HTML or under 1 MB, the URL expired -- re-grab it.

HOOK="<newest Gemini hook-card image in your controlled download path>"
BASE="rendered_shorts/<SHORT_ID>_<date>.mp4"

# head: first GOP, hook card overlaid full-frame for the first 2s (audio kept)
ffmpeg -y -t 8.333333 -i "$BASE" -i "$HOOK" -filter_complex \
  "[1:v]scale=2160:3840:force_original_aspect_ratio=increase,crop=2160:3840,setsar=1[hk];\
   [0:v][hk]overlay=x=0:y=0:enable='lt(t,2)'[v]" \
  -map "[v]" -map 0:a -c:v libx264 -preset veryfast -crf 18 -pix_fmt yuv420p \
  -c:a aac -b:a 192k head.mp4

# tail: stream-copy from the 8.333s keyframe (no re-encode -> fast, lossless)
ffmpeg -y -ss 8.333333 -i "$BASE" -c copy tail.mp4

# concat head + tail
printf "file 'head.mp4'\nfile 'tail.mp4'\n" > concat.txt
ffmpeg -y -f concat -safe 0 -i concat.txt -c copy "<SHORT_ID>_<date>.mp4"