Badger Uploader source code
The complete tool: two Python 3 files, standard library only.
yt_auth.py does the one-time Google sign-in (OAuth 2.0, installed-app flow with PKCE).
badger_upload.py uploads a video from our upload notes, schedules it, sets the thumbnail and playlist,
and checks the result. Plain-text copies: badger_upload.py ·
yt_auth.py.
yt_auth.py
#!/usr/bin/env python3
"""Badger Uploader one-time Google sign-in (installed-app flow, PKCE).
yt_auth.py url <name> -> prints the sign-in link, remembers the PKCE verifier
yt_auth.py code <name> <url> -> takes the http://localhost/?code=... address the browser ended on,
stores the refresh token as ~/.config/argus/youtube/token_<name>.json (600)
One token per CHANNEL: a Brand Account channel is chosen on Google's consent screen."""
import base64, hashlib, json, os, secrets, sys, urllib.parse, urllib.request
D = os.path.expanduser('~/.config/argus/youtube')
C = json.load(open(f'{D}/client_secret.json'))['installed']
SCOPES = 'https://www.googleapis.com/auth/youtube.upload https://www.googleapis.com/auth/youtube'
REDIRECT = 'http://localhost'
def w600(p, obj):
fd = os.open(p, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, 'w') as f: json.dump(obj, f, indent=1)
cmd, name = sys.argv[1], sys.argv[2]
if cmd == 'url':
ver = secrets.token_urlsafe(64); state = secrets.token_urlsafe(16)
ch = base64.urlsafe_b64encode(hashlib.sha256(ver.encode()).digest()).rstrip(b'=').decode()
w600(f'{D}/pending_{name}.json', {'verifier': ver, 'state': state})
print(C['auth_uri'] + '?' + urllib.parse.urlencode({'client_id': C['client_id'], 'redirect_uri': REDIRECT,
'response_type': 'code', 'scope': SCOPES, 'access_type': 'offline', 'prompt': 'consent select_account',
'code_challenge': ch, 'code_challenge_method': 'S256', 'state': state}))
elif cmd == 'code':
p = json.load(open(f'{D}/pending_{name}.json'))
q = urllib.parse.parse_qs(urllib.parse.urlparse(sys.argv[3]).query)
if q.get('state', [''])[0] != p['state']: sys.exit('state mismatch: use the link from the latest url step')
body = urllib.parse.urlencode({'code': q['code'][0], 'client_id': C['client_id'], 'client_secret': C['client_secret'],
'redirect_uri': REDIRECT, 'grant_type': 'authorization_code', 'code_verifier': p['verifier']}).encode()
tok = json.load(urllib.request.urlopen(C['token_uri'], body))
if 'refresh_token' not in tok: sys.exit('no refresh token returned: ' + json.dumps(tok)[:300])
w600(f'{D}/token_{name}.json', {'refresh_token': tok['refresh_token'], 'scope': tok.get('scope')})
os.remove(f'{D}/pending_{name}.json')
req = urllib.request.Request('https://www.googleapis.com/youtube/v3/channels?part=snippet&mine=true',
headers={'Authorization': 'Bearer ' + tok['access_token']})
for it in json.load(urllib.request.urlopen(req)).get('items', []):
print('signed in to channel:', it['snippet']['title'], it['id'])
badger_upload.py
#!/usr/bin/env python3
"""
Badger Uploader - uploads Joe Badger Productions videos to our own YouTube channels.
Internal tool, one user (the channel owner). Standard library only.
Uses YouTube Data API v3 with the owner's own OAuth 2.0 consent:
videos.insert (resumable) upload the video with its title, description, tags, category,
made-for-kids setting and altered/synthetic content disclosure
status.publishAt schedule when it goes public
thumbnails.set custom thumbnail
playlists.list / playlistItems.insert add it to one of our own playlists
channels.list (mine=true) confirm we are signed in to the intended channel
videos.list check the upload processed
Usage:
badger_upload.py --channel nursery --kit "/path/to/Upload kit.txt" [--publish-at "2026-10-23 19:00"] [--dry-run]
badger_upload.py --channel nursery --status VIDEO_ID
badger_upload.py --channel nursery --delete VIDEO_ID (removes one of our own uploads)
--publish-at is Korea time (KST). Without it the video stays private.
Credentials: ~/.config/argus/youtube/client_secret.json + token_<channel>.json (created by yt_auth.py).
Stored data: only the IDs of our own uploads (uploads.json), so nothing is uploaded twice.
"""
import argparse, datetime, json, os, re, sys, time, urllib.error, urllib.parse, urllib.request
CONF = os.path.expanduser('~/.config/argus/youtube')
API = 'https://www.googleapis.com/youtube/v3'
UPLOAD_API = 'https://www.googleapis.com/upload/youtube/v3'
CHUNK = 64 * 1024 * 1024 # resumable upload chunk (multiple of 256 KiB)
KST = datetime.timezone(datetime.timedelta(hours=9))
CATEGORIES = {'music': '10', 'pets & animals': '15', 'entertainment': '24', 'people & blogs': '22'}
EXPECTED_CHANNEL = {'nursery': 'UCQYf8gVLyVv4MNGj-6H6wzw'} # refuse to upload to the wrong channel
def log(*a):
print(time.strftime('%H:%M:%S'), *a, flush=True)
# ------------------------------------------------------------------ auth
class Auth:
def __init__(self, channel):
self.client = json.load(open(f'{CONF}/client_secret.json'))['installed']
self.refresh = json.load(open(f'{CONF}/token_{channel}.json'))['refresh_token']
self.token, self.expires = None, 0
def header(self):
if time.time() > self.expires - 120:
body = urllib.parse.urlencode({'client_id': self.client['client_id'],
'client_secret': self.client['client_secret'],
'refresh_token': self.refresh,
'grant_type': 'refresh_token'}).encode()
t = json.load(urllib.request.urlopen(self.client['token_uri'], body, timeout=60))
self.token, self.expires = t['access_token'], time.time() + t.get('expires_in', 3600)
return {'Authorization': 'Bearer ' + self.token}
def call(auth, method, url, body=None, headers=None, raw=False, ok=(200, 201, 204)):
"""One API request with retries on transient errors. Returns (status, headers, parsed body)."""
for attempt in range(6):
h = dict(auth.header())
h.update(headers or {})
data = body
if body is not None and not raw:
data = json.dumps(body).encode()
h['Content-Type'] = 'application/json; charset=utf-8'
req = urllib.request.Request(url, data=data, method=method, headers=h)
try:
with urllib.request.urlopen(req, timeout=300) as r:
txt = r.read().decode() or '{}'
return r.status, r.headers, json.loads(txt)
except urllib.error.HTTPError as e:
txt = e.read().decode(errors='replace')
if e.code == 308: # resumable: "incomplete, carry on"
return 308, e.headers, {}
if e.code in (500, 502, 503, 504) and attempt < 5:
time.sleep(2 ** attempt * 5); continue
raise SystemExit(f'API error {e.code} on {method} {url.split("?")[0]}: {txt[:600]}')
except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
if attempt < 5:
time.sleep(2 ** attempt * 5); continue
raise SystemExit(f'network error: {e}')
# ------------------------------------------------------------------ the upload kit
def parse_kit(path):
"""Read our 'Upload kit.txt' (written for humans) into API fields."""
txt = open(path, encoding='utf-8').read()
folder = os.path.dirname(os.path.abspath(path))
def field(name):
m = re.search(rf'^{name}:\s*(.+)$', txt, re.M)
return m.group(1).strip() if m else None
def after(heading):
m = re.search(rf'^{re.escape(heading)}.*\n(.+)$', txt, re.M)
return m.group(1).strip() if m else None
desc = re.search(r'^DESCRIPTION.*\n-{10,}\n(.*?)\n-{10,}', txt, re.M | re.S)
settings = {k.strip().lower(): v.strip() for k, v in
re.findall(r'^- ([^:\n]+):\s*(.+)$', txt.split('SETTINGS:', 1)[-1], re.M)}
kit = {
'file': os.path.join(folder, field('FILE')),
'thumbnail': os.path.join(folder, field('THUMBNAIL')) if field('THUMBNAIL') else None,
'title': after('TITLE'),
'description': desc.group(1).strip() if desc else '',
'tags': [t.strip() for t in (after('TAGS') or '').split(',') if t.strip()],
'made_for_kids': settings.get('audience', '').lower().startswith('yes'),
'synthetic': settings.get('altered or synthetic content', '').lower().startswith('yes'),
'playlist': settings.get('playlist'),
'category': CATEGORIES.get(settings.get('category', 'music').lower(), '10'),
}
problems = [k for k in ('title', 'description') if not kit[k]]
if not os.path.isfile(kit['file']): problems.append('file not found: ' + kit['file'])
if kit['thumbnail'] and not os.path.isfile(kit['thumbnail']): problems.append('thumbnail not found')
if len(kit['title']) > 100: problems.append('title over 100 characters')
if len(kit['description']) > 5000: problems.append('description over 5000 characters')
if problems: raise SystemExit('kit problems: ' + '; '.join(problems))
return kit
# ------------------------------------------------------------------ steps
def check_channel(auth, channel):
_, _, r = call(auth, 'GET', f'{API}/channels?part=snippet&mine=true')
items = r.get('items', [])
if not items: raise SystemExit('this sign-in has no channel')
ch = items[0]
want = EXPECTED_CHANNEL.get(channel)
if want and ch['id'] != want:
raise SystemExit(f'signed in to {ch["snippet"]["title"]} ({ch["id"]}), expected {want}: refusing')
log('channel:', ch['snippet']['title'], ch['id'])
return ch['id']
def publish_time(s):
if not s: return None
t = datetime.datetime.strptime(s, '%Y-%m-%d %H:%M').replace(tzinfo=KST)
if t < datetime.datetime.now(KST) + datetime.timedelta(minutes=30):
raise SystemExit('--publish-at must be at least 30 minutes in the future')
return t.astimezone(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
def upload(auth, kit, publish_at):
size = os.path.getsize(kit['file'])
status = {'privacyStatus': 'private',
'selfDeclaredMadeForKids': kit['made_for_kids'],
'containsSyntheticMedia': kit['synthetic'],
'embeddable': True}
if publish_at: status['publishAt'] = publish_at # requires privacyStatus private
meta = {'snippet': {'title': kit['title'], 'description': kit['description'],
'tags': kit['tags'], 'categoryId': kit['category']},
'status': status}
st, h, _ = call(auth, 'POST', f'{UPLOAD_API}/videos?uploadType=resumable&part=snippet,status',
meta, {'X-Upload-Content-Length': str(size), 'X-Upload-Content-Type': 'video/mp4'})
session = h['Location']
log(f'upload session open, {size / 1e9:.2f} GB')
sent, t0 = 0, time.time()
with open(kit['file'], 'rb') as f:
while sent < size:
f.seek(sent)
chunk = f.read(CHUNK)
end = sent + len(chunk) - 1
st, h, r = call(auth, 'PUT', session, chunk, raw=True,
headers={'Content-Length': str(len(chunk)),
'Content-Range': f'bytes {sent}-{end}/{size}'})
if st in (200, 201):
log(f'upload complete in {(time.time() - t0) / 60:.1f} min')
return r['id']
rng = h.get('Range') # what the server actually holds
sent = int(rng.split('-')[1]) + 1 if rng else 0
pct = sent / size * 100
log(f' {pct:5.1f}% {sent / 1e9:.2f} GB {sent / 1e6 / max(time.time() - t0, 1):.1f} MB/s')
raise SystemExit('upload ended without a video id')
def confirm_status(auth, vid, kit, publish_at):
"""Re-apply the audience and disclosure settings with videos.update after the upload, so they
are set explicitly even if the insert call left any of them at the channel default."""
status = {'privacyStatus': 'private', 'selfDeclaredMadeForKids': kit['made_for_kids'],
'containsSyntheticMedia': kit['synthetic'], 'embeddable': True,
'license': 'youtube', 'publicStatsViewable': True}
if publish_at: status['publishAt'] = publish_at
_, _, r = call(auth, 'PUT', f'{API}/videos?part=status', {'id': vid, 'status': status})
s = r.get('status', {})
log(f'settings confirmed: madeForKids={s.get("selfDeclaredMadeForKids")} '
f'synthetic={s.get("containsSyntheticMedia")} publishAt={s.get("publishAt", "-")}')
def set_thumbnail(auth, vid, path):
if not path: return
data = open(path, 'rb').read()
if len(data) > 2 * 1024 * 1024: raise SystemExit('thumbnail over 2 MB')
ctype = 'image/png' if path.lower().endswith('.png') else 'image/jpeg'
call(auth, 'POST', f'{UPLOAD_API}/thumbnails/set?videoId={vid}', data, {'Content-Type': ctype}, raw=True)
log('thumbnail set')
def add_to_playlist(auth, vid, name):
if not name: return
page = ''
while True:
_, _, r = call(auth, 'GET', f'{API}/playlists?part=snippet&mine=true&maxResults=50{page}')
for p in r.get('items', []):
if p['snippet']['title'].strip().lower() == name.strip().lower():
call(auth, 'POST', f'{API}/playlistItems?part=snippet',
{'snippet': {'playlistId': p['id'], 'resourceId': {'kind': 'youtube#video', 'videoId': vid}}})
log('added to playlist:', p['snippet']['title'])
return
if not r.get('nextPageToken'): break
page = '&pageToken=' + r['nextPageToken']
log(f'WARNING: playlist "{name}" not found on this channel; not added')
def show_status(auth, vid):
_, _, r = call(auth, 'GET', f'{API}/videos?part=snippet,status,processingDetails&id={vid}')
if not r.get('items'): raise SystemExit(f'video {vid} not found on this channel')
v = r['items'][0]
s = v['status']
log(f'video {vid}: "{v["snippet"]["title"]}"')
log(f' privacy={s.get("privacyStatus")} publishAt={s.get("publishAt", "-")} '
f'madeForKids={s.get("madeForKids")} synthetic={s.get("containsSyntheticMedia")} '
f'upload={s.get("uploadStatus")} processing={v.get("processingDetails", {}).get("processingStatus", "-")}')
if s.get('failureReason') or s.get('rejectionReason'):
log(' PROBLEM:', s.get('failureReason') or s.get('rejectionReason'))
log(f' https://studio.youtube.com/video/{vid}/edit')
return v
def record(channel, vid, kit):
p = f'{CONF}/uploads.json'
d = json.load(open(p)) if os.path.exists(p) else []
d.append({'channel': channel, 'video_id': vid, 'file': os.path.basename(kit['file']),
'title': kit['title'], 'uploaded_at': datetime.datetime.now(KST).isoformat(timespec='seconds')})
fd = os.open(p, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, 'w') as f: json.dump(d, f, indent=1, ensure_ascii=False)
def already_uploaded(channel, kit):
p = f'{CONF}/uploads.json'
if not os.path.exists(p): return None
for u in json.load(open(p)):
if u['channel'] == channel and u['file'] == os.path.basename(kit['file']):
return u['video_id']
# ------------------------------------------------------------------ main
def main():
ap = argparse.ArgumentParser(description='Badger Uploader')
ap.add_argument('--channel', required=True)
ap.add_argument('--kit')
ap.add_argument('--publish-at', help='KST, "YYYY-MM-DD HH:MM"; omit to keep private')
ap.add_argument('--dry-run', action='store_true')
ap.add_argument('--force', action='store_true', help='upload even if this file was uploaded before')
ap.add_argument('--chunk-mb', type=int, default=64, help='upload chunk size (multiple of 0.25 MB)')
ap.add_argument('--status')
ap.add_argument('--delete')
a = ap.parse_args()
global CHUNK
CHUNK = a.chunk_mb * 1024 * 1024
auth = Auth(a.channel)
if a.status: return show_status(auth, a.status)
if a.delete:
check_channel(auth, a.channel)
call(auth, 'DELETE', f'{API}/videos?id={a.delete}')
return log('deleted', a.delete)
if not a.kit: ap.error('--kit is required to upload')
kit = parse_kit(a.kit)
when = publish_time(a.publish_at)
log('kit ok:', kit['title'])
log(f' madeForKids={kit["made_for_kids"]} synthetic={kit["synthetic"]} category={kit["category"]} '
f'playlist={kit["playlist"]} publishAt={when or "stays private"} tags={len(kit["tags"])}')
check_channel(auth, a.channel)
prev = already_uploaded(a.channel, kit)
if prev and not a.force: raise SystemExit(f'already uploaded as {prev} (use --force to upload again)')
if a.dry_run: return log('dry run: nothing uploaded')
vid = upload(auth, kit, when)
record(a.channel, vid, kit)
log('video id:', vid)
confirm_status(auth, vid, kit, when)
set_thumbnail(auth, vid, kit['thumbnail'])
add_to_playlist(auth, vid, kit['playlist'])
show_status(auth, vid)
if __name__ == '__main__':
main()