196 lines
8.0 KiB
Python
196 lines
8.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
from datetime import datetime
|
|
|
|
from playwright.async_api import Playwright, async_playwright
|
|
import os
|
|
import asyncio
|
|
|
|
from utils.base_social_media import set_init_script
|
|
from utils.files_times import get_absolute_path
|
|
from utils.log import kuaishou_logger
|
|
|
|
|
|
async def cookie_auth(account_file):
|
|
async with async_playwright() as playwright:
|
|
browser = await playwright.chromium.launch(headless=True)
|
|
context = await browser.new_context(storage_state=account_file)
|
|
context = await set_init_script(context)
|
|
# 创建一个新的页面
|
|
page = await context.new_page()
|
|
# 访问指定的 URL
|
|
await page.goto("https://cp.kuaishou.com/article/publish/video")
|
|
try:
|
|
await page.wait_for_selector("div.names div.container div.name:text('机构服务')", timeout=5000) # 等待5秒
|
|
|
|
kuaishou_logger.info("[+] 等待5秒 cookie 失效")
|
|
return False
|
|
except:
|
|
kuaishou_logger.success("[+] cookie 有效")
|
|
return True
|
|
|
|
|
|
async def ks_setup(account_file, handle=False):
|
|
account_file = get_absolute_path(account_file, "ks_uploader")
|
|
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
|
if not handle:
|
|
return False
|
|
kuaishou_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
|
await get_ks_cookie(account_file)
|
|
return True
|
|
|
|
|
|
async def get_ks_cookie(account_file):
|
|
async with async_playwright() as playwright:
|
|
options = {
|
|
'args': [
|
|
'--lang en-GB'
|
|
],
|
|
'headless': False, # Set headless option here
|
|
}
|
|
# Make sure to run headed.
|
|
browser = await playwright.chromium.launch(**options)
|
|
# Setup context however you like.
|
|
context = await browser.new_context() # Pass any options
|
|
context = await set_init_script(context)
|
|
# Pause the page, and start recording manually.
|
|
page = await context.new_page()
|
|
await page.goto("https://cp.kuaishou.com")
|
|
await page.pause()
|
|
# 点击调试器的继续，保存cookie
|
|
await context.storage_state(path=account_file)
|
|
|
|
|
|
class KSVideo(object):
|
|
def __init__(self, title, file_path, tags, publish_date: datetime, account_file):
|
|
self.title = title # 视频标题
|
|
self.file_path = file_path
|
|
self.tags = tags
|
|
self.publish_date = publish_date
|
|
self.account_file = account_file
|
|
self.date_format = '%Y-%m-%d %H:%M'
|
|
|
|
async def handle_upload_error(self, page):
|
|
kuaishou_logger.error("视频出错了，重新上传中")
|
|
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path)
|
|
|
|
async def upload(self, playwright: Playwright) -> None:
|
|
# 使用 Chromium 浏览器启动一个浏览器实例
|
|
browser = await playwright.chromium.launch(headless=False)
|
|
# 创建一个浏览器上下文，使用指定的 cookie 文件
|
|
context = await browser.new_context(storage_state=f"{self.account_file}")
|
|
context = await set_init_script(context)
|
|
|
|
# 创建一个新的页面
|
|
page = await context.new_page()
|
|
# 访问指定的 URL
|
|
await page.goto("https://cp.kuaishou.com/article/publish/video")
|
|
kuaishou_logger.info('正在上传-------{}.mp4'.format(self.title))
|
|
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
|
|
kuaishou_logger.info('正在打开主页...')
|
|
await page.wait_for_url("https://cp.kuaishou.com/article/publish/video")
|
|
# 点击 "上传视频" 按钮
|
|
upload_button = page.locator("button[class^='_upload-btn']")
|
|
await upload_button.wait_for(state='visible') # 确保按钮可见
|
|
|
|
async with page.expect_file_chooser() as fc_info:
|
|
await upload_button.click()
|
|
file_chooser = await fc_info.value
|
|
await file_chooser.set_files(self.file_path)
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
if not await page.get_by_text("封面编辑").count():
|
|
raise Exception("似乎没有跳转到到编辑页面")
|
|
|
|
await asyncio.sleep(1)
|
|
|
|
# 等待按钮可交互
|
|
new_feature_button = page.locator('button[type="button"] span:text("我知道了")')
|
|
if await new_feature_button.count() > 0:
|
|
await new_feature_button.click()
|
|
|
|
kuaishou_logger.info("正在填充标题和话题...")
|
|
await page.get_by_text('填写描述').locator("xpath=following-sibling::div").click()
|
|
kuaishou_logger.info("clear existing title")
|
|
await page.keyboard.press("Backspace")
|
|
await page.keyboard.press("Control+KeyA")
|
|
await page.keyboard.press("Delete")
|
|
kuaishou_logger.info("filling new title")
|
|
await page.keyboard.type(self.title)
|
|
await page.keyboard.press("Enter")
|
|
|
|
# 快手只能添加3个话题
|
|
for index, tag in enumerate(self.tags[:3], start=1):
|
|
kuaishou_logger.info("正在添加第%s个话题" % index)
|
|
await page.locator('span:text("#话题")').click()
|
|
await page.type('div.clGhv3UpdEo-', tag, delay=100)
|
|
await asyncio.sleep(2)
|
|
await page.locator('div.FZcv90s7kFs- > div').nth(0).click()
|
|
|
|
while True:
|
|
try:
|
|
number = await page.locator('div > span:text("上传成功")').count()
|
|
if number > 0:
|
|
kuaishou_logger.success("视频上传完毕")
|
|
break
|
|
else:
|
|
kuaishou_logger.info("正在上传视频中...")
|
|
await asyncio.sleep(2)
|
|
except:
|
|
kuaishou_logger.info("正在上传视频中...")
|
|
await asyncio.sleep(2)
|
|
|
|
# 定时任务
|
|
if self.publish_date != 0:
|
|
await self.set_schedule_time(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 asyncio.sleep(1)
|
|
confirm_button = page.locator("button > span:text('确认发布')")
|
|
if await confirm_button.count():
|
|
await page.locator("button > span:text('确认发布')").click()
|
|
|
|
await page.wait_for_url("https://cp.kuaishou.com/article/manage/video?status=2&from=publish",
|
|
timeout=1500)
|
|
kuaishou_logger.success("视频发布成功")
|
|
break
|
|
except:
|
|
kuaishou_logger.info("视频正在发布中...")
|
|
await page.screenshot(full_page=True)
|
|
await asyncio.sleep(0.5)
|
|
|
|
await context.storage_state(path=self.account_file) # 保存cookie
|
|
kuaishou_logger.info('cookie更新完毕！')
|
|
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
|
# 关闭浏览器上下文和浏览器实例
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
async def main(self):
|
|
async with async_playwright() as playwright:
|
|
await self.upload(playwright)
|
|
|
|
async def set_schedule_time(self, page, publish_date):
|
|
kuaishou_logger.info("click schedule")
|
|
publish_date_hour = publish_date.strftime("%Y-%m-%d %H:%M:%S")
|
|
await page.locator("label:text('发布时间')").locator('xpath=following-sibling::div').locator(
|
|
'.ant-radio-input').nth(1).click()
|
|
await asyncio.sleep(1)
|
|
|
|
await page.locator('div.ant-picker-input input[placeholder="选择日期时间"]').click()
|
|
await asyncio.sleep(1)
|
|
|
|
await page.keyboard.press("Control+KeyA")
|
|
await page.keyboard.type(str(publish_date_hour))
|
|
await page.keyboard.press("Enter")
|
|
await asyncio.sleep(1)
|
|
|
|
|