# -*- 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" 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): async with async_playwright() as playwright: browser = await playwright.chromium.launch(headless=True, channel="chrome") 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") try: await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=5000) except Exception: print("[+] 等待5秒 cookie 失效") await context.close() await browser.close() return False if await page.get_by_text("手机号登录").count() or await page.get_by_text("扫码登录").count(): print("[+] 等待5秒 cookie 失效") return False print("[+] cookie 有效") return True async def douyin_setup(account_file, handle=False, return_detail=False, qrcode_callback=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("[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件") result = await douyin_cookie_gen(account_file, qrcode_callback=qrcode_callback) 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: scan_login_tab = page.get_by_text("扫码登录", exact=True).first await scan_login_tab.wait_for(timeout=30000) qrcode_img = ( scan_login_tab .locator("..") .locator("xpath=following-sibling::div[1]") .locator('img[aria-label="二维码"]') .first ) if not await qrcode_img.count(): qrcode_img = page.get_by_role("img", name="二维码").first await qrcode_img.wait_for(state="visible", timeout=30000) src = await qrcode_img.get_attribute("src") if not src: raise RuntimeError("未获取到抖音登录二维码地址") return src 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) 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(f"[+] 已删除临时二维码文件: {previous_qrcode_path}") douyin_logger.info(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(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 _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"]) for _ in range(max_checks): if page.url.startswith("https://creator.douyin.com/creator-micro/home"): douyin_logger.info(f"[+] 检测到已跳转到登录后页面: {page.url}") return _build_login_result(True, "success", "抖音扫码登录成功", account_file, qrcode_info, page.url) expired_box = page.get_by_text("二维码失效", exact=True).locator("..").first if await expired_box.count() and await expired_box.is_visible(): douyin_logger.warning("[!] 二维码失效，正在刷新二维码...") 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"]) 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 = 3, max_checks: int = 100): async with async_playwright() as playwright: browser = await playwright.chromium.launch(headless=True, channel="chrome") context = await browser.new_context() 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"]) douyin_logger.info("[+] 等待扫码登录完成...") 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) 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(f"[+] 已删除临时二维码文件: {qrcode_path}") 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): 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) 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) await description_editor.click() await page.keyboard.press("Control+KeyA") await page.keyboard.press("Delete") await page.keyboard.type(description) for tag in tags or []: await page.keyboard.type(" #" + tag) await page.keyboard.press("Space") 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("[-] 未找到商品短标题输入框") 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("[+] 成功点击'完成编辑'按钮") await page.wait_for_selector(".semi-modal-content", state="hidden", timeout=5000) return True douyin_logger.error("[-] '完成编辑'按钮处于禁用状态，尝试直接关闭对话框") 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("[-] 未找到标签下拉框") return False douyin_logger.debug("[-] 找到标签下拉框，准备选择'购物车'") await dropdown.click() await page.wait_for_selector('[role="listbox"]', timeout=5000) await page.locator('[role="option"]:has-text("购物车")').click() douyin_logger.debug("[+] 成功选择'购物车'") await page.wait_for_selector('input[placeholder="粘贴商品链接"]', timeout=5000) input_field = page.locator('input[placeholder="粘贴商品链接"]') await input_field.fill(product_link) douyin_logger.debug(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("[-] '添加链接'按钮不可用") return False await add_button.click() douyin_logger.debug("[+] 成功点击'添加链接'按钮") 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("[-] 商品链接无效") return False if not await self.handle_product_dialog(page, product_title): return False douyin_logger.debug("[+] 成功设置商品链接") return True except Exception as e: douyin_logger.error(f"[-] 设置商品链接时出错: {str(e)}") return False 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, 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 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.info("视频出错了，重新上传中") 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(): print(" [-] 检测到需要设置封面提示...") recommend_cover = page.locator('[class^="recommendCover-"]').first if await recommend_cover.count(): print(" [-] 正在选择第一个推荐封面...") try: await recommend_cover.click() await asyncio.sleep(1) confirm_text = "是否确认应用此封面？" if await page.get_by_text(confirm_text).first.is_visible(): print(f" [-] 检测到确认弹窗: {confirm_text}") await page.get_by_role("button", name="确定").click() print(" [-] 已点击确认应用封面") await asyncio.sleep(1) print(" [-] 已完成封面选择流程") return True except Exception as e: print(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(" [-] 正在设置视频封面...") 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) 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(" [-] 横版封面上传完成") 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(" [-] 竖版封面上传完成") await cover_locator.locator('button:visible:has-text("完成")').click() douyin_logger.info(" [+] 视频封面设置完成！") await page.wait_for_selector("div.extractFooter", state="detached") async def upload(self, playwright: Playwright) -> None: douyin_logger.info("[-] 正在校验 cookie、视频文件、封面和发布时间...") await self.validate_upload_args() douyin_logger.info("[+] 上传前校验通过") browser = await playwright.chromium.launch(headless=self.headless, channel="chrome") 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") douyin_logger.info(f"[+]正在上传-------{self.title}.mp4") douyin_logger.info("[-] 正在打开主页...") await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload") 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("[+] 成功进入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("[+] 成功进入version_2发布页面!") break except Exception: print(" [-] 超时未进入视频发布页面，重新尝试...") await asyncio.sleep(0.5) await asyncio.sleep(1) douyin_logger.info(" [-] 正在填充标题、描述和话题...") await self.fill_title_and_description(page, self.title, self.title, self.tags) douyin_logger.info(f"总共添加{len(self.tags)}个话题") while True: try: number = await page.locator('[class^="long-card"] div:has-text("重新上传")').count() if number > 0: douyin_logger.success(" [-]视频上传完毕") break douyin_logger.info(" [-] 正在上传视频中...") await asyncio.sleep(2) if await page.locator('div.progress-div > div:has-text("上传失败")').count(): douyin_logger.error(" [-] 发现上传出错了... 准备重试") await self.handle_upload_error(page) except Exception: douyin_logger.info(" [-] 正在上传视频中...") await asyncio.sleep(2) if self.productLink and self.productTitle: douyin_logger.info(" [-] 正在设置商品链接...") await self.set_product_link(page, self.productLink, self.productTitle) douyin_logger.info(" [+] 完成设置商品链接...") await self.set_thumbnail(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: publish_button = page.get_by_role("button", name="发布", exact=True) if await publish_button.count(): await publish_button.click() await page.wait_for_url( "https://creator.douyin.com/creator-micro/content/manage**", timeout=3000, ) douyin_logger.success(" [-]视频发布成功") break except Exception: await self.handle_auto_video_cover(page) douyin_logger.info(" [-] 视频正在发布中...") 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(" [-]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, 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.image_paths = image_paths self.note = note self.tags = tags or [] async def validate_upload_args(self): await self.validate_base_args() if not self.note or not str(self.note).strip(): raise ValueError("图文模式下，note 是必须的") 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 张图片") 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(f"[+]正在上传图文，共 {len(self.image_paths)} 张图片") douyin_logger.info("[-] 正在切换到图文发布...") await page.get_by_text("发布图文", exact=True).click() await page.wait_for_timeout(1000) douyin_logger.info("[-] 正在上传图片...") 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("[+] 成功进入图文发布页面!") break except Exception: douyin_logger.info(" [-] 正在上传图片中...") await asyncio.sleep(0.5) await asyncio.sleep(1) douyin_logger.info(" [-] 正在填充标题、描述和话题...") await self.fill_title_and_description(page, self.note, self.note, self.tags) douyin_logger.info(f"总共添加{len(self.tags)}个话题") 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(" [-]图文发布成功") break except Exception: douyin_logger.info(" [-] 图文正在发布中...") await asyncio.sleep(0.5) async def upload(self, playwright: Playwright) -> None: douyin_logger.info("[-] 正在校验 cookie、图片和发布时间...") await self.validate_upload_args() douyin_logger.info("[+] 图文上传前校验通过") browser = await playwright.chromium.launch(headless=self.headless, channel="chrome") 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") douyin_logger.info("[-] 正在打开图文发布页...") await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload") await self.upload_note_content(page) upload_success = True finally: if upload_success: await context.storage_state(path=self.account_file) douyin_logger.success(" [-]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)