# -*- coding: utf-8 -*- from datetime import datetime from playwright.async_api import Page import pyperclip import asyncio from auto_media_publisher.uploaders.uploader_base import UploaderBase CONTENT_UPLOAD_URL="https://x.com/home" VIDEO_PUBLISH_SECCESSFULL_URL = "https://www.tiktok.com/tiktokstudio/content" class UploaderTwitterX(UploaderBase): def get_year_index(self,publish_year:int)-> int: current_year = datetime.now().year # 获取当前年份 base_year = current_year # 设定基准年份，例如当前年份 base_index = 3 # 当前年份对应序号为4 # 计算目标年份相对于当前年份的序号 return base_index - (publish_year - base_year) async def setPostScheduleDateTime(self, page): self.logger.info(f"start to set publish date time ...{datetime}") await self.set_single_option(page,'#SELECTOR_3',self.publish_datetime.year) # year await self.set_single_option(page,'#SELECTOR_1', self.publish_datetime.month) # month await self.set_single_option(page,'#SELECTOR_2', self.publish_datetime.day) # day await self.set_single_option(page,'#SELECTOR_4', self.publish_datetime.hour) # hour await self.set_single_option(page,'#SELECTOR_5', self.publish_datetime.minute) # min # 确认按钮 await page.click('span:text("确认")') await asyncio.sleep(1) async def set_single_option(self,page,selector,value): # Ensure the select element is visible and available select_locator = page.locator(selector) # Select the option by its value await select_locator.select_option(value=str(value)) # Select by value 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'[+]Uploading-------{self.title}.mp4') # 等待页面加载完成 await page.wait_for_url("**/home") if "/home" not in page.url: self.logger.info("账号未登录") self.logger.info(f"账号未登录:{self.cookie_path}") return False self.logger.info(f"twitterX {self.user} 账号已登录") # 发布视频流程 try: upload_button = await page.wait_for_selector('button[aria-label="添加照片或视频"], button[aria-label="添加媒体"]', timeout=10000) await upload_button.click() # 等待文件选择器出现并上传视频 file_input = await page.wait_for_selector('input[type="file"]', timeout=10000) await file_input.set_input_files(self.path_video) except Exception as e: self.logger.info("发布视频失败，可能网页加载失败了\n", e) self.logger.info("发布视频失败，可能网页加载失败了") self.logger.info("视频标题输入完毕，等待发布") await page.wait_for_timeout(5000) # 等待 5 秒 # 设置发布的时间 datetime_button = await page.wait_for_selector('button[aria-label="预排期帖子"]', timeout=10000) await datetime_button.click() await page.wait_for_timeout(2000) # 等待 2 秒 await self.setPostScheduleDateTime(page) video_desc = self.title+"\n" + ' '.join(['#' + tag for tag in self.tags]) pyperclip.copy(video_desc) # 输入视频描述 description_field = await page.wait_for_selector('[aria-label="帖子文本"] div div div', timeout=10000) await description_field.click() # 模拟 Ctrl + V 粘贴 await page.keyboard.down("Control") await page.keyboard.press("V") await page.keyboard.up("Control") await page.mouse.click(0, 0) # 点击发布按钮 publish_button_locator = page.locator('button:has-text("发帖"), button:has-text("安排表")') # 等待按钮可见并点击 await publish_button_locator.wait_for(state='visible', timeout=10000) # 等待最多 10 秒 await publish_button_locator.click() while True: try: button = page.locator('button:has-text("发帖")') # 最多等待 10 秒，每 0.5 秒检查一次 for _ in range(20): is_disabled = await button.get_attribute("disabled") if is_disabled is not None: self.logger.info("按钮已禁用，发帖成功") break await asyncio.sleep(0.5) else: self.logger.info("等待超时：按钮未变为禁用") self.logger.info(" [-] video published success") break except Exception as e: self.logger.exception(f" [-] Exception: {e}") self.logger.info(" [-] video publishing") await asyncio.sleep(0.5) self.logger.info("视频发布成功") await self.context.storage_state(path=f"{self.cookie_path}") # save cookie self.logger.info('twitterX [-] update cookie！') await asyncio.sleep(2) # close delay for look the video status return True except Exception as e: self.logger.info(f"异常类型: {type(e).__name__}") # 打印异常的类型 self.logger.info(f"异常消息: {str(e)}") # 打印异常消息 self.logger.info(f"异常详细信息: {repr(e)}") # 打印异常的完整表示 self.logger.info(f"异常的所有属性:", {e.__dict__}) self.logger.info("视频发布失败，请管理员处理。") return False