main.py
8.89 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
# -*- coding: utf-8 -*-
from datetime import datetime
from playwright.async_api import Playwright, async_playwright
import os
import asyncio
from conf import LOCAL_CHROME_PATH
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 asyncio.sleep(1)
await page.wait_for_selector(".publish-button",timeout=50000)
print("快手登录成功")
await asyncio.sleep(3)
# 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'
self.local_executable_path = LOCAL_CHROME_PATH
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 浏览器启动一个浏览器实例
print(self.local_executable_path)
if self.local_executable_path:
browser = await playwright.chromium.launch(
headless=False,
executable_path=self.local_executable_path,
)
else:
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)
context.on("close", lambda: context.storage_state(path=self.account_file))
# 创建一个新的页面
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.keyboard.type(f"#{tag} ")
await asyncio.sleep(2)
max_retries = 60 # 设置最大重试次数,最大等待时间为 2 分钟
retry_count = 0
while retry_count < max_retries:
try:
# 获取包含 '上传中' 文本的元素数量
number = await page.locator("text=上传中").count()
if number == 0:
kuaishou_logger.success("视频上传完毕")
break
else:
if retry_count % 5 == 0:
kuaishou_logger.info("正在上传视频中...")
await asyncio.sleep(2)
except Exception as e:
kuaishou_logger.error(f"检查上传状态时发生错误: {e}")
await asyncio.sleep(2) # 等待 2 秒后重试
retry_count += 1
if retry_count == max_retries:
kuaishou_logger.warning("超过最大重试次数，视频上传可能未完成。")
# 定时任务
if self.publish_date != 0:
await self.set_schedule_time(page, self.publish_date)
# 判断视频是否发布成功
while True:
try:
publish_button = page.get_by_text("发布", exact=True)
if await publish_button.count() > 0:
await publish_button.click()
await asyncio.sleep(1)
confirm_button = page.get_by_text("确认发布")
if await confirm_button.count() > 0:
await confirm_button.click()
# 等待页面跳转，确认发布成功
await page.wait_for_url(
"https://cp.kuaishou.com/article/manage/video?status=2&from=publish",
timeout=5000,
)
kuaishou_logger.success("视频发布成功")
break
except Exception as e:
kuaishou_logger.info(f"视频正在发布中... 错误: {e}")
await page.screenshot(full_page=True)
await asyncio.sleep(1)
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)