Upload script
kochinaze uploader · the complete source of the tool (Python, standard library only)
This is the working script used to upload the operator's own videos. Flow: login (one-time Google OAuth sign-in with a loopback redirect; the refresh token is kept in the operator's password manager) → upload spec.json (resumable videos.insert → thumbnails.set → captions.insert for Japanese and English → videos.update for English localizations; the video is uploaded as private with publishAt) → status VIDEO_ID.
Scopes: youtube.upload, youtube.force-ssl. No data about viewers or other channels is read or stored.
"""Upload one finished video to the こちなぜ channel with the YouTube Data API (stdlib only).
usage:
python -I tools/youtube_api/yt_upload.py login # one-time browser sign-in; stores the refresh token in 1Password
python -I tools/youtube_api/yt_upload.py upload spec.json # upload -> thumbnail -> captions -> English -> schedule
python -I tools/youtube_api/yt_upload.py status VIDEO_ID # processing / privacy state (for the Studio patrol)
spec.json (paths relative to the spec file):
{"video": "x.mp4", "title": "...", "description_file": "desc_ja.txt", "tags": ["..."], "category": "28",
"publish_at": "2026-11-25T18:00:00+09:00", "made_for_kids": false, "synthetic_media": false,
"thumbnail": "thumb.jpg", "captions": {"ja": "ja.srt", "en": "en.srt"},
"en": {"title": "...", "description_file": "desc_en.txt"}}
Each finished step is written to <spec>.state.json, so a rerun continues where it stopped (the upload itself resumes too).
Secrets stay in 1Password: the OAuth client is the document "Google OAuth client (kochinaze uploader)", the refresh token
the item "YouTube refresh token (kochinaze)" (both in the Automation vault). Nothing secret is printed or written to disk.
Until the API project passes YouTube's audit, uploaded videos are locked to private (scheduling has no effect).
"""
import http.server, json, os, secrets, subprocess, sys, threading, urllib.error, urllib.parse, urllib.request, uuid, webbrowser
VAULT = "Automation"
CLIENT_DOC = "Google OAuth client (kochinaze uploader)"
TOKEN_ITEM = "YouTube refresh token (kochinaze)"
SCOPES = "https://www.googleapis.com/auth/youtube.upload https://www.googleapis.com/auth/youtube.force-ssl"
API = "https://www.googleapis.com"
CHUNK = 16 * 1024 * 1024
def op(*args):
r = subprocess.run(["op", *args], capture_output=True)
if r.returncode != 0:
sys.exit("op failed: " + r.stderr.decode("utf-8", "replace").strip())
return r.stdout.decode("utf-8")
def client():
c = json.loads(op("document", "get", CLIENT_DOC, "--vault", VAULT))
c = c.get("installed") or c.get("web")
return c["client_id"], c["client_secret"]
def post_form(url, data):
req = urllib.request.Request(url, urllib.parse.urlencode(data).encode(), {"Content-Type": "application/x-www-form-urlencoded"})
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read())
def login():
cid, secret = client()
state, got = secrets.token_urlsafe(16), {}
class H(http.server.BaseHTTPRequestHandler):
def do_GET(self):
q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
if q.get("state", [""])[0] == state:
got.update({k: v[0] for k, v in q.items()})
self.send_response(200); self.end_headers()
self.wfile.write("サインインできました。このタブは閉じてかまいません。".encode("utf-8"))
def log_message(self, *a):
pass
srv = http.server.HTTPServer(("127.0.0.1", 0), H)
redirect = f"http://127.0.0.1:{srv.server_address[1]}/"
url = "https://accounts.google.com/o/oauth2/v2/auth?" + urllib.parse.urlencode({
"client_id": cid, "redirect_uri": redirect, "response_type": "code", "scope": SCOPES,
"access_type": "offline", "prompt": "consent", "state": state})
print("ブラウザでサインインしてください（チャンネルは「こちら名瀬山公園前研究所」を選ぶ）:\n" + url, flush=True)
webbrowser.open(url)
while "code" not in got and "error" not in got:
srv.handle_request()
if "error" in got:
sys.exit("sign-in refused: " + got["error"])
tok = post_form("https://oauth2.googleapis.com/token", {"code": got["code"], "client_id": cid, "client_secret": secret,
"redirect_uri": redirect, "grant_type": "authorization_code"})
rt = tok.get("refresh_token") or sys.exit("no refresh token returned")
existing = subprocess.run(["op", "item", "get", TOKEN_ITEM, "--vault", VAULT], capture_output=True).returncode == 0
if existing:
op("item", "edit", TOKEN_ITEM, "--vault", VAULT, f"credential={rt}")
else:
op("item", "create", "--vault", VAULT, "--category", "API Credential", "--title", TOKEN_ITEM, f"credential={rt}")
me = call("GET", "/youtube/v3/channels?part=snippet&mine=true", token=tok["access_token"])
print("signed in as channel:", me["items"][0]["snippet"]["title"] if me.get("items") else "(none)")
_access = None
def access():
global _access
if not _access:
cid, secret = client()
rt = op("item", "get", TOKEN_ITEM, "--vault", VAULT, "--fields", "credential", "--reveal").strip()
_access = post_form("https://oauth2.googleapis.com/token", {"refresh_token": rt, "client_id": cid,
"client_secret": secret, "grant_type": "refresh_token"})["access_token"]
return _access
def call(method, path, body=None, headers=None, token=None, raw=False):
h = {"Authorization": "Bearer " + (token or access()), **(headers or {})}
data = body if isinstance(body, (bytes, type(None))) else json.dumps(body).encode()
if body is not None and not isinstance(body, bytes):
h["Content-Type"] = "application/json; charset=UTF-8"
req = urllib.request.Request(path if path.startswith("http") else API + path, data, h, method=method)
try:
with urllib.request.urlopen(req, timeout=600) as r:
return Raw(r.status, r.headers, r.read()) if raw else json.loads(r.read() or b"{}")
except urllib.error.HTTPError as e:
if raw:
return Raw(e.code, e.headers, e.read())
sys.exit(f"{method} {path.split('?')[0]} -> HTTP {e.code} {e.read().decode('utf-8', 'replace')[:400]}")
class Raw:
"""A finished response (the body is read before the connection closes)."""
def __init__(self, code, headers, body):
self.code, self.headers, self.body = code, headers, body
def read(self):
return self.body
def upload_video(spec, base, st, save):
path = os.path.join(base, spec["video"]); size = os.path.getsize(path)
if not st.get("upload_url"):
meta = {"snippet": {"title": spec["title"], "description": open(os.path.join(base, spec["description_file"]), encoding="utf-8").read(),
"tags": spec.get("tags", []), "categoryId": spec.get("category", "28"), "defaultLanguage": "ja",
"defaultAudioLanguage": "ja"},
"status": {"privacyStatus": "private", "selfDeclaredMadeForKids": spec.get("made_for_kids", False),
"containsSyntheticMedia": spec.get("synthetic_media", False)}}
if spec.get("publish_at"):
meta["status"]["publishAt"] = spec["publish_at"]
r = call("POST", "/upload/youtube/v3/videos?uploadType=resumable&part=snippet,status", meta,
{"X-Upload-Content-Type": "video/*", "X-Upload-Content-Length": str(size)}, raw=True)
st["upload_url"] = r.headers["Location"]; save()
done = 0
probe = call("PUT", st["upload_url"], b"", {"Content-Range": f"bytes */{size}"}, raw=True)
if probe.code in (200, 201):
st["video_id"] = json.loads(probe.read())["id"]; save(); return
rng = probe.headers.get("Range")
if rng:
done = int(rng.split("-")[1]) + 1
with open(path, "rb") as f:
while done < size:
f.seek(done); chunk = f.read(CHUNK); end = done + len(chunk) - 1
r = call("PUT", st["upload_url"], chunk, {"Content-Range": f"bytes {done}-{end}/{size}"}, raw=True)
code = r.code
if code in (200, 201):
st["video_id"] = json.loads(r.read())["id"]; save(); print(f" uploaded 100%"); return
if code != 308:
sys.exit(f"upload chunk failed: HTTP {code}")
done = int(r.headers["Range"].split("-")[1]) + 1
print(f" uploaded {done * 100 // size}%", flush=True)
def multipart(meta, blob, ctype):
b = uuid.uuid4().hex
body = (f"--{b}\r\nContent-Type: application/json; charset=UTF-8\r\n\r\n" + json.dumps(meta) + f"\r\n--{b}\r\nContent-Type: {ctype}\r\n\r\n").encode() + blob + f"\r\n--{b}--".encode()
return body, {"Content-Type": f"multipart/related; boundary={b}"}
def upload(spec_path):
spec = json.load(open(spec_path, encoding="utf-8")); base = os.path.dirname(os.path.abspath(spec_path))
sp = spec_path + ".state.json"
st = json.load(open(sp, encoding="utf-8")) if os.path.exists(sp) else {}
save = lambda: json.dump(st, open(sp, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
if not st.get("video_id"):
print("1/4 動画を上げる"); upload_video(spec, base, st, save)
vid = st["video_id"]; print(" video:", vid)
if spec.get("thumbnail") and not st.get("thumbnail"):
print("2/4 表紙")
p = os.path.join(base, spec["thumbnail"]); ct = "image/png" if p.lower().endswith(".png") else "image/jpeg"
call("POST", f"/upload/youtube/v3/thumbnails/set?videoId={vid}", open(p, "rb").read(), {"Content-Type": ct})
st["thumbnail"] = True; save()
for lang, f in (spec.get("captions") or {}).items():
if st.get("caption_" + lang):
continue
print(f"3/4 字幕 {lang}")
body, h = multipart({"snippet": {"videoId": vid, "language": lang, "name": "", "isDraft": False}},
open(os.path.join(base, f), "rb").read(), "application/octet-stream")
call("POST", "/upload/youtube/v3/captions?uploadType=multipart&part=snippet", body, h)
st["caption_" + lang] = True; save()
if spec.get("en") and not st.get("en"):
print("4/4 英語のタイトル・説明")
v = call("GET", f"/youtube/v3/videos?part=snippet,localizations&id={vid}")["items"][0]
loc = v.get("localizations", {})
loc["en"] = {"title": spec["en"]["title"], "description": open(os.path.join(base, spec["en"]["description_file"]), encoding="utf-8").read()}
call("PUT", "/youtube/v3/videos?part=snippet,localizations", {"id": vid, "snippet": {**{k: v["snippet"][k] for k in ("title", "description", "categoryId")},
"tags": v["snippet"].get("tags", []), "defaultLanguage": "ja"}, "localizations": loc})
st["en"] = True; save()
s = call("GET", f"/youtube/v3/videos?part=status&id={vid}")["items"][0]["status"]
print(f"done: https://youtu.be/{vid} privacy={s.get('privacyStatus')} publishAt={s.get('publishAt')}")
def status(vid):
v = call("GET", f"/youtube/v3/videos?part=status,processingDetails&id={vid}").get("items")
print(json.dumps(v[0] if v else {"error": "not found"}, ensure_ascii=False, indent=1))
if __name__ == "__main__":
cmd = sys.argv[1] if len(sys.argv) > 1 else ""
if cmd == "login":
login()
elif cmd == "upload" and len(sys.argv) > 2:
upload(sys.argv[2])
elif cmd == "status" and len(sys.argv) > 2:
status(sys.argv[2])
else:
sys.exit(__doc__)