# -*- coding: utf-8 -*- from datetime import datetime import asyncio import inspect import os from pathlib import Path from patchright.async_api import Page from patchright.async_api import Playwright from patchright.async_api import async_playwright from conf import DEBUG_MODE, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH from uploader.base_video import BaseVideoUploader from utils.base_social_media import set_init_script from utils.login_qrcode import build_login_qrcode_path from utils.login_qrcode import decode_qrcode_from_path from utils.login_qrcode import print_terminal_qrcode from utils.login_qrcode import remove_qrcode_file from utils.login_qrcode import save_data_url_image from utils.log import douyin_logger DOUYIN_PUBLISH_STRATEGY_IMMEDIATE = "immediate" DOUYIN_PUBLISH_STRATEGY_SCHEDULED = "scheduled" def _msg(emoji: str, text: str) -> str: return f"{emoji} {text}" async def _emit_qrcode_callback(qrcode_callback, payload: dict): if not qrcode_callback: return callback_result = qrcode_callback(payload) if inspect.isawaitable(callback_result): await callback_result def _build_login_result(success: bool, status: str, message: str, account_file: str, qrcode: dict | None = None, current_url: str = "") -> dict: return { "success": success, "status": status, "message": message, "account_file": str(account_file), "qrcode": qrcode, "current_url": current_url, } async def cookie_auth(account_file): # 抖音无头会撞反爬墙→content/upload 跳登录→误判 cookie 失效（间歇性）。校验必须有头。 # 即便有头，页面慢/瞬时跳转仍会让 wait_for_url(精确URL,5s) 误判→重试3次+宽松判定(URL含 content/upload 且无登录文案)。 # 允许 linux server 用户通过 env var 强制无头: DOUYIN_COOKIE_AUTH_HEADLESS=true use_headless = os.environ.get("DOUYIN_COOKIE_AUTH_HEADLESS", "").lower() in ("1", "true", "yes") launch_kwargs = {"headless": use_headless, "channel": "chrome", "args": ["--no-sandbox", "--disable-blink-features=AutomationControlled"]} for _attempt in range(3): async with async_playwright() as playwright: browser = await playwright.chromium.launch(**launch_kwargs) 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", wait_until="domcontentloaded", timeout=90000) await page.wait_for_timeout(2500) # 等页面稳定，避免瞬时跳转误判 has_login = await page.get_by_text("手机号登录").count() or await page.get_by_text("扫码登录").count() if "content/upload" in page.url and not has_login: return True except Exception: pass finally: await browser.close() return False async def douyin_setup(account_file, handle=False, return_detail=False, qrcode_callback=None, headless: bool = LOCAL_CHROME_HEADLESS, cdp_url: str | None = None): if not os.path.exists(account_file) or not await cookie_auth(account_file): if not handle: result = _build_login_result(False, "cookie_invalid", "cookie文件不存在或已失效", account_file) return result if return_detail else False douyin_logger.info(_msg("🥹", "cookie 失效了，准备打开浏览器重新登录")) result = await douyin_cookie_gen(account_file, qrcode_callback=qrcode_callback, headless=headless, cdp_url=cdp_url) return result if return_detail else result["success"] result = _build_login_result(True, "cookie_valid", "cookie有效", account_file) return result if return_detail else True async def _extract_douyin_qrcode_src(page: Page) -> str: # 等 SPA 加载完成（不只等"扫码登录"文字，否则抖音慢加载时 30s 就超时）。 # 给 domcontentloaded 后足够时间让客户端 JS 注入登录卡。 try: await page.wait_for_load_state("networkidle", timeout=15000) except Exception: pass scan_login_tab = page.get_by_text("扫码登录", exact=True).first # attached 状态：DOM 里出现即可，不要求 visible/渲染完整，避免 race await scan_login_tab.wait_for(state="attached", timeout=60000) # 新版抖音创作者中心 (single_tab + animate_qrcode_container) 不再用 aria-label="二维码"。 # 按优先级兜底多个 selector，至少一个能命中即可。 qrcode_selectors = [ 'div#animate_qrcode_container img[src^="data:image"]', 'div[class*="animate_qrcode_container"] img[src^="data:image"]', 'div[class*="scan_qrcode_login_content"] img[src^="data:image"]', 'img[aria-label="二维码"]', ] last_err: Exception | None = None for sel in qrcode_selectors: qrcode_img = page.locator(sel).first try: await qrcode_img.wait_for(state="attached", timeout=10000) except Exception as e: last_err = e continue src = await qrcode_img.get_attribute("src") if src: return src last_err = RuntimeError(f"selector {sel} 命中但 src 为空") raise RuntimeError(f"未获取到抖音登录二维码地址 (last_err={last_err})") async def _save_douyin_qrcode(page: Page, account_file: str, previous_qrcode_path: Path | None = None, qrcode_callback=None) -> dict: # 提取二维码 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): douyin_logger.info(_msg("🧹", f"临时二维码文件已清理: {previous_qrcode_path}")) douyin_logger.info(_msg("🖼️", f"二维码已经准备好啦，已保存到: {qrcode_path}")) qrcode_content = decode_qrcode_from_path(qrcode_path) if qrcode_content: print_terminal_qrcode(qrcode_content, qrcode_path, "抖音APP") else: douyin_logger.warning(_msg("😵", f"终端没法完整显示二维码，请打开 {qrcode_path} 扫码")) qrcode_info = { "image_path": str(qrcode_path), "image_data_url": qrcode_src, } await _emit_qrcode_callback(qrcode_callback, qrcode_info) return qrcode_info async def _is_douyin_login_completed(page: Page) -> bool: # 登录后会跳到 creator-micro 下任意页（home/content 等）；登录页是 creator.douyin.com/ 根路径 if "creator.douyin.com/creator-micro" not in page.url: return False login_markers = [ page.get_by_text("扫码登录", exact=True).first, page.get_by_text("手机号登录", exact=True).first, page.get_by_text("二维码失效", exact=True).first, page.get_by_role("img", name="二维码").first, ] for marker in login_markers: if not await marker.count(): continue try: if await marker.is_visible(): return False except Exception: continue return True 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"]) 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}")) return _build_login_result(True, "success", "抖音扫码登录成功", account_file, qrcode_info, page.url) # URL 变化 + sessionid 未到位 → 二验流程，继续等 if page.url != original_url and not await _is_douyin_login_completed(page): sms_input = page.locator('input[placeholder*="验证码"], input[type="tel"], input[placeholder*="短信"], input[placeholder*="手机号"]') if await sms_input.count() > 0: if not saw_2fa: douyin_logger.warning(_msg("⚠️", f"检测到抖音短信/安全二次验证，请在弹出的浏览器中手动输入。等待 sessionid ({i}/{max_checks})")) saw_2fa = True await asyncio.sleep(poll_interval) continue expired_box = page.get_by_text("二维码失效", exact=True).locator("..").first if await expired_box.count() and await expired_box.is_visible(): douyin_logger.warning(_msg("😵", "二维码失效了，小人马上去刷新")) 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"]) if qrcode_info.get("image_path") else None await asyncio.sleep(poll_interval) return _build_login_result(False, "timeout", "等待抖音扫码登录超时", account_file, qrcode_info, page.url) async def douyin_cookie_gen( account_file, qrcode_callback=None, poll_interval: int = 2, max_checks: int = 60, headless: bool = LOCAL_CHROME_HEADLESS, cdp_url: str | None = None, ): async with async_playwright() as playwright: if cdp_url: browser = await playwright.chromium.connect_over_cdp(cdp_url) context = browser.contexts[0] if browser.contexts else await browser.new_context() should_close_context = False else: browser = await playwright.chromium.launch(headless=headless, channel="chromium") context = await browser.new_context() should_close_context = True context = await set_init_script(context) qrcode_path = None result = _build_login_result(False, "failed", "抖音登录失败", account_file) try: 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"]) if qrcode_info.get("image_path") else None douyin_logger.info(_msg("🧍", "请扫码，小人正在耐心等待登录完成")) result = await _wait_for_douyin_login( page, account_file, qrcode_info, qrcode_callback=qrcode_callback, poll_interval=poll_interval, max_checks=max_checks, ) if result["success"]: await asyncio.sleep(2) await context.storage_state(path=account_file) if not await cookie_auth(account_file): result = _build_login_result( False, "cookie_invalid", "抖音扫码流程结束，但 cookie 校验失败", account_file, qrcode_info, page.url, ) except Exception as exc: result = _build_login_result(False, "failed", str(exc), account_file, current_url=page.url if "page" in locals() else "") finally: if remove_qrcode_file(qrcode_path): douyin_logger.info(_msg("🧹", f"临时二维码文件已清理: {qrcode_path}")) if not result["success"]: douyin_logger.error(_msg("😢", f"登录失败: {result['message']}")) if should_close_context: await context.close() await browser.close() return result class DouYinBaseUploader(BaseVideoUploader): def __init__( self, publish_date: datetime | int, account_file, publish_strategy: str = DOUYIN_PUBLISH_STRATEGY_IMMEDIATE, debug: bool = DEBUG_MODE, headless: bool = LOCAL_CHROME_HEADLESS, ): self.publish_date = publish_date self.account_file = account_file self.publish_strategy = publish_strategy self.debug = debug self.date_format = "%Y年%m月%d日 %H:%M" self.local_executable_path = LOCAL_CHROME_PATH self.headless = headless async def validate_base_args(self): if not os.path.exists(self.account_file): raise RuntimeError(f"cookie文件不存在，请先完成抖音登录: {self.account_file}") if not await cookie_auth(self.account_file): raise RuntimeError(f"cookie文件已失效，请先完成抖音登录: {self.account_file}") if self.publish_strategy not in {DOUYIN_PUBLISH_STRATEGY_IMMEDIATE, DOUYIN_PUBLISH_STRATEGY_SCHEDULED}: raise ValueError(f"不支持的发布策略: {self.publish_strategy}") if self.publish_strategy == DOUYIN_PUBLISH_STRATEGY_SCHEDULED: self.publish_date = self.validate_publish_date(self.publish_date) else: self.publish_date = 0 async def set_schedule_time_douyin(self, page, publish_date): label_element = page.locator("[class^='radio']:has-text('定时发布')") await label_element.click() await asyncio.sleep(1) publish_date_hour = publish_date.strftime("%Y-%m-%d %H:%M") await asyncio.sleep(1) await page.locator('.semi-input[placeholder="日期和时间"]').click() await page.keyboard.press("Control+KeyA") await page.keyboard.type(str(publish_date_hour)) await page.keyboard.press("Enter") await asyncio.sleep(1) async def fill_title_and_description(self, page: Page, title: str, description: str, tags: list[str] | None = None): # 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 = 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") 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: return await page.locator('div.semi-select span:has-text("输入地理位置")').click() await page.keyboard.press("Backspace") await page.wait_for_timeout(2000) await page.keyboard.type(location) await page.wait_for_selector('div[role="listbox"] [role="option"]', timeout=5000) await page.locator('div[role="listbox"] [role="option"]').first.click() async def handle_product_dialog(self, page: Page, product_title: str): await page.wait_for_timeout(2000) await page.wait_for_selector('input[placeholder="请输入商品短标题"]', timeout=10000) short_title_input = page.locator('input[placeholder="请输入商品短标题"]') if not await short_title_input.count(): douyin_logger.error(_msg("😵", "没找到商品短标题输入框")) return False product_title = product_title[:10] await short_title_input.fill(product_title) await page.wait_for_timeout(1000) finish_button = page.locator('button:has-text("完成编辑")') if "disabled" not in await finish_button.get_attribute("class"): await finish_button.click() douyin_logger.debug(_msg("🥳", "已点击“完成编辑”按钮")) await page.wait_for_selector(".semi-modal-content", state="hidden", timeout=5000) return True douyin_logger.error(_msg("😵", "“完成编辑”按钮是灰的，小人先把弹窗关掉")) cancel_button = page.locator('button:has-text("取消")') if await cancel_button.count(): await cancel_button.click() else: close_button = page.locator(".semi-modal-close") await close_button.click() await page.wait_for_selector(".semi-modal-content", state="hidden", timeout=5000) return False async def set_product_link(self, page: Page, product_link: str, product_title: str): await page.wait_for_timeout(2000) try: await page.wait_for_selector("text=添加标签", timeout=10000) dropdown = page.get_by_text("添加标签").locator("..").locator("..").locator("..").locator(".semi-select").first if not await dropdown.count(): douyin_logger.error(_msg("😵", "没找到标签下拉框")) return False douyin_logger.debug(_msg("🧍", "找到标签下拉框，小人准备选择“购物车”")) await dropdown.click() await page.wait_for_selector('[role="listbox"]', timeout=5000) await page.locator('[role="option"]:has-text("购物车")').click() douyin_logger.debug(_msg("🥳", "已经选中“购物车”")) await page.wait_for_selector('input[placeholder="粘贴商品链接"]', timeout=5000) input_field = page.locator('input[placeholder="粘贴商品链接"]') await input_field.fill(product_link) douyin_logger.debug(_msg("🔗", f"商品链接已经填好了: {product_link}")) add_button = page.locator('span:has-text("添加链接")') button_class = await add_button.get_attribute("class") if "disable" in button_class: douyin_logger.error(_msg("😵", "“添加链接”按钮现在点不了")) return False await add_button.click() douyin_logger.debug(_msg("🥳", "已点击“添加链接”按钮")) await page.wait_for_timeout(2000) error_modal = page.locator("text=未搜索到对应商品") if await error_modal.count(): confirm_button = page.locator('button:has-text("确定")') await confirm_button.click() douyin_logger.error(_msg("😢", "这个商品链接无效")) return False if not await self.handle_product_dialog(page, product_title): return False douyin_logger.debug(_msg("🥳", "商品链接设置好了")) return True except Exception as e: douyin_logger.error(_msg("😢", f"设置商品链接时出错: {str(e)}")) return False async def set_self_declaration(self, page: Page, declaration: str = "内容为个人观点或见解") -> None: """抖音「自主声明」为发布必选项：打开声明弹窗 → 选指定类型 → 确定。 入口和弹窗都是异步渲染，等不到就记 warning 跳过、继续发布，绝不因此中断 （与小红书话题、视频号声明原创的容错策略保持一致）。 """ try: # 发布页底部「自主声明」行，未选时显示占位文案「请选择自主声明」 entry = page.get_by_text("请选择自主声明").first await entry.wait_for(state="visible", timeout=6000) await entry.click() # 弹窗标题「对作品内容添加声明」 dialog = page.locator(".semi-modal-content").filter(has_text="对作品内容添加声明").first await dialog.wait_for(state="visible", timeout=6000) # 单选项：Semi 的文字是 .semi-radio-addon（常带 pointer-events:none，直接点会卡 30s 超时）， # 要点可交互的 .semi-radio 外层；找不到外层再退回 force 强制点文字。exact 避免误命中预览「作者声明：…」。 option = dialog.locator(".semi-radio").filter(has_text=declaration).first if await option.count(): await option.click(timeout=6000) else: await dialog.get_by_text(declaration, exact=True).first.click(timeout=6000, force=True) await dialog.get_by_role("button", name="确定").click(timeout=6000) await dialog.wait_for(state="hidden", timeout=6000) douyin_logger.info(_msg("🧾", f"自主声明已选择「{declaration}」")) except Exception as exc: douyin_logger.warning(_msg("🧾", f"自主声明设置失败，跳过该步骤继续发布：{exc}")) async def select_bgm(self, page: Page, bgm_name: str) -> bool: """为图文发布选择 BGM：可选增强功能，搜索无结果或异常均跳过不中断发布。""" try: # 点击「选择音乐」按钮 music_entry = page.locator('text="选择音乐"').nth(1) if not await music_entry.count(): music_entry = page.locator('text="选择音乐"').first await music_entry.wait_for(state="visible", timeout=10000) await music_entry.click() # 等待侧边栏出现并搜索 sidesheet = page.locator(".semi-sidesheet-content").first await sidesheet.wait_for(state="visible", timeout=8000) search_input = sidesheet.locator('input.semi-input[placeholder="搜索音乐"]').first await search_input.wait_for(state="visible", timeout=5000) await search_input.fill(bgm_name) await search_input.press("Enter") # 等待搜索结果 await asyncio.sleep(2) first_card = sidesheet.locator(".card-container-tmocjc").first try: await first_card.wait_for(state="visible", timeout=8000) except Exception: douyin_logger.warning(_msg("🎵", f"音乐「{bgm_name}」搜索结果为空，小人跳过")) await self._close_music_sidesheet(page) return False # 打印找到的音乐名称 try: song_name_el = first_card.locator(".song-name-oRge4d").first if await song_name_el.count(): song_name = await song_name_el.inner_text() douyin_logger.info(_msg("🎵", f"小人找到了: {song_name}")) except Exception: pass # JS 点击「使用」（按钮 visibility:hidden，普通 click 无效） apply_btn = first_card.locator(".apply-btn-LUPP0D").first await apply_btn.evaluate("el => el.click()") douyin_logger.info(_msg("🥳", f"BGM「{bgm_name}」已应用")) # 等待侧边栏关闭，超时则手动关闭 try: await sidesheet.wait_for(state="hidden", timeout=5000) except Exception: await self._close_music_sidesheet(page) return True except Exception as exc: douyin_logger.warning(_msg("🎵", f"添加 BGM 时出错，跳过该步骤继续发布：{exc}")) try: await self._close_music_sidesheet(page) except Exception: pass return False async def _close_music_sidesheet(self, page: Page) -> None: try: close_btn = page.locator(".semi-sidesheet-close").first if await close_btn.count() and await close_btn.is_visible(): await close_btn.click() await asyncio.sleep(1) except Exception: pass class DouYinVideo(DouYinBaseUploader): def __init__( self, title, file_path, tags, publish_date: datetime | int, account_file, thumbnail_landscape_path=None, productLink="", productTitle="", thumbnail_portrait_path=None, desc: str | None = None, publish_strategy: str = DOUYIN_PUBLISH_STRATEGY_IMMEDIATE, debug: bool = DEBUG_MODE, headless: bool = LOCAL_CHROME_HEADLESS, ): super().__init__( publish_date=publish_date, account_file=account_file, publish_strategy=publish_strategy, debug=debug, headless=headless, ) self.title = title self.file_path = file_path self.tags = tags self.thumbnail_landscape_path = thumbnail_landscape_path self.thumbnail_portrait_path = thumbnail_portrait_path self.productLink = productLink self.productTitle = productTitle self.desc = desc or "" async def validate_upload_args(self): await self.validate_base_args() if not self.title or not str(self.title).strip(): raise ValueError("视频模式下，title 是必须的") self.file_path = str(self.validate_video_file(self.file_path)) if self.thumbnail_landscape_path: self.thumbnail_landscape_path = str(self.validate_image_file(self.thumbnail_landscape_path)) if self.thumbnail_portrait_path: self.thumbnail_portrait_path = str(self.validate_image_file(self.thumbnail_portrait_path)) async def handle_upload_error(self, page): douyin_logger.warning(_msg("😵", "视频上传摔了一跤，小人马上重新上传")) await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path) async def handle_auto_video_cover(self, page): if await page.get_by_text("请设置封面后再发布").first.is_visible(): douyin_logger.info(_msg("🧍", "发布前还得先把封面弄好")) recommend_cover = page.locator('[class^="recommendCover-"]').first if await recommend_cover.count(): douyin_logger.info(_msg("🏃", "小人去选第一个推荐封面")) try: await recommend_cover.click() await asyncio.sleep(1) confirm_text = "是否确认应用此封面？" if await page.get_by_text(confirm_text).first.is_visible(): douyin_logger.info(_msg("🪟", f"弹出确认框了: {confirm_text}")) await page.get_by_role("button", name="确定").click() douyin_logger.info(_msg("🥳", "推荐封面已经应用")) await asyncio.sleep(1) douyin_logger.info(_msg("🥳", "封面选择流程完成")) return True except Exception as e: douyin_logger.warning(_msg("😵", f"推荐封面没选成功: {e}")) return False async def set_thumbnail(self, page: Page): if not self.thumbnail_landscape_path and not self.thumbnail_portrait_path: return douyin_logger.info(_msg("🏃", "小人正在设置视频封面")) # 先清掉 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) 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: # 弹窗默认就在“设置竖封面”页；防御性点一下 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("🖼️", "横版封面已上传到预览")) # 点红色主按钮“完成”应用封面（exact 避免误中“完成编辑”） await cover_locator.get_by_role("button", name="完成", exact=True).first.click() douyin_logger.info(_msg("🥳", "视频封面设置完成")) await cover_locator.wait_for(state="detached", timeout=20000) async def upload(self, playwright: Playwright) -> None: douyin_logger.info(_msg("🧍", "小人先检查 cookie、视频文件、封面和发布时间")) await self.validate_upload_args() douyin_logger.info(_msg("🥳", "上传前检查通过")) browser = await playwright.chromium.launch(headless=self.headless, channel="chromium") context = await browser.new_context( storage_state=f"{self.account_file}", permissions=["geolocation"], ) context = await set_init_script(context) page = await context.new_page() 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", 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: try: await page.wait_for_url( "https://creator.douyin.com/creator-micro/content/publish?enter_from=publish_page", timeout=3000, ) douyin_logger.info(_msg("🥳", "已经进入 version_1 发布页面")) break except Exception: try: await page.wait_for_url( "https://creator.douyin.com/creator-micro/content/post/video?enter_from=publish_page", timeout=3000, ) douyin_logger.info(_msg("🥳", "已经进入 version_2 发布页面")) break except Exception: douyin_logger.debug(_msg("🧍", "还没进到视频发布页面，小人继续等一会")) await asyncio.sleep(0.5) await asyncio.sleep(1) douyin_logger.info(_msg("✍️", "小人开始填标题、描述和话题")) await self.fill_title_and_description(page, self.title, self.desc or self.title, self.tags) douyin_logger.info(_msg("🏷️", f"小人一共贴了 {len(self.tags)} 个话题")) while True: try: number = await page.locator('[class^="long-card"] div:has-text("重新上传")').count() if number > 0: douyin_logger.success(_msg("🥳", "视频已经传完啦")) break douyin_logger.info(_msg("🏃", "小人正在努力上传视频")) await asyncio.sleep(2) if await page.locator('div.progress-div > div:has-text("上传失败")').count(): douyin_logger.error(_msg("😵", "检测到上传失败，小人准备重试")) await self.handle_upload_error(page) except Exception: douyin_logger.debug(_msg("🧍", "小人还在等视频上传完成")) await asyncio.sleep(2) if self.productLink and self.productTitle: douyin_logger.info(_msg("🛒", "小人正在设置商品链接")) await self.set_product_link(page, self.productLink, self.productTitle) douyin_logger.info(_msg("🥳", "商品链接设置完成")) await self.set_thumbnail(page) await self.set_self_declaration(page) third_part_element = '[class^="info"] > [class^="first-part"] div div.semi-switch' if await page.locator(third_part_element).count(): if "semi-switch-checked" not in await page.eval_on_selector(third_part_element, "div => div.className"): await page.locator(third_part_element).locator("input.semi-switch-native-control").click() if self.publish_strategy == DOUYIN_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: await self.set_schedule_time_douyin(page, self.publish_date) 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(force=True) await page.wait_for_url( "https://creator.douyin.com/creator-micro/content/manage**", timeout=3000, ) douyin_logger.success(_msg("🥳", "视频发布成功，小人开心收工")) break except Exception: await self.handle_auto_video_cover(page) douyin_logger.info(_msg("🏃", "小人正在冲刺发布视频")) if self.debug: await page.screenshot(full_page=True) await asyncio.sleep(0.5) await context.storage_state(path=self.account_file) douyin_logger.success(_msg("🥳", "cookie 更新完毕")) await asyncio.sleep(2) await context.close() await browser.close() async def douyin_upload_video(self): async with async_playwright() as playwright: await self.upload(playwright) async def main(self): await self.douyin_upload_video() class DouYinNote(DouYinBaseUploader): def __init__( self, image_paths, note, tags, publish_date: datetime | int, account_file, title: str | None = None, publish_strategy: str = DOUYIN_PUBLISH_STRATEGY_IMMEDIATE, debug: bool = DEBUG_MODE, headless: bool = LOCAL_CHROME_HEADLESS, bgm: str = "", ): super().__init__( publish_date=publish_date, account_file=account_file, publish_strategy=publish_strategy, debug=debug, headless=headless, ) self.image_paths = image_paths self.note = note or "" self.title = title or (self.note[:30] if self.note else "") self.tags = tags or [] self.bgm = bgm or "" async def validate_upload_args(self): await self.validate_base_args() if not self.title or not str(self.title).strip(): raise ValueError("图文模式下，title 是必须的") if len(self.title) > 20: raise ValueError(f"标题不能超过20字符，当前: {len(self.title)}字符") if not self.image_paths: raise ValueError("图文模式下，图片是必须的") if isinstance(self.image_paths, (str, Path)): self.image_paths = [self.image_paths] if len(self.image_paths) > 35: raise ValueError("图文模式下最多只支持上传 35 张图片") note_len = len(self.note) if self.note else 0 if note_len > 1000: raise ValueError(f"正文不能超过1000字符，当前: {note_len}字符") normalized_image_paths = [] for image_path in self.image_paths: normalized_image_paths.append(str(self.validate_image_file(image_path))) self.image_paths = normalized_image_paths async def upload_note_content(self, page: Page) -> None: douyin_logger.info(_msg("🏃", f"小人开始搬运图文，共 {len(self.image_paths)} 张图片")) douyin_logger.info(_msg("🔀", "小人正在切换到图文发布")) await page.get_by_text("发布图文", exact=True).click() await page.wait_for_timeout(1000) douyin_logger.info(_msg("📤", "小人正在上传图片")) await page.locator("div[class^='container'] input[accept*='image']").set_input_files(self.image_paths) while True: try: await page.wait_for_url( "**/creator-micro/content/post/image?**", timeout=3000, ) douyin_logger.info(_msg("🥳", "已经进入图文发布页面")) break except Exception: douyin_logger.debug(_msg("🧍", "小人还在等图片上传完成")) await asyncio.sleep(0.5) await asyncio.sleep(1) douyin_logger.info(_msg("✍️", "小人开始填标题、描述和话题")) await self.fill_title_and_description(page, self.title, self.note, self.tags) title_len = len(self.title) if self.title else 0 tags_text = " ".join(f"#{t}" for t in self.tags) if self.tags else "" desc_and_tags_len = len(self.note or "") + (len(tags_text) + 2 if self.tags else 0) douyin_logger.info(_msg("📝", f"标题总字数: {title_len}，描述+话题总字数: {desc_and_tags_len}")) douyin_logger.info(_msg("🏷️", f"小人一共贴了 {len(self.tags)} 个话题")) if self.bgm: await self.select_bgm(page, self.bgm) if self.publish_strategy == DOUYIN_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: await self.set_schedule_time_douyin(page, self.publish_date) while True: try: publish_button = page.get_by_role("button", name="发布", exact=True) if await publish_button.count(): await publish_button.click() await page.wait_for_url( "**/creator-micro/content/manage?enter_from=publish**", timeout=3000, ) douyin_logger.success(_msg("🥳", "图文发布成功，小人开心收工")) break except Exception: douyin_logger.info(_msg("🏃", "小人正在冲刺发布图文")) await asyncio.sleep(0.5) async def upload(self, playwright: Playwright) -> None: douyin_logger.info(_msg("🧍", "小人先检查 cookie、图片和发布时间")) await self.validate_upload_args() douyin_logger.info(_msg("🥳", "图文上传前检查通过")) browser = await playwright.chromium.launch(headless=self.headless, channel="chromium") context = await browser.new_context( storage_state=f"{self.account_file}", permissions=["geolocation"], ) context = await set_init_script(context) upload_success = False try: page = await context.new_page() 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", timeout=90000) await self.upload_note_content(page) upload_success = True finally: if upload_success: await context.storage_state(path=self.account_file) douyin_logger.success(_msg("🥳", "cookie 更新完毕")) await asyncio.sleep(2) await context.close() await browser.close() async def douyin_upload_note(self): async with async_playwright() as playwright: await self.upload(playwright)