# -*- 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.scheduled_time.year) # year
|
await self.set_single_option(page,'#SELECTOR_1', self.scheduled_time.month) # month
|
await self.set_single_option(page,'#SELECTOR_2', self.scheduled_time.day) # day
|
await self.set_single_option(page,'#SELECTOR_4', self.scheduled_time.hour) # hour
|
await self.set_single_option(page,'#SELECTOR_5', self.scheduled_time.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_media)
|
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
|
|
