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 18c40f0..e5c8727 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"
@@ -161,6 +166,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()
@@ -275,6 +294,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)
@@ -650,6 +705,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
@@ -896,6 +972,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/douyin_uploader/main.py b/uploader/douyin_uploader/main.py
index 38bd9d0..5a76a3c 100644
--- a/uploader/douyin_uploader/main.py
+++ b/uploader/douyin_uploader/main.py
@@ -50,12 +50,16 @@ def _build_login_result(success: bool, status: str, message: str, account_file:
async def cookie_auth(account_file):
async with async_playwright() as playwright:
- browser = await playwright.chromium.launch(headless=True, channel="chrome")
+ # 抖音无头会撞反爬墙→content/upload 跳登录→误判 cookie 失效（间歇性）。校验必须有头。
+ browser = await playwright.chromium.launch(
+ headless=False, channel="chrome",
+ args=["--no-sandbox", "--disable-blink-features=AutomationControlled"],
+ )
try:
context = await browser.new_context(storage_state=account_file)
context = await set_init_script(context)
page = await context.new_page()
- await page.goto("https://creator.douyin.com/creator-micro/content/upload")
+ await page.goto("https://creator.douyin.com/creator-micro/content/upload", wait_until="domcontentloaded", timeout=90000)
try:
await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=5000)
except Exception:
@@ -106,7 +110,12 @@ async def _extract_douyin_qrcode_src(page: Page) -> str:
async def _save_douyin_qrcode(page: Page, account_file: str, previous_qrcode_path: Path | None = None, qrcode_callback=None) -> dict:
- qrcode_src = await _extract_douyin_qrcode_src(page)
+ # 提取二维码 src 仅为了保存/终端显示；定位不到时不致命——有头浏览器里二维码可见，直接扫码即可
+ try:
+ qrcode_src = await _extract_douyin_qrcode_src(page)
+ except Exception as exc:
+ douyin_logger.warning(_msg("😵", f"没定位到二维码元素（{str(exc)[:50]}）——请直接在弹出的浏览器里扫码，小人继续等登录跳转"))
+ return {"image_path": "", "image_data_url": ""}
qrcode_path = save_data_url_image(qrcode_src, build_login_qrcode_path(account_file))
if previous_qrcode_path and previous_qrcode_path != qrcode_path:
if remove_qrcode_file(previous_qrcode_path):
@@ -126,7 +135,8 @@ async def _save_douyin_qrcode(page: Page, account_file: str, previous_qrcode_pat
async def _is_douyin_login_completed(page: Page) -> bool:
- if not page.url.startswith("https://creator.douyin.com/creator-micro/home"):
+ # 登录后会跳到 creator-micro 下任意页（home/content 等）；登录页是 creator.douyin.com/ 根路径
+ if "creator.douyin.com/creator-micro" not in page.url:
return False
login_markers = [
@@ -149,7 +159,7 @@ async def _is_douyin_login_completed(page: Page) -> bool:
async def _wait_for_douyin_login(page: Page, account_file: str, qrcode_info: dict, qrcode_callback=None, poll_interval: int = 3, max_checks: int = 100) -> dict:
- qrcode_path = Path(qrcode_info["image_path"])
+ qrcode_path = Path(qrcode_info["image_path"]) if qrcode_info.get("image_path") else None
for _ in range(max_checks):
if await _is_douyin_login_completed(page):
douyin_logger.info(_msg("🥳", f"扫码成功，已经跳转到登录后页面: {page.url}"))
@@ -161,7 +171,7 @@ async def _wait_for_douyin_login(page: Page, account_file: str, qrcode_info: dic
await expired_box.click()
await asyncio.sleep(1)
qrcode_info = await _save_douyin_qrcode(page, account_file, qrcode_path, qrcode_callback=qrcode_callback)
- qrcode_path = Path(qrcode_info["image_path"])
+ qrcode_path = Path(qrcode_info["image_path"]) if qrcode_info.get("image_path") else None
await asyncio.sleep(poll_interval)
@@ -185,7 +195,7 @@ async def douyin_cookie_gen(
page = await context.new_page()
await page.goto("https://creator.douyin.com/")
qrcode_info = await _save_douyin_qrcode(page, account_file, qrcode_callback=qrcode_callback)
- qrcode_path = Path(qrcode_info["image_path"])
+ qrcode_path = Path(qrcode_info["image_path"]) if qrcode_info.get("image_path") else None
douyin_logger.info(_msg("🧍", "请扫码，小人正在耐心等待登录完成"))
result = await _wait_for_douyin_login(
page,
@@ -263,30 +273,22 @@ class DouYinBaseUploader(BaseVideoUploader):
await asyncio.sleep(1)
async def fill_title_and_description(self, page: Page, title: str, description: str, tags: list[str] | None = None):
- description_section = (
- page.get_by_text("作品描述", exact=True)
- .locator("xpath=ancestor::div[2]")
- .locator("xpath=following-sibling::div[1]")
- )
-
- title_input = description_section.locator('input[type="text"]').first
- await title_input.wait_for(state="visible", timeout=10000)
+ # 2026-06 抖音发布页 DOM：标题=input[placeholder*=填写作品标题]，描述=div.zone-container[contenteditable]
+ # version_2(post/video) 发布页要等视频上传完才渲染表单（实测约 40s），故等待超时给到 120s
+ title_input = page.locator('input[placeholder*="填写作品标题"]').first
+ await title_input.wait_for(state="visible", timeout=120000)
await title_input.fill(title[:30])
- description_editor = description_section.locator('.zone-container[contenteditable="true"]').first
- await description_editor.wait_for(state="visible", timeout=10000)
+ description_editor = page.locator('div.zone-container[contenteditable="true"]').first
+ await description_editor.wait_for(state="visible", timeout=120000)
await description_editor.click()
await page.keyboard.press("Control+KeyA")
await page.keyboard.press("Delete")
- # tags 合并到 description 结尾，一起用 textContent 赋值
- if tags:
- tags_text = " ".join(f"#{t}" for t in tags)
- full_text = description + "\n\n" + tags_text
- else:
- full_text = description
-
- await description_editor.evaluate('(el, text) => el.textContent = text', full_text)
+ for tag in tags or []:
+ await page.keyboard.type(" #" + tag)
+ await page.keyboard.press("Space")
+ await page.keyboard.press("Escape") # 收起话题下拉，避免浮层拦截后续点击
async def set_location(self, page: Page, location: str = ""):
if not location:
@@ -541,29 +543,46 @@ class DouYinVideo(DouYinBaseUploader):
return
douyin_logger.info(_msg("🏃", "小人正在设置视频封面"))
- await page.click('text="选择封面"')
- cover_locator_str = 'div[id*="creator-content-modal"]'
- cover_locator = page.locator(cover_locator_str)
- await page.wait_for_selector(cover_locator_str)
+ # 先清掉 shepherd 新手引导浮层，否则它会拦截“选择封面”点击导致弹窗打不开
+ await page.evaluate(
+ "() => document.querySelectorAll('.shepherd-element,.shepherd-modal-overlay-container').forEach(e=>e.remove())"
+ )
+ await page.get_by_text("选择封面", exact=True).first.click(force=True)
+ cover_locator_str = 'div.dy-creator-content-modal'
+ cover_locator = page.locator(cover_locator_str).first
+ await page.wait_for_selector(cover_locator_str, timeout=20000)
- upload_input = cover_locator.locator("div[class^='semi-upload upload'] >> input.semi-upload-hidden-input")
-
- if self.thumbnail_landscape_path:
- await page.wait_for_timeout(1000)
- await upload_input.set_input_files(self.thumbnail_landscape_path)
- await page.wait_for_timeout(2000)
- douyin_logger.info(_msg("🖼️", "横版封面上传完成"))
+ await page.wait_for_timeout(1500)
+ # version_2 封面弹窗有 4 个隐藏 file input：
+ # [0]/[1] 左侧“AI生成参考图”上传/替换，[2]/[3] 才是“上传封面”/替换。
+ # 旧代码用 .first 传到了 AI 参考图（不会成为封面）→ 这就是“传了却没封面”的根因。
+ # 取 input.semi-upload-hidden-input 的第 2 个（nth(1)），即真正的封面上传输入。
+ cover_upload = cover_locator.locator("input.semi-upload-hidden-input").nth(1)
if self.thumbnail_portrait_path:
- await cover_locator.locator("div[class*='steps'] div").nth(1).click()
- await page.wait_for_timeout(1000)
- await upload_input.set_input_files(self.thumbnail_portrait_path)
- await page.wait_for_timeout(2000)
- douyin_logger.info(_msg("🖼️", "竖版封面上传完成"))
+ # 弹窗默认就在“设置竖封面”页；防御性点一下 tab（已激活则忽略）
+ try:
+ await cover_locator.get_by_text("设置竖封面", exact=True).first.click(timeout=3000)
+ await page.wait_for_timeout(800)
+ except Exception:
+ pass
+ await cover_upload.set_input_files(self.thumbnail_portrait_path)
+ await page.wait_for_timeout(3000)
+ douyin_logger.info(_msg("🖼️", "竖版封面已上传到预览"))
+ elif self.thumbnail_landscape_path:
+ try:
+ await cover_locator.get_by_text("设置横封面", exact=True).first.click(timeout=3000)
+ await page.wait_for_timeout(800)
+ except Exception:
+ pass
+ await cover_upload.set_input_files(self.thumbnail_landscape_path)
+ await page.wait_for_timeout(3000)
+ douyin_logger.info(_msg("🖼️", "横版封面已上传到预览"))
- await cover_locator.locator('button:visible:has-text("完成")').click()
+ # 点红色主按钮“完成”应用封面（exact 避免误中“完成编辑”）
+ await cover_locator.get_by_role("button", name="完成", exact=True).first.click()
douyin_logger.info(_msg("🥳", "视频封面设置完成"))
- await page.wait_for_selector("div.extractFooter", state="detached")
+ await cover_locator.wait_for(state="detached", timeout=20000)
async def upload(self, playwright: Playwright) -> None:
douyin_logger.info(_msg("🧍", "小人先检查 cookie、视频文件、封面和发布时间"))
@@ -578,10 +597,12 @@ class DouYinVideo(DouYinBaseUploader):
context = await set_init_script(context)
page = await context.new_page()
- await page.goto("https://creator.douyin.com/creator-micro/content/upload")
+ await page.goto("https://creator.douyin.com/creator-micro/content/upload", wait_until="domcontentloaded", timeout=90000)
douyin_logger.info(_msg("🏃", f"小人开始搬运视频: {self.title}.mp4"))
douyin_logger.info(_msg("🧭", "小人正在赶往上传主页"))
- await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload")
+ await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=90000)
+ # wait_for_url 完成时上传页可能尚未渲染出文件 input（实测偶发），先等它挂载再 set_input_files
+ await page.wait_for_selector("div[class^='container'] input", state="attached", timeout=60000)
await page.locator("div[class^='container'] input").set_input_files(self.file_path)
while True:
@@ -643,9 +664,13 @@ class DouYinVideo(DouYinBaseUploader):
while True:
try:
+ # 移除会拦截发布按钮点击的新手引导/话题下拉浮层
+ await page.evaluate(
+ "() => { document.querySelectorAll('.shepherd-element, .shepherd-modal-overlay-container, [class*=\"mention-wrapper\"]').forEach(e => e.remove()); }"
+ )
publish_button = page.get_by_role("button", name="发布", exact=True)
if await publish_button.count():
- await publish_button.click()
+ await publish_button.click(force=True)
await page.wait_for_url(
"https://creator.douyin.com/creator-micro/content/manage**",
timeout=3000,
@@ -792,9 +817,9 @@ class DouYinNote(DouYinBaseUploader):
upload_success = False
try:
page = await context.new_page()
- await page.goto("https://creator.douyin.com/creator-micro/content/upload")
+ await page.goto("https://creator.douyin.com/creator-micro/content/upload", wait_until="domcontentloaded", timeout=90000)
douyin_logger.info(_msg("🧭", "小人正在赶往图文发布页"))
- await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload")
+ await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=90000)
await self.upload_note_content(page)
upload_success = True
diff --git a/uploader/tencent_uploader/main.py b/uploader/tencent_uploader/main.py
index 94cded1..32feaa2 100644
--- a/uploader/tencent_uploader/main.py
+++ b/uploader/tencent_uploader/main.py
@@ -492,15 +492,44 @@ class TencentBaseUploader(BaseVideoUploader):
await page.click('input[placeholder="请选择时间"]')
await page.keyboard.press("Control+KeyA")
await page.keyboard.type(publish_date.strftime("%H"))
- await page.locator("div.input-editor").click()
+ await page.keyboard.press("Enter") # 确认小时并关闭时间下拉
+ await page.wait_for_timeout(500)
+ # 收起时间选择浮层：直接点描述区可能被 weui-desktop-dialog 遮挡，做容错
+ try:
+ await page.locator("div.input-editor").click(timeout=5000)
+ except Exception:
+ await page.keyboard.press("Escape")
async def open_upload_page(self, page: Page) -> None:
- await page.goto(TENCENT_UPLOAD_URL)
- await page.wait_for_url(TENCENT_UPLOAD_URL)
+ await page.goto(TENCENT_UPLOAD_URL, timeout=120000, wait_until="domcontentloaded")
+ await page.wait_for_url(TENCENT_UPLOAD_URL, timeout=120000)
async def upload_video_file(self, page: Page, file_path: str) -> None:
- file_input = page.locator('input[type="file"]')
- await file_input.set_input_files(file_path)
+ async def find_file_input():
+ for fr in page.frames: # 主 frame + 所有 iframe（视频号编辑器可能在 iframe 内）
+ try:
+ fi = fr.locator('input[type="file"]')
+ if await fi.count():
+ return fi.first
+ except Exception:
+ continue
+ return None
+
+ fi = await find_file_input()
+ if fi is None:
+ # 助手落在首页：先点「发表视频」唤出编辑器与上传控件
+ publish_btn = page.get_by_text("发表视频").first
+ if await publish_btn.count():
+ await publish_btn.click()
+ await asyncio.sleep(3)
+ for _ in range(20):
+ fi = await find_file_input()
+ if fi is not None:
+ break
+ await asyncio.sleep(1)
+ if fi is None:
+ raise RuntimeError("未找到视频号文件上传框")
+ await fi.set_input_files(file_path)
async def set_short_title(self, page: Page, title: str, short_title: str | None = None) -> None:
short_title_element = (
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')