# -*- coding: utf-8 -*- from playwright.async_api import Page import asyncio import traceback from auto_media_publisher.uploaders.uploader_base import UploaderBase CONTENT_UPLOAD_URL="https://creator.douyin.com/creator-micro/content/upload" VIDEO_PUBLISH_SECCESSFULL_URL = "https://creator.douyin.com/creator-micro/content/manage?enter_from=publish" class UploaderDouyin(UploaderBase): async def set_schedule_time_douyin(self, page): try: # 选择包含特定文本内容的 label 元素 label_element = page.locator("[class^='radio']:has-text('定时发布')") # 在选中的 label 元素下点击 checkbox await label_element.click() await asyncio.sleep(1) scheduled_time_hour = self.scheduled_time.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(scheduled_time_hour)) await page.keyboard.press("Enter") await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 发布时间 发生错误：{e}') async def _try_set_cover_image(self, page: Page): try: if self.path_cover: await page.click('text="选择封面"') await page.locator("#dy-creator-content-modal-body").wait_for(timeout=30000) await page.click('text="设置竖封面"') await page.wait_for_timeout(2000) # 等待2秒 # 定位到上传区域并点击 await page.locator("div[class^='semi-upload upload'] >> input.semi-upload-hidden-input").set_input_files(self.path_cover) await page.wait_for_timeout(3000) # 等待2秒 await page.locator("button:visible:has-text('完成')").click() except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}') async def _try_set_location(self, page: Page): try: # 1. 打开下拉框 await page.locator(".anchor-container-hgj7gj > div > .semi-select-arrow").click() await asyncio.sleep(1) # 这段可以保留，也可以换成 wait_for await page.get_by_role("option", name="位置").click() await page.locator('div.semi-select span:has-text("输入地理位置")').click() await page.keyboard.press("Control+A") # 推荐比按Backspace更可靠 await page.keyboard.press("Backspace") # 4. 输入地理位置 await page.wait_for_timeout(1000) await page.keyboard.type(self.location) await page.wait_for_timeout(2000) # 可以尝试替换成更精确的 wait_for() 检查 listbox 是否加载 listbox = page.get_by_role("listbox") target = listbox.locator("div").filter(has_text=self.location).nth(2) await target.click() except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 位置 发生错误：{e}') async def handle_upload_error(self, page): self.logger.info('视频出错了，重新上传中') await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.path_media) async def _try_upload_mp4_and_wait(self,page): try: await page.locator("div[class^='container'] input").set_input_files(self.path_media) while True: # 判断重新上传按钮是否存在，如果不存在，代表视频正在上传，则等待 try: number = await page.locator('[class^="long-card"] div:has-text("重新上传")').count() if number > 0: self.logger.info(" [-]视频上传完毕") break else: self.logger.info(" [-] 正在上传视频中...") await asyncio.sleep(2) if await page.locator('div.progress-div > div:has-text("上传失败")').count(): self.logger.error(" [-] 发现上传出错了... 准备重试") await self.handle_upload_error(page) except Exception as e: self.logger.error(f" [exception] 正在上传视频中... 错误：{e}") self.logger.debug(traceback.format_exc()) # 打印堆栈信息（如果 logger 支持 debug） await asyncio.sleep(2) return True except Exception as e: self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}") return False async def _try_set_tilte_and_description(self,page): try: await asyncio.sleep(1) self.logger.info(f' [-] 正在填充标题和话题...') await page.get_by_placeholder("填写作品标题，为作品获得更多流量").click() await page.get_by_placeholder("填写作品标题，为作品获得更多流量").fill(self.title[:30]) await page.locator(".zone-container").click() description_tags = self.description + ' ' + ' '.join(f"#{tag}" for tag in self.tags) await page.locator(".zone-container").fill(description_tags) self.logger.info(f'总共添加{len(self.tags)}个话题') except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置title 发生错误：{e}') async def _upload_core(self) -> bool: try: page = await self.context.new_page() await page.goto(CONTENT_UPLOAD_URL, timeout=120000) self.logger.info(f'[+]正在上传-------{self.title}.mp4') await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000) # 1、上传并等待视频上传结束 await self._try_upload_mp4_and_wait(page) # 2、设置标题和描述 await self._try_set_tilte_and_description(page) # 3、设置封面 await self._try_set_cover_image(page) # 4、设置位置 和 商品 await self._try_set_location(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() # 5、设置定时发布 if self.scheduled_time != 0: await self.set_schedule_time_douyin(page) # 6 发布并判断视频是否发布成功 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(VIDEO_PUBLISH_SECCESSFULL_URL,timeout=3000) # 如果自动跳转到作品页面，则代表发布成功 self.logger.info(" [-]视频发布成功") break except: self.logger.info(" [-] 视频正在发布中...") await page.screenshot(full_page=True) await asyncio.sleep(0.5) # 7、 更新Cookie信息，退出 await self.context.storage_state(path=self.cookie_path) # 保存cookie self.logger.info(f' {self.user} , {self.platform} [-]cookie更新完毕！') await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 return True except Exception as e: self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}") return False