diff --git a/README.md b/README.md
index cee4e55..44ebe4f 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
# social-auto-upload
`social-auto-upload` 是一个强大的自动化工具，旨在帮助内容创作者和运营者高效地将视频内容一键发布到多个国内外主流社交媒体平台。
-项目实现了对 `抖音`、`Bilibili`、`小红书`、`快手`、`视频号`、`百家号` 以及 `TikTok` 等平台的视频上传、定时发布等功能。
+项目实现了对 `抖音`、`Bilibili`、`小红书`、`快手`、`视频号`、`百家号`、`TikTok` 以及 `YouTube` 等平台的视频上传、定时发布等功能。
结合各平台 `uploader` 模块，您可以轻松配置和扩展支持的平台，并通过示例脚本快速上手。
@@ -69,6 +69,7 @@
| 视频号 | ✅ | ✅ | ❌ | ✅ | ❌ | ❌ | 对应 `tencent_uploader` |
| 百家号 | ✅ | ✅ | ❌ | ✅ | ❌ | ❌ | 浏览器自动化 |
| TikTok | ✅ | ✅ | ❌ | ✅ | ❌ | ❌ | 当前示例走 Chrome 版实现 |
+| YouTube | ✅ | ✅ | ❌ | ❌ | ✅ | ❌ | 浏览器自动化（Studio），支持加入播放列表/可见性 |
### AI这么强，为什么还需要这个项目
在你使用AI的能力，browser agent等等，每次都让 agent 重新解析网页、截图理解, 临场判断
@@ -176,8 +177,19 @@ sau xiaohongshu upload-note --account --images videos/1.png video
sau bilibili login --account
sau bilibili check --account
sau bilibili upload-video --account --file videos/demo.mp4 --title "示例标题" --desc "示例简介" --tid 249
+
+sau youtube login --account
+sau youtube check --account
+sau youtube upload-video --account --file videos/demo.mp4 --title "示例标题" --desc "示例简介" --tags tag1,tag2 --playlist "我的系列" --visibility public
```
+> YouTube 说明：登录是交互式的（Google 账号，浏览器里完成，无二维码）。这里走浏览器自动化而不是官方 API，
+> 是因为**未通过 Google 合规审核的 API 项目上传的视频会被强制锁为私享、无法改公开**，对个人/单频道不实用；
+> 浏览器自动化没有此限制，可直接发布公开视频，也与本项目其它平台的 cookie 方案一致。
+> `--playlist` 适合连载/系列追更；`--visibility` 可选 `public`/`unlisted`/`private`。
+> 上传会**等进度到 100% 再点发布**（浏览器上传靠窗口开着传，传一半就发布会被掐断卡在中途）。
+> youtube.com 被墙的地区：在 `conf.py` 设 `YT_PROXY = "http://127.0.0.1:7890"`（chromium 不吃系统代理，需显式指定）。
+
补充说明：
- `creator` 之类的名字只是示例值，真正含义是 `account_name`
diff --git a/conf.example.py b/conf.example.py
index 11e33c6..cf2e942 100644
--- a/conf.example.py
+++ b/conf.example.py
@@ -5,3 +5,7 @@ XHS_SERVER = "http://127.0.0.1:11901" # only used by xhs-related flows
LOCAL_CHROME_PATH = "" # optional, e.g. C:/Program Files/Google/Chrome/Application/chrome.exe
LOCAL_CHROME_HEADLESS = True # default headless behavior for uploader/examples
DEBUG_MODE = True # default debug behavior
+# Optional proxy for the YouTube uploader. Where youtube.com is blocked, direct
+# connections time out and the (patchright) chromium does NOT use the system proxy.
+# Point this at your local proxy port, e.g. "http://127.0.0.1:7890". None = no proxy.
+YT_PROXY = None
diff --git a/sau_cli.py b/sau_cli.py
index b4152aa..d0475e3 100644
--- a/sau_cli.py
+++ b/sau_cli.py
@@ -41,6 +41,11 @@ from uploader.xiaohongshu_uploader.main import (
cookie_auth as xiaohongshu_cookie_auth,
xiaohongshu_setup,
)
+from uploader.youtube_uploader.main import (
+ YouTubeVideo,
+ cookie_auth as youtube_cookie_auth,
+ youtube_setup,
+)
SCHEDULE_FORMAT = "%Y-%m-%d %H:%M"
@@ -160,6 +165,20 @@ class TencentVideoUploadRequest:
headless: bool = True
+@dataclass(slots=True)
+class YouTubeVideoUploadRequest:
+ account_name: str
+ video_file: Path
+ title: str
+ description: str
+ tags: list[str]
+ thumbnail_file: Path | None = None
+ playlist: str | None = None
+ visibility: str = "public"
+ debug: bool = True
+ headless: bool = False
+
+
def has_interactive_terminal() -> bool:
return sys.stdin.isatty() and sys.stdout.isatty()
@@ -274,6 +293,42 @@ async def check_tencent_account(account_name: str) -> bool:
return await tencent_cookie_auth(str(account_file))
+async def login_youtube_account(account_name: str, headless: bool = False) -> dict:
+ account_file = resolve_account_file("youtube", account_name)
+ return await youtube_setup(str(account_file), handle=True, return_detail=True, headless=headless)
+
+
+async def check_youtube_account(account_name: str) -> bool:
+ account_file = resolve_account_file("youtube", account_name)
+ if not account_file.exists():
+ return False
+ return await youtube_cookie_auth(str(account_file))
+
+
+async def upload_youtube_video(request: YouTubeVideoUploadRequest) -> Path:
+ account_file = resolve_account_file("youtube", request.account_name)
+ is_ready = await youtube_setup(str(account_file), handle=False)
+ if not is_ready:
+ raise RuntimeError(
+ f"YouTube cookie is missing or expired: {account_file}. Run `sau youtube login --account {request.account_name}` first."
+ )
+
+ app = YouTubeVideo(
+ request.title,
+ str(request.video_file),
+ request.tags,
+ str(account_file),
+ description=request.description,
+ thumbnail_path=str(request.thumbnail_file) if request.thumbnail_file else None,
+ playlist=request.playlist,
+ visibility=request.visibility,
+ debug=request.debug,
+ headless=request.headless,
+ )
+ await app.main()
+ return account_file
+
+
async def upload_video(request: DouyinVideoUploadRequest) -> Path:
account_file = resolve_account_file("douyin", request.account_name)
is_ready = await douyin_setup(str(account_file), handle=False)
@@ -646,6 +701,27 @@ def build_parser() -> argparse.ArgumentParser:
tencent_upload_video_parser.add_argument("--category", help="Optional original content category")
tencent_upload_video_parser.add_argument("--draft", action="store_true", help="Save as draft instead of publishing")
add_runtime_flags(tencent_upload_video_parser)
+
+ youtube_parser = platform_parsers.add_parser("youtube", help="YouTube operations")
+ youtube_actions = youtube_parser.add_subparsers(dest="action", required=True)
+
+ for action_name in ("login", "check"):
+ action_parser = youtube_actions.add_parser(action_name, help=f"YouTube {action_name}")
+ action_parser.add_argument("--account", required=True, help="YouTube user-defined account_name")
+ if action_name == "login":
+ add_runtime_flags(action_parser)
+
+ youtube_upload_video_parser = youtube_actions.add_parser("upload-video", help="Upload one video to YouTube")
+ youtube_upload_video_parser.add_argument("--account", required=True, help="YouTube user-defined account_name")
+ youtube_upload_video_parser.add_argument("--file", required=True, type=existing_file_path, help="Video file path")
+ youtube_upload_video_parser.add_argument("--title", required=True, help="Video title (<=100 chars)")
+ youtube_upload_video_parser.add_argument("--desc", default="", help="Optional video description")
+ youtube_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2")
+ youtube_upload_video_parser.add_argument("--thumbnail", type=existing_file_path, help="Optional thumbnail image path")
+ youtube_upload_video_parser.add_argument("--playlist", help="Optional playlist name to add the video to (for series)")
+ youtube_upload_video_parser.add_argument(
+ "--visibility", default="public", choices=["public", "unlisted", "private"], help="Video visibility")
+ add_runtime_flags(youtube_upload_video_parser)
return parser
@@ -882,6 +958,38 @@ async def dispatch(args: argparse.Namespace) -> int:
raise RuntimeError(f"Unsupported Tencent/WeChat Channels action: {args.action}")
+ if args.platform == "youtube":
+ if args.action == "login":
+ result = await login_youtube_account(args.account, headless=args.headless)
+ if not result["success"]:
+ raise RuntimeError(result["message"])
+ print(f"YouTube login flow completed: {result['account_file']}")
+ return 0
+
+ if args.action == "check":
+ is_valid = await check_youtube_account(args.account)
+ print("valid" if is_valid else "invalid")
+ return 0 if is_valid else 1
+
+ if args.action == "upload-video":
+ request = YouTubeVideoUploadRequest(
+ account_name=args.account,
+ video_file=args.file,
+ title=args.title,
+ description=args.desc,
+ tags=parse_tags(args.tags),
+ thumbnail_file=args.thumbnail,
+ playlist=args.playlist,
+ visibility=args.visibility,
+ debug=args.debug,
+ headless=args.headless,
+ )
+ await upload_youtube_video(request)
+ print(f"YouTube video upload submitted: {request.video_file}")
+ return 0
+
+ raise RuntimeError(f"Unsupported YouTube action: {args.action}")
+
raise RuntimeError(f"Unsupported platform: {args.platform}")
diff --git a/uploader/youtube_uploader/__init__.py b/uploader/youtube_uploader/__init__.py
new file mode 100644
index 0000000..69323f7
--- /dev/null
+++ b/uploader/youtube_uploader/__init__.py
@@ -0,0 +1,5 @@
+from pathlib import Path
+
+from conf import BASE_DIR
+
+Path(BASE_DIR / "cookies" / "youtube_uploader").mkdir(exist_ok=True)
diff --git a/uploader/youtube_uploader/main.py b/uploader/youtube_uploader/main.py
new file mode 100644
index 0000000..a036e24
--- /dev/null
+++ b/uploader/youtube_uploader/main.py
@@ -0,0 +1,307 @@
+# -*- coding: utf-8 -*-
+"""YouTube uploader (browser automation via YouTube Studio).
+
+Unlike the other platforms here, YouTube also offers an official Data API. We deliberately
+use browser automation instead, because videos uploaded through an *unaudited* API project
+are force-locked to private and cannot be made public without passing Google's compliance
+audit (which is impractical for personal/single-channel use). Browser automation has no such
+restriction and can publish public videos right away, and it matches the cookie-based pattern
+used by every other uploader in this project.
+
+Login is interactive (Google account, no QR code): the browser opens, the user signs in, and
+the storage_state is saved. Reuse it afterwards for fully unattended uploads.
+"""
+import asyncio
+from pathlib import Path
+
+from patchright.async_api import Page, Playwright, async_playwright
+
+from conf import DEBUG_MODE
+from uploader.base_video import BaseVideoUploader
+from utils.base_social_media import set_init_script
+from utils.log import youtube_logger
+
+try:
+ # 国内直连 youtube.com 会超时，且 patchright 启的 chromium 不吃系统代理。
+ # 在 conf.py 设 YT_PROXY = "http://127.0.0.1:7890"（本地代理端口）即可；不设则不走代理。
+ from conf import YT_PROXY
+except Exception:
+ YT_PROXY = None
+
+STUDIO_URL = "https://studio.youtube.com"
+UPLOAD_URL = "https://www.youtube.com/upload"
+VISIBILITY = {"public": "PUBLIC", "unlisted": "UNLISTED", "private": "PRIVATE"}
+
+
+def _msg(emoji: str, text: str) -> str:
+ return f"{emoji} {text}"
+
+
+def _build_login_result(success, status, message, account_file, current_url=""):
+ return {
+ "success": success,
+ "status": status,
+ "message": message,
+ "account_file": str(account_file),
+ "current_url": current_url,
+ }
+
+
+async def cookie_auth(account_file) -> bool:
+ """登录态是否仍有效：带 cookie 打开 Studio，没被踢到 Google 登录页且进入了频道页即有效。"""
+ async with async_playwright() as playwright:
+ browser = await playwright.chromium.launch(headless=True, channel="chrome")
+ try:
+ context = await browser.new_context(storage_state=account_file)
+ context = await set_init_script(context)
+ page = await context.new_page()
+ await page.goto(STUDIO_URL, wait_until="domcontentloaded")
+ await page.wait_for_timeout(3000)
+ url = page.url
+ if "accounts.google.com" in url or "/signin" in url.lower():
+ return False
+ return "/channel/" in url
+ except Exception:
+ return False
+ finally:
+ await browser.close()
+
+
+async def youtube_cookie_gen(account_file, headless: bool = False):
+ """交互式登录：开浏览器让用户登录 Google/YouTube，进入频道页后保存 storage_state。"""
+ async with async_playwright() as playwright:
+ # 登录必须显形，让用户输账号密码/二步验证
+ browser = await playwright.chromium.launch(headless=False, channel="chrome")
+ context = await browser.new_context()
+ context = await set_init_script(context)
+ page = await context.new_page()
+ await page.goto(STUDIO_URL, wait_until="domcontentloaded")
+ youtube_logger.info(_msg("🔐", "请在弹出的浏览器里登录 Google / YouTube 账号，登录后会自动保存"))
+ ok = False
+ for _ in range(600): # 最多等 10 分钟
+ if "/channel/" in page.url:
+ await page.wait_for_timeout(2000) # 让 cookie 落定
+ ok = True
+ break
+ await asyncio.sleep(1)
+ if ok:
+ await context.storage_state(path=account_file)
+ youtube_logger.success(_msg("✅", f"YouTube 登录态已保存: {account_file}"))
+ else:
+ youtube_logger.error(_msg("😵", "等待登录超时，未保存登录态"))
+ await browser.close()
+ return _build_login_result(ok, "logged_in" if ok else "timeout",
+ "登录成功" if ok else "登录超时", account_file, page.url)
+
+
+async def youtube_setup(account_file, handle: bool = False, return_detail: bool = False, headless: bool = False):
+ """校验登录态，失效且 handle=True 时拉起交互式登录。"""
+ if not Path(account_file).exists() or not await cookie_auth(account_file):
+ if not handle:
+ result = _build_login_result(False, "cookie_invalid", "登录态不存在或已失效", account_file)
+ return result if return_detail else False
+ youtube_logger.info(_msg("🥹", "YouTube 登录态不存在或失效，准备打开浏览器登录"))
+ result = await youtube_cookie_gen(account_file, headless=headless)
+ return result if return_detail else result["success"]
+ result = _build_login_result(True, "cookie_valid", "登录态有效", account_file)
+ return result if return_detail else True
+
+
+async def _fill_editable(page: Page, selector: str, text: str):
+ """填 YouTube Studio 的 contenteditable 富文本框（标题/简介），先清空再输入。"""
+ box = page.locator(selector).first
+ await box.wait_for(state="visible", timeout=30000)
+ await box.click()
+ await page.keyboard.press("Control+A")
+ await page.keyboard.press("Delete")
+ await box.type(text, delay=6)
+ await page.wait_for_timeout(400)
+
+
+async def _click_if_present(page: Page, selector: str, timeout: int = 4000) -> bool:
+ try:
+ el = page.locator(selector).first
+ await el.wait_for(state="visible", timeout=timeout)
+ await el.click()
+ return True
+ except Exception:
+ return False
+
+
+async def _wait_upload_complete(page: Page, max_polls: int = 360) -> bool:
+ """等网页上传从 X% 跑到 100% 再发布。浏览器上传靠窗口开着才传得完，
+ 若上传到一半就点发布并关闭浏览器，上传会被掐断卡在中途（如 76%）。
+ 出现“处理/检查/上传完成”或不再“正在上传”即视为传完。max_polls*5s=30min 上限。"""
+ last = ""
+ for _ in range(max_polls):
+ txt = ""
+ for sel in (".progress-label", "span.progress-label", "ytcp-video-upload-progress"):
+ loc = page.locator(sel).first
+ try:
+ if await loc.count():
+ txt = (await loc.inner_text()).strip()
+ if txt:
+ break
+ except Exception:
+ pass
+ if txt:
+ if any(k in txt for k in ("处理", "检查", "上传完成", "已上传", "Processing", "complete", "Checks", "Finished")):
+ youtube_logger.info(_msg("✅", f"上传完成: {txt[:40]}"))
+ return True
+ if txt != last:
+ youtube_logger.info(_msg("⏳", f"上传中: {txt[:40]}"))
+ last = txt
+ await page.wait_for_timeout(5000)
+ youtube_logger.warning(_msg("⚠️", "等上传超时(30min)，仍尝试发布"))
+ return False
+
+
+class YouTubeVideo(BaseVideoUploader):
+ def __init__(self, title, file_path, tags, account_file, *,
+ description="", thumbnail_path=None, playlist=None,
+ visibility="public", debug=DEBUG_MODE, headless=False):
+ self.title = title
+ self.file_path = str(file_path)
+ self.tags = tags or []
+ self.account_file = str(account_file)
+ self.description = description or ""
+ self.thumbnail_path = str(thumbnail_path) if thumbnail_path else None
+ self.playlist = playlist
+ self.visibility = visibility if visibility in VISIBILITY else "public"
+ self.debug = debug
+ self.headless = headless
+
+ async def upload(self, playwright: Playwright) -> None:
+ browser = await playwright.chromium.launch(
+ headless=self.headless, channel="chrome",
+ proxy={"server": YT_PROXY} if YT_PROXY else None,
+ )
+ context = await browser.new_context(storage_state=self.account_file)
+ context = await set_init_script(context)
+ page = await context.new_page()
+ page.set_default_timeout(60000)
+
+ youtube_logger.info(_msg("🎬", f"开始上传: {Path(self.file_path).name}"))
+ await page.goto(UPLOAD_URL, wait_until="domcontentloaded")
+ await page.wait_for_timeout(3000)
+ if "accounts.google.com" in page.url or "signin" in page.url.lower():
+ await browser.close()
+ raise RuntimeError("YouTube 登录态失效，请重新执行 login")
+
+ # 1) 选择视频文件
+ file_input = page.locator('input[type="file"]').first
+ await file_input.wait_for(state="attached", timeout=60000)
+ await file_input.set_input_files(self.file_path)
+
+ # 2) 等详情对话框
+ await page.locator("#title-textarea").wait_for(state="visible", timeout=120000)
+
+ # 3) 标题
+ youtube_logger.info(_msg("✍️", "填写标题"))
+ await _fill_editable(page, "#title-textarea #textbox", self.title[:100])
+
+ # 4) 简介
+ if self.description.strip():
+ youtube_logger.info(_msg("✍️", "填写简介"))
+ await _fill_editable(page, "#description-textarea #textbox", self.description)
+
+ # 5) 封面（处理到一定进度才允许传，失败不致命）
+ if self.thumbnail_path and Path(self.thumbnail_path).exists():
+ try:
+ thumb_input = page.locator(
+ "#file-loader input[type='file'], ytcp-thumbnail-uploader input[type='file']"
+ ).first
+ await thumb_input.wait_for(state="attached", timeout=20000)
+ await thumb_input.set_input_files(self.thumbnail_path)
+ await page.wait_for_timeout(2000)
+ youtube_logger.info(_msg("🖼️", "封面已上传"))
+ except Exception as exc:
+ youtube_logger.warning(_msg("⚠️", f"封面上传跳过（不影响发布）: {exc}"))
+
+ # 6) 加入播放列表（连载/系列追更）。弹窗务必关闭，否则挡住后续步骤。
+ if self.playlist:
+ try:
+ await _click_if_present(
+ page, "#basics ytcp-text-dropdown-trigger, ytcp-video-metadata-playlists ytcp-dropdown-trigger", 8000)
+ await page.wait_for_timeout(1200)
+ existing = page.locator(
+ f"tp-yt-paper-checkbox:has-text('{self.playlist}'), "
+ f"ytcp-checkbox-group:has-text('{self.playlist}')").first
+ if await existing.count():
+ await existing.click()
+ else:
+ if await _click_if_present(page, "ytcp-button:has-text('New playlist'), ytcp-button:has-text('创建播放列表')", 4000):
+ await page.wait_for_timeout(800)
+ await _click_if_present(page, "tp-yt-paper-item:has-text('New playlist'), tp-yt-paper-item:has-text('新建播放列表')", 3000)
+ title_box = page.locator("ytcp-playlist-metadata-editor #textbox, #create-playlist-form #textbox").first
+ if await title_box.count():
+ await title_box.click()
+ await title_box.type(self.playlist, delay=6)
+ await _click_if_present(page, "ytcp-button#create-button, tp-yt-paper-dialog ytcp-button:has-text('Create'), tp-yt-paper-dialog ytcp-button:has-text('创建')", 4000)
+ except Exception as exc:
+ youtube_logger.warning(_msg("⚠️", f"播放列表处理跳过（不影响发布）: {exc}"))
+ finally:
+ await _click_if_present(page, "ytcp-playlist-dialog #save-button, ytcp-button:has-text('Done'), ytcp-button:has-text('完成')", 3000)
+ await page.keyboard.press("Escape")
+ await page.wait_for_timeout(600)
+
+ # 7) 受众：非儿童向（必填）
+ if not await _click_if_present(page, "tp-yt-paper-radio-button[name='VIDEO_MADE_FOR_KIDS_NOT_MFK']", 10000):
+ await _click_if_present(page, "tp-yt-paper-radio-button:has-text('not made for kids'), tp-yt-paper-radio-button:has-text('不是面向儿童')", 6000)
+
+ # 8) 标签（“显示更多”里）
+ if self.tags:
+ try:
+ await _click_if_present(page, "#toggle-button", 6000)
+ await page.wait_for_timeout(800)
+ tag_input = page.locator("#tags-container #text-input, ytcp-form-input-container#tags-container input").first
+ await tag_input.click()
+ await tag_input.type(",".join(self.tags)[:500] + ",", delay=4)
+ except Exception as exc:
+ youtube_logger.warning(_msg("⚠️", f"标签填写跳过（不影响发布）: {exc}"))
+
+ # 9) 连点 Next 到“可见性”步骤
+ for _ in range(5):
+ vis = page.locator("tp-yt-paper-radio-button[name='PUBLIC']")
+ if await vis.count() and await vis.first.is_visible():
+ break
+ if not await _click_if_present(page, "#next-button", 6000):
+ await page.wait_for_timeout(1200)
+ await page.wait_for_timeout(1000)
+
+ # 10) 可见性
+ youtube_logger.info(_msg("🌐", f"设置可见性 = {self.visibility}"))
+ await _click_if_present(page, f"tp-yt-paper-radio-button[name='{VISIBILITY[self.visibility]}']", 10000)
+
+ # 10.5) 关键：等上传真正传完再发布。浏览器上传靠窗口开着传，
+ # 传到一半就点发布+关浏览器 = 上传被掐断卡在中途（如 76%）。
+ youtube_logger.info(_msg("📤", "等待上传完成（传完才发布）…"))
+ await _wait_upload_complete(page)
+
+ # 11) 发布
+ await page.wait_for_timeout(1200)
+ if not await _click_if_present(page, "#done-button", 15000):
+ youtube_logger.warning(_msg("🤔", "未找到发布按钮，可能上传未到可发布进度；请在窗口里手动发布"))
+ else:
+ await page.wait_for_timeout(4000)
+ video_url = ""
+ try:
+ link = page.locator("a[href*='youtu.be'], a[href*='watch?v=']").first
+ if await link.count():
+ video_url = await link.get_attribute("href") or ""
+ except Exception:
+ pass
+ await _click_if_present(page, "ytcp-button:has-text('Close'), ytcp-button:has-text('关闭'), #close-button", 8000)
+ youtube_logger.success(_msg("🥳", f"发布完成（{self.visibility}）{(' ' + video_url) if video_url else ''}"))
+
+ # 刷新 cookie
+ try:
+ await context.storage_state(path=self.account_file)
+ except Exception:
+ pass
+ await page.wait_for_timeout(2000)
+ await browser.close()
+
+ async def main(self):
+ async with async_playwright() as playwright:
+ await self.upload(playwright)
diff --git a/utils/log.py b/utils/log.py
index 3271da9..5a2de54 100644
--- a/utils/log.py
+++ b/utils/log.py
@@ -57,3 +57,4 @@ bilibili_logger = create_logger('bilibili', 'logs/bilibili.log')
kuaishou_logger = create_logger('kuaishou', 'logs/kuaishou.log')
baijiahao_logger = create_logger('baijiahao', 'logs/baijiahao.log')
xiaohongshu_logger = create_logger('xiaohongshu', 'logs/xiaohongshu.log')
+youtube_logger = create_logger('youtube', 'logs/youtube.log')