diff --git a/examples/get_xiaohongshu_cookie.py b/examples/get_xiaohongshu_cookie.py index 03eaa23..5c065cb 100644 --- a/examples/get_xiaohongshu_cookie.py +++ b/examples/get_xiaohongshu_cookie.py @@ -7,4 +7,10 @@ from uploader.xiaohongshu_uploader.main import xiaohongshu_setup if __name__ == '__main__': account_file = Path(BASE_DIR / "cookies" / "xiaohongshu_uploader" / "account.json") account_file.parent.mkdir(exist_ok=True) - cookie_setup = asyncio.run(xiaohongshu_setup(str(account_file), handle=True)) + result = asyncio.run( + xiaohongshu_setup( + str(account_file), + handle=True, + return_detail=True, + ) + ) diff --git a/examples/upload_video_to_xiaohongshu.py b/examples/upload_video_to_xiaohongshu.py index c6bc6d6..24da441 100644 --- a/examples/upload_video_to_xiaohongshu.py +++ b/examples/upload_video_to_xiaohongshu.py @@ -1,31 +1,90 @@ import asyncio +from datetime import datetime, timedelta from pathlib import Path from conf import BASE_DIR -from uploader.xiaohongshu_uploader.main import xiaohongshu_setup, XiaoHongShuVideo -from utils.files_times import generate_schedule_time_next_day, get_title_and_hashtags +from uploader.xiaohongshu_uploader.main import XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE +from uploader.xiaohongshu_uploader.main import XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED +from uploader.xiaohongshu_uploader.main import XiaoHongShuNote +from uploader.xiaohongshu_uploader.main import XiaoHongShuVideo + + +ACCOUNT_FILE = Path(BASE_DIR / "cookies" / "xiaohongshu_uploader" / "account.json") + + +def upload_video_to_xiaohongshu(): + video_file = Path(BASE_DIR) / "videos" / "demo.mp4" + thumbnail_path = video_file.with_suffix(".png") + app = XiaoHongShuVideo( + title="小红书视频示例", + file_path=str(video_file), + desc="你好", + tags=["小红书", "视频示例", "调试入口"], + publish_strategy=XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + publish_date=0, + account_file=str(ACCOUNT_FILE), + thumbnail_path=str(thumbnail_path) if thumbnail_path.exists() else None, + ) + asyncio.run(app.xiaohongshu_upload_video()) + + +def upload_video_to_xiaohongshu_scheduled(): + video_file = Path(BASE_DIR) / "videos" / "demo.mp4" + thumbnail_path = video_file.with_suffix(".png") + publish_time = (datetime.now() + timedelta(hours=3)).replace(second=0, microsecond=0) + app = XiaoHongShuVideo( + title="小红书视频定时发布示例", + file_path=str(video_file), + desc="这是一条定时发布的小红书视频示例", + tags=["小红书", "定时发布", "调试入口"], + publish_strategy=XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED, + publish_date=publish_time, + account_file=str(ACCOUNT_FILE), + thumbnail_path=str(thumbnail_path) if thumbnail_path.exists() else None, + ) + asyncio.run(app.xiaohongshu_upload_video()) + + +def upload_note_to_xiaohongshu(): + image_candidates = [ + Path(BASE_DIR) / "videos" / "demo.png", + Path(BASE_DIR) / "videos" / "demo1.png", + Path(BASE_DIR) / "videos" / "demo2.png", + ] + image_paths = [str(path) for path in image_candidates if path.exists()] + app = XiaoHongShuNote( + image_paths=image_paths, + note="小红书图文内容示例 #图文调试", + tags=["小红书图文", "自动上传", "调试入口"], + publish_strategy=XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + publish_date=0, + account_file=str(ACCOUNT_FILE), + title="小红书图文示例", + ) + asyncio.run(app.xiaohongshu_upload_note()) + +def upload_note_to_xiaohongshu_scheduled(): + image_candidates = [ + Path(BASE_DIR) / "videos" / "demo.png", + Path(BASE_DIR) / "videos" / "demo1.png", + Path(BASE_DIR) / "videos" / "demo2.png", + ] + image_paths = [str(path) for path in image_candidates if path.exists()] + publish_time = (datetime.now() + timedelta(hours=3)).replace(second=0, microsecond=0) + app = XiaoHongShuNote( + image_paths=image_paths, + note="小红书图文内容示例 #图文调试", + tags=["小红书图文", "自动上传", "调试入口"], + publish_strategy=XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED, + publish_date=publish_time, + account_file=str(ACCOUNT_FILE), + title="小红书图文示例", + ) + asyncio.run(app.xiaohongshu_upload_note()) if __name__ == '__main__': - filepath = Path(BASE_DIR) / "videos" - account_file = Path(BASE_DIR / "cookies" / "xiaohongshu_uploader" / "58a391ba-4082-11f0-a321-44e51723d63c.json") - # 获取视频目录 - folder_path = Path(filepath) - # 获取文件夹中的所有文件 - files = list(folder_path.glob("*.mp4")) - file_num = len(files) - publish_datetimes = generate_schedule_time_next_day(file_num, 1, daily_times=[16]) - cookie_setup = asyncio.run(xiaohongshu_setup(account_file, handle=False)) - for index, file in enumerate(files): - title, tags = get_title_and_hashtags(str(file)) - thumbnail_path = file.with_suffix('.png') - # 打印视频文件名、标题和 hashtag - print(f"视频文件名：{file}") - print(f"标题：{title}") - print(f"Hashtag：{tags}") - # 暂时没有时间修复封面上传，故先隐藏掉该功能 - # if thumbnail_path.exists(): - # app = XiaoHongShuVideo(title, file, tags, publish_datetimes[index], account_file, thumbnail_path=thumbnail_path) - # else: - app = XiaoHongShuVideo(title, file, tags, 0, account_file) - asyncio.run(app.main(), debug=False) + # upload_video_to_xiaohongshu() + # upload_video_to_xiaohongshu_scheduled() + # upload_note_to_xiaohongshu() + upload_note_to_xiaohongshu_scheduled() diff --git a/uploader/xiaohongshu_uploader/main.py b/uploader/xiaohongshu_uploader/main.py index e6f902b..5ad08f4 100644 --- a/uploader/xiaohongshu_uploader/main.py +++ b/uploader/xiaohongshu_uploader/main.py @@ -1,294 +1,345 @@ # -*- coding: utf-8 -*- -from datetime import datetime +from __future__ import annotations -from playwright.async_api import Playwright, async_playwright, Page -import os import asyncio +import inspect +import os +from datetime import datetime +from pathlib import Path -from conf import LOCAL_CHROME_PATH, LOCAL_CHROME_HEADLESS +from patchright.async_api import Page +from patchright.async_api import Playwright +from patchright.async_api import async_playwright + +from conf import DEBUG_MODE, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH +from uploader.base_video import BaseVideoUploader from utils.base_social_media import set_init_script +from utils.login_qrcode import build_login_qrcode_path +from utils.login_qrcode import decode_qrcode_from_path +from utils.login_qrcode import print_terminal_qrcode +from utils.login_qrcode import remove_qrcode_file +from utils.login_qrcode import save_data_url_image from utils.log import xiaohongshu_logger +XHS_LOGIN_URL = "https://creator.xiaohongshu.com/login" +XHS_PUBLISH_VIDEO_URL = "https://creator.xiaohongshu.com/publish/publish?from=homepage&target=video" +XHS_PUBLISH_NOTE_URL = "https://creator.xiaohongshu.com/publish/publish?from=homepage&target=image" +XHS_PUBLISH_SUCCESS_URL_PATTERN = "**/publish/success?**" +XHS_LOGIN_BOX_SELECTOR = "div[class*='login-box']" +XHS_LOGIN_SWITCH_SELECTOR = "img.css-wemwzq" +XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE = "immediate" +XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED = "scheduled" + + +def _msg(emoji: str, text: str) -> str: + return f"{emoji} {text}" + + +async def _emit_qrcode_callback(qrcode_callback, payload: dict): + if not qrcode_callback: + return + + callback_result = qrcode_callback(payload) + if inspect.isawaitable(callback_result): + await callback_result + + +def _build_login_result( + success: bool, + status: str, + message: str, + account_file: str, + qrcode: dict | None = None, + current_url: str = "", +) -> dict: + return { + "success": success, + "status": status, + "message": message, + "account_file": str(account_file), + "qrcode": qrcode, + "current_url": current_url, + } + + +async def _open_xhs_qrcode_panel(page: Page) -> None: + login_box = page.locator(XHS_LOGIN_BOX_SELECTOR).first + await login_box.wait_for(state="visible", timeout=30000) + + scan_text = login_box.locator("div:has-text('扫一扫')").first + if await scan_text.count(): + return + + switch_img = login_box.locator(XHS_LOGIN_SWITCH_SELECTOR).first + await switch_img.wait_for(state="visible", timeout=10000) + await switch_img.click() + await login_box.locator("div:has-text('扫一扫')").first.wait_for(state="visible", timeout=10000) + + +async def _find_xhs_qrcode_locator(page: Page): + await _open_xhs_qrcode_panel(page) + + qrcode_img = page.locator('.login-box-container').get_by_text("APP扫一扫登录").filter(visible=True).locator("xpath=..//following-sibling::div//img").nth(0) + + if await qrcode_img.count(): + return qrcode_img + + raise RuntimeError("未在扫一扫登录区域找到小红书二维码图片") + + +async def _extract_xhs_qrcode_src(page: Page) -> str: + qrcode_img = await _find_xhs_qrcode_locator(page) + await qrcode_img.wait_for(state="visible", timeout=30000) + qrcode_src = await qrcode_img.get_attribute("src") + if not qrcode_src: + raise RuntimeError("未获取到小红书登录二维码地址") + return qrcode_src + + +async def _save_xhs_qrcode( + page: Page, + account_file: str, + previous_qrcode_path: Path | None = None, + qrcode_callback=None, +) -> dict: + qrcode_src = await _extract_xhs_qrcode_src(page) + qrcode_path = build_login_qrcode_path(account_file, suffix="xhs_login_qrcode") + qrcode_img = await _find_xhs_qrcode_locator(page) + + if qrcode_src.startswith("data:image/"): + save_data_url_image(qrcode_src, qrcode_path) + else: + qrcode_path.parent.mkdir(parents=True, exist_ok=True) + await qrcode_img.screenshot(path=str(qrcode_path)) + + if previous_qrcode_path and previous_qrcode_path != qrcode_path: + if remove_qrcode_file(previous_qrcode_path): + xiaohongshu_logger.info(_msg("🧹", f"临时二维码文件已清理: {previous_qrcode_path}")) + + xiaohongshu_logger.info(_msg("🖼️", f"二维码已经准备好啦，已保存到: {qrcode_path}")) + qrcode_content = decode_qrcode_from_path(qrcode_path) + if qrcode_content: + print_terminal_qrcode(qrcode_content, qrcode_path, "小红书APP") + else: + xiaohongshu_logger.warning(_msg("😵", f"终端没法完整显示二维码，请打开 {qrcode_path} 扫码")) + + qrcode_info = { + "image_path": str(qrcode_path), + "image_data_url": qrcode_src, + } + await _emit_qrcode_callback(qrcode_callback, qrcode_info) + return qrcode_info + + +async def _is_xhs_login_completed(page: Page) -> bool: + if page.url.startswith(XHS_LOGIN_URL): + return False + + login_box = page.locator(XHS_LOGIN_BOX_SELECTOR).first + if not await login_box.count(): + return True + + try: + return not await login_box.is_visible() + except Exception: + return True + async def cookie_auth(account_file): + if not os.path.exists(account_file): + return False + async with async_playwright() as playwright: - browser = await playwright.chromium.launch(headless=LOCAL_CHROME_HEADLESS) - 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://creator.xiaohongshu.com/creator-micro/content/upload") - try: - await page.wait_for_url("https://creator.xiaohongshu.com/creator-micro/content/upload", timeout=5000) - except: - print("[+] 等待5秒 cookie 失效") - await context.close() - await browser.close() - return False - # 2024.06.17 抖音创作者中心改版 - if await page.get_by_text('手机号登录').count() or await page.get_by_text('扫码登录').count(): - print("[+] 等待5秒 cookie 失效") - return False + if LOCAL_CHROME_PATH: + browser = await playwright.chromium.launch(headless=True, executable_path=LOCAL_CHROME_PATH) else: - print("[+] cookie 有效") + browser = await playwright.chromium.launch(headless=True, channel="chrome") + try: + context = await browser.new_context(storage_state=account_file) + context = await set_init_script(context) + page = await context.new_page() + await page.goto(XHS_PUBLISH_VIDEO_URL) + await page.wait_for_timeout(3000) + + if page.url.startswith(XHS_LOGIN_URL): + xiaohongshu_logger.info(_msg("🥹", "cookie 已失效，得重新登录一下")) + return False + + login_box = page.locator(XHS_LOGIN_BOX_SELECTOR).first + if await login_box.count(): + try: + if await login_box.is_visible(): + xiaohongshu_logger.info(_msg("🥹", "页面仍然停留在登录二维码页，按 cookie 失效处理")) + return False + except Exception: + return False + + xiaohongshu_logger.success(_msg("🥳", "cookie 有效")) return True + except Exception as exc: + xiaohongshu_logger.warning(_msg("😵", f"cookie 校验时出错，按失效处理: {exc}")) + return False + finally: + await browser.close() -async def xiaohongshu_setup(account_file, handle=False): +async def xiaohongshu_setup( + account_file, + handle=False, + return_detail=False, + qrcode_callback=None, + headless: bool = LOCAL_CHROME_HEADLESS, +): if not os.path.exists(account_file) or not await cookie_auth(account_file): if not handle: - # Todo alert message - return False - xiaohongshu_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件') - await xiaohongshu_cookie_gen(account_file) - return True + result = _build_login_result(False, "cookie_invalid", "cookie文件不存在或已失效", account_file) + return result if return_detail else False + xiaohongshu_logger.info(_msg("🥹", "cookie 失效了，准备打开浏览器重新登录")) + result = await xiaohongshu_cookie_gen( + account_file, + qrcode_callback=qrcode_callback, + headless=headless, + ) + return result if return_detail else result["success"] + + result = _build_login_result(True, "cookie_valid", "cookie有效", account_file) + return result if return_detail else True -async def xiaohongshu_cookie_gen(account_file): +async def xiaohongshu_cookie_gen( + account_file, + qrcode_callback=None, + poll_interval: int = 3, + max_checks: int = 100, + headless: bool = LOCAL_CHROME_HEADLESS, +): + if headless: + xiaohongshu_logger.info(_msg("🖼️", "小红书登录将以无头模式运行，小人会输出终端二维码并保存本地二维码图片")) + + account_path = Path(account_file) + account_path.parent.mkdir(parents=True, exist_ok=True) + async with async_playwright() as playwright: - options = { - 'headless': LOCAL_CHROME_HEADLESS - } - # Make sure to run headed. - browser = await playwright.chromium.launch(**options) - # Setup context however you like. - context = await browser.new_context() # Pass any options + browser = await playwright.chromium.launch(headless=headless, channel="chrome") + context = await browser.new_context() context = await set_init_script(context) - # Pause the page, and start recording manually. - page = await context.new_page() - await page.goto("https://creator.xiaohongshu.com/") - await page.pause() - # 点击调试器的继续，保存cookie - await context.storage_state(path=account_file) + qrcode_path = None + qrcode_info = None + result = _build_login_result(False, "failed", "小红书登录失败", account_file) + try: + page = await context.new_page() + await page.goto(XHS_LOGIN_URL) + qrcode_info = await _save_xhs_qrcode(page, account_file, qrcode_callback=qrcode_callback) + qrcode_path = Path(qrcode_info["image_path"]) + xiaohongshu_logger.info(_msg("🧍", "请扫码，小人正在耐心等待登录完成")) + + for _ in range(max_checks): + if await _is_xhs_login_completed(page): + await asyncio.sleep(2) + await context.storage_state(path=account_file) + if await cookie_auth(account_file): + xiaohongshu_logger.success(_msg("🥳", "小红书扫码登录成功，小人开心收工")) + result = _build_login_result(True, "success", "小红书扫码登录成功", account_file, qrcode_info, page.url) + else: + result = _build_login_result( + False, + "cookie_invalid", + "小红书扫码流程结束，但 cookie 校验失败", + account_file, + qrcode_info, + page.url, + ) + return result + + await asyncio.sleep(poll_interval) + + result = _build_login_result( + False, + "timeout", + "等待小红书扫码登录超时", + account_file, + qrcode_info, + page.url, + ) + except Exception as exc: + result = _build_login_result(False, "failed", str(exc), account_file, current_url=page.url if "page" in locals() else "") + finally: + if remove_qrcode_file(qrcode_path): + xiaohongshu_logger.info(_msg("🧹", f"临时二维码文件已清理: {qrcode_path}")) + if not result["success"]: + xiaohongshu_logger.error(_msg("😢", f"登录失败: {result['message']}")) + await context.close() + await browser.close() + return result -class XiaoHongShuVideo(object): - def __init__(self, title, file_path, tags, publish_date: datetime, account_file, thumbnail_path=None): - self.title = title # 视频标题 - self.file_path = file_path - self.tags = tags +class XiaoHongShuBaseUploader(BaseVideoUploader): + def __init__( + self, + publish_date: datetime | int, + account_file, + publish_strategy: str = XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + ): self.publish_date = publish_date - self.account_file = account_file - self.date_format = '%Y年%m月%d日 %H:%M' + self.account_file = str(account_file) + self.publish_strategy = publish_strategy + self.debug = debug + self.date_format = "%Y年%m月%d日 %H:%M" self.local_executable_path = LOCAL_CHROME_PATH - self.headless = LOCAL_CHROME_HEADLESS - self.thumbnail_path = thumbnail_path + self.headless = headless - async def set_schedule_time_xiaohongshu(self, page, publish_date): + async def validate_base_args(self): + if not os.path.exists(self.account_file): + raise RuntimeError(f"cookie文件不存在，请先完成小红书登录: {self.account_file}") + if not await cookie_auth(self.account_file): + raise RuntimeError(f"cookie文件已失效，请先完成小红书登录: {self.account_file}") + + if self.publish_strategy not in { + XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED, + }: + raise ValueError(f"不支持的发布策略: {self.publish_strategy}") + + if self.publish_strategy == XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED: + self.publish_date = self.validate_publish_date(self.publish_date) + else: + self.publish_date = 0 + + async def set_schedule_time_xiaohongshu(self, page: Page, publish_date: datetime): print(" [-] 正在设置定时发布时间...") print(f"publish_date: {publish_date}") - - # 使用文本内容定位元素 - # element = await page.wait_for_selector( - # 'label:has-text("定时发布")', - # timeout=5000 # 5秒超时时间 - # ) - # await element.click() - - # # 选择包含特定文本内容的 label 元素 - label_element = page.locator("label:has-text('定时发布')") - # # 在选中的 label 元素下点击 checkbox - await label_element.click() + await page.locator('.custom-switch-card').filter(has_text="定时发布").locator('.d-switch').click() await asyncio.sleep(1) publish_date_hour = publish_date.strftime("%Y-%m-%d %H:%M") + publish_date_hour = publish_date.strftime("%Y-%m-%d %H:%M") print(f"publish_date_hour: {publish_date_hour}") - await asyncio.sleep(1) - await page.locator('.el-input__inner[placeholder="选择日期和时间"]').click() - await page.keyboard.press("Control+KeyA") - await page.keyboard.type(str(publish_date_hour)) - await page.keyboard.press("Enter") - - await asyncio.sleep(1) - - async def handle_upload_error(self, page): - xiaohongshu_logger.info('视频出错了，重新上传中') - await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path) - - async def upload(self, playwright: Playwright) -> None: - # 使用 Chromium 浏览器启动一个浏览器实例 - if self.local_executable_path: - browser = await playwright.chromium.launch(headless=self.headless, executable_path=self.local_executable_path) - else: - browser = await playwright.chromium.launch(headless=self.headless) - # 创建一个浏览器上下文，使用指定的 cookie 文件 - context = await browser.new_context( - viewport={"width": 1600, "height": 900}, - storage_state=f"{self.account_file}" - ) - context = await set_init_script(context) - - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://creator.xiaohongshu.com/publish/publish?from=homepage&target=video") - xiaohongshu_logger.info(f'[+]正在上传-------{self.title}.mp4') - # 等待页面跳转到指定的 URL，没进入，则自动等待到超时 - xiaohongshu_logger.info(f'[-] 正在打开主页...') - await page.wait_for_url("https://creator.xiaohongshu.com/publish/publish?from=homepage&target=video") - # 点击 "上传视频" 按钮 - await page.locator("div[class^='upload-content'] input[class='upload-input']").set_input_files(self.file_path) - - # 等待页面跳转到指定的 URL 2025.01.08修改在原有基础上兼容两种页面 - while True: - try: - # 等待upload-input元素出现 - upload_input = await page.wait_for_selector('input.upload-input', timeout=3000) - # 获取下一个兄弟元素 - preview_new = await upload_input.query_selector( - 'xpath=following-sibling::div[contains(@class, "preview-new")]') - if preview_new: - # 在preview-new元素中查找包含"上传成功"的stage元素 - stage_elements = await preview_new.query_selector_all('div.stage') - upload_success = False - for stage in stage_elements: - text_content = await page.evaluate('(element) => element.textContent', stage) - if '上传成功' in text_content: - upload_success = True - break - if upload_success: - xiaohongshu_logger.info("[+] 检测到上传成功标识!") - break # 成功检测到上传成功后跳出循环 - else: - print(" [-] 未找到上传成功标识，继续等待...") - else: - print(" [-] 未找到预览元素，继续等待...") - await asyncio.sleep(1) - except Exception as e: - print(f" [-] 检测过程出错: {str(e)}，重新尝试...") - await asyncio.sleep(0.5) # 等待0.5秒后重新尝试 - - # 填充标题和话题 - # 检查是否存在包含输入框的元素 - # 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素 - await asyncio.sleep(1) - xiaohongshu_logger.info(f' [-] 正在填充标题和话题...') - title_container = page.locator('div.plugin.title-container').locator('input.d-text') - if await title_container.count(): - await title_container.fill(self.title[:30]) - else: - titlecontainer = page.locator(".notranslate") - await titlecontainer.click() - await page.keyboard.press("Backspace") - await page.keyboard.press("Control+KeyA") - await page.keyboard.press("Delete") - await page.keyboard.type(self.title) - await page.keyboard.press("Enter") - css_selector = ".ql-editor" # 不能加上 .ql-blank 属性，这样只能获取第一次非空状态 - for index, tag in enumerate(self.tags, start=1): - await page.type(css_selector, "#" + tag) - await page.press(css_selector, "Space") - xiaohongshu_logger.info(f'总共添加{len(self.tags)}个话题') - - # while True: - # # 判断重新上传按钮是否存在，如果不存在，代表视频正在上传，则等待 - # try: - # # 新版：定位重新上传 - # number = await page.locator('[class^="long-card"] div:has-text("重新上传")').count() - # if number > 0: - # xiaohongshu_logger.success(" [-]视频上传完毕") - # break - # else: - # xiaohongshu_logger.info(" [-] 正在上传视频中...") - # await asyncio.sleep(2) - - # if await page.locator('div.progress-div > div:has-text("上传失败")').count(): - # xiaohongshu_logger.error(" [-] 发现上传出错了... 准备重试") - # await self.handle_upload_error(page) - # except: - # xiaohongshu_logger.info(" [-] 正在上传视频中...") - # await asyncio.sleep(2) - - # 上传视频封面 - # await self.set_thumbnail(page, self.thumbnail_path) - - # 更换可见元素 - # await self.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() - - if self.publish_date != 0: - await self.set_schedule_time_xiaohongshu(page, self.publish_date) - - # 判断视频是否发布成功 - while True: - try: - # 等待包含"定时发布"文本的button元素出现并点击 - if self.publish_date != 0: - await page.locator('button:has-text("定时发布")').click() - else: - await page.locator('button:has-text("发布")').click() - await page.wait_for_url( - "https://creator.xiaohongshu.com/publish/success?**", - timeout=3000 - ) # 如果自动跳转到作品页面，则代表发布成功 - xiaohongshu_logger.success(" [-]视频发布成功") - break - except: - xiaohongshu_logger.info(" [-] 视频正在发布中...") - await page.screenshot(full_page=True) - await asyncio.sleep(0.5) - - await context.storage_state(path=self.account_file) # 保存cookie - xiaohongshu_logger.success(' [-]cookie更新完毕！') - await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 - # 关闭浏览器上下文和浏览器实例 - await context.close() - await browser.close() - - async def set_thumbnail(self, page: Page, thumbnail_path: str): - if thumbnail_path: - await page.click('text="选择封面"') - await page.wait_for_selector("div.semi-modal-content:visible") - 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(thumbnail_path) - await page.wait_for_timeout(2000) # 等待2秒 - await page.locator("div[class^='extractFooter'] button:visible:has-text('完成')").click() - # finish_confirm_element = page.locator("div[class^='confirmBtn'] >> div:has-text('完成')") - # if await finish_confirm_element.count(): - # await finish_confirm_element.click() - # await page.locator("div[class^='footer'] button:has-text('完成')").click() + time_input = page.locator('.d-datepicker-input-filter input.d-text') + await time_input.fill(str(publish_date_hour)) async def set_location(self, page: Page, location: str = "青岛市"): print(f"开始设置位置: {location}") - - # 点击地点输入框 print("等待地点输入框加载...") loc_ele = await page.wait_for_selector('div.d-text.d-select-placeholder.d-text-ellipsis.d-text-nowrap') print(f"已定位到地点输入框: {loc_ele}") await loc_ele.click() print("点击地点输入框完成") - - # 输入位置名称 print(f"等待1秒后输入位置名称: {location}") await page.wait_for_timeout(1000) await page.keyboard.type(location) print(f"位置名称输入完成: {location}") - - # 等待下拉列表加载 print("等待下拉列表加载...") dropdown_selector = 'div.d-popover.d-popover-default.d-dropdown.--size-min-width-large' - await page.wait_for_timeout(3000) + await page.wait_for_timeout(2000) try: await page.wait_for_selector(dropdown_selector, timeout=3000) print("下拉列表已加载") - except: + except Exception: print("下拉列表未按预期显示，可能结构已变化") - - # 增加等待时间以确保内容加载完成 print("额外等待1秒确保内容渲染完成...") await page.wait_for_timeout(1000) - - # 尝试更灵活的XPath选择器 print("尝试使用更灵活的XPath选择器...") flexible_xpath = ( f'//div[contains(@class, "d-popover") and contains(@class, "d-dropdown")]' @@ -297,20 +348,16 @@ class XiaoHongShuVideo(object): f'//div[contains(@class, "name") and text()="{location}"]' ) await page.wait_for_timeout(3000) - - # 尝试定位元素 print(f"尝试定位包含'{location}'的选项...") try: - # 先尝试使用更灵活的选择器 location_option = await page.wait_for_selector( flexible_xpath, timeout=3000 ) - + if location_option: print(f"使用灵活选择器定位成功: {location_option}") else: - # 如果灵活选择器失败，再尝试原选择器 print("灵活选择器未找到元素，尝试原始选择器...") location_option = await page.wait_for_selector( f'//div[contains(@class, "d-popover") and contains(@class, "d-dropdown")]' @@ -319,26 +366,18 @@ class XiaoHongShuVideo(object): f'/div[1]//div[contains(@class, "name") and text()="{location}"]', timeout=2000 ) - - # 滚动到元素并点击 + print("滚动到目标选项...") await location_option.scroll_into_view_if_needed() print("元素已滚动到视图内") - - # 增加元素可见性检查 is_visible = await location_option.is_visible() print(f"目标选项是否可见: {is_visible}") - - # 点击元素 print("准备点击目标选项...") await location_option.click() print(f"成功选择位置: {location}") return True - except Exception as e: print(f"定位位置失败: {e}") - - # 打印更多调试信息 print("尝试获取下拉列表中的所有选项...") try: all_options = await page.query_selector_all( @@ -348,21 +387,319 @@ class XiaoHongShuVideo(object): '/div' ) print(f"找到 {len(all_options)} 个选项") - - # 打印前3个选项的文本内容 for i, option in enumerate(all_options[:3]): option_text = await option.inner_text() - print(f"选项 {i+1}: {option_text.strip()[:50]}...") - - except Exception as e: - print(f"获取选项列表失败: {e}") - - # 截图保存（取消注释使用） - # await page.screenshot(path=f"location_error_{location}.png") + print(f"选项 {i + 1}: {option_text.strip()[:50]}...") + except Exception as inner_e: + print(f"获取选项列表失败: {inner_e}") return False - async def main(self): + async def fill_title(self, page: Page) -> None: + title_container = page.locator('input[placeholder*="填写标题"]') + await title_container.fill(self.title[:20]) + + async def fill_desc(self, page: Page) -> None: + if not getattr(self, "desc", ""): + return + + desc = page.locator('p[data-placeholder*="输入正文描述"]') + await desc.click() + await page.keyboard.press("Backspace") + await page.keyboard.press("Control+KeyA") + await page.keyboard.press("Delete") + await page.keyboard.type(self.desc) + await page.keyboard.press("Enter") + + async def fill_tags(self, page: Page) -> None: + if not getattr(self, "tags", None): + return + + if not getattr(self, "desc", ""): + desc = page.locator('p[data-placeholder*="输入正文描述"]') + await desc.click() + + await page.keyboard.type("#" + self.tags[0], delay=30) + await page.locator('#creator-editor-topic-container').wait_for( + state="visible", + timeout=3000 + ) + first_item = page.locator('#creator-editor-topic-container .item').first + await first_item.wait_for(state="visible", timeout=2000) + await first_item.click() + + async def fill_meta(self, page: Page) -> None: + await self.fill_title(page) + await self.fill_desc(page) + await self.fill_tags(page) + + +class XiaoHongShuVideo(XiaoHongShuBaseUploader): + def __init__( + self, + title, + file_path, + tags, + publish_date: datetime | int, + account_file, + thumbnail_path=None, + desc: str | None = None, + publish_strategy: str = XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + super().__init__( + publish_date=publish_date, + account_file=account_file, + publish_strategy=publish_strategy, + debug=debug, + headless=headless, + ) + self.title = title + self.file_path = file_path + self.tags = tags or [] + self.thumbnail_path = thumbnail_path + self.desc = desc or "" + + async def validate_upload_args(self): + await self.validate_base_args() + if not self.title or not str(self.title).strip(): + raise ValueError("视频模式下，title 是必须的") + + self.file_path = str(self.validate_video_file(self.file_path)) + if self.thumbnail_path: + self.thumbnail_path = str(self.validate_image_file(self.thumbnail_path)) + + async def handle_upload_error(self, page: Page): + xiaohongshu_logger.info('视频出错了，重新上传中') + await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path) + + async def set_thumbnail(self, page: Page, thumbnail_path: str): + if not thumbnail_path: + return + + xiaohongshu_logger.info(_msg("🖼️", "小人准备设置封面")) + + cover_plugin_title = page.locator("div.cover-plugin-title").filter(has_text="设置封面") + cover_upload_dialog = cover_plugin_title.locator( + "xpath=ancestor::div[contains(@class, 'cover-plugin-preview')]" + ).locator("div.cover > div.default:visible") + await cover_upload_dialog.wait_for(state="visible", timeout=30000) + + await cover_upload_dialog.click(force=True) + + modal = page.locator("div.d-modal.cover-modal") + await modal.wait_for(state="visible", timeout=30000) + + file_input = modal.locator('input[type="file"][accept*="image"]').first + await file_input.wait_for(state="attached", timeout=10000) + await file_input.set_input_files(thumbnail_path) + await page.wait_for_timeout(2000) + + confirm_button = modal.locator("button.mojito-button").filter(has_text="确定").first + await confirm_button.wait_for(state="visible", timeout=10000) + await confirm_button.click() + + await modal.wait_for(state="hidden", timeout=30000) + xiaohongshu_logger.success(_msg("🥳", "封面已经设置完成")) + + async def upload_video_content(self, page: Page) -> None: + xiaohongshu_logger.info(f'[+]正在上传-------{self.title}.mp4') + xiaohongshu_logger.info(f'[-] 正在打开主页...') + await page.goto(XHS_PUBLISH_VIDEO_URL) + await page.wait_for_url(XHS_PUBLISH_VIDEO_URL) + await page.locator("div[class^='upload-content'] input[class='upload-input']").set_input_files(self.file_path) + + while True: + try: + upload_input = await page.wait_for_selector('input.upload-input', timeout=3000) + preview_new = await upload_input.query_selector( + 'xpath=following-sibling::div[contains(@class, "preview-new")]') + if preview_new: + stage_elements = await preview_new.query_selector_all('div.stage') + upload_success = False + for stage in stage_elements: + text_content = await page.evaluate('(element) => element.textContent', stage) + if '上传成功' in text_content or '分辨率' in text_content : + upload_success = True + break + if upload_success: + xiaohongshu_logger.info("[+] 检测到上传成功标识!") + break + print(" [-] 未找到上传成功标识，继续等待...") + else: + print(" [-] 未找到预览元素，继续等待...") + await asyncio.sleep(3) + except Exception as e: + print(f" [-] 检测过程出错: {str(e)}，重新尝试...") + await asyncio.sleep(3) + + xiaohongshu_logger.info(' [-] 正在填写标题、描述和话题...') + await self.fill_meta(page) + + await self.set_thumbnail(page, self.thumbnail_path) + + # await self.set_location(page, "青岛市") + + if self.publish_strategy == XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: + await self.set_schedule_time_xiaohongshu(page, self.publish_date) + + while True: + try: + if self.publish_strategy == XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED: + await page.locator('button:has-text("定时发布")').click() + else: + await page.locator('button:has-text("发布")').click() + await page.wait_for_url( + "https://creator.xiaohongshu.com/publish/success?**", + timeout=3000 + ) + xiaohongshu_logger.success(" [-]视频发布成功") + break + except Exception: + xiaohongshu_logger.info(" [-] 视频正在发布中...") + await page.screenshot(full_page=True) + await asyncio.sleep(0.5) + + async def upload(self, playwright: Playwright) -> None: + xiaohongshu_logger.info(_msg("🧍", "小人先检查 cookie、视频文件、封面和发布时间")) + await self.validate_upload_args() + xiaohongshu_logger.info(_msg("🥳", "上传前检查通过")) + + browser = await playwright.chromium.launch(headless=self.headless) + context = await browser.new_context( + permissions=["geolocation"], + storage_state=self.account_file, + ) + context = await set_init_script(context) + + try: + page = await context.new_page() + await self.upload_video_content(page) + await context.storage_state(path=self.account_file) + xiaohongshu_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def xiaohongshu_upload_video(self): async with async_playwright() as playwright: await self.upload(playwright) + async def main(self): + await self.xiaohongshu_upload_video() + +class XiaoHongShuNote(XiaoHongShuBaseUploader): + def __init__( + self, + image_paths, + note, + tags, + publish_date: datetime | int, + account_file, + title: str | None = None, + desc: str | None = None, + publish_strategy: str = XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + super().__init__( + publish_date=publish_date, + account_file=account_file, + publish_strategy=publish_strategy, + debug=debug, + headless=headless, + ) + self.image_paths = image_paths + self.note = note or "" + self.tags = tags or [] + self.desc = desc if desc is not None else self.note + self.title = title or ((self.desc or self.note)[:20] if (self.desc or self.note) else "") + + async def validate_upload_args(self): + await self.validate_base_args() + if not self.image_paths: + raise ValueError("图文模式下，图片是必须的") + if not self.title or not str(self.title).strip(): + raise ValueError("图文模式下，title 是必须的") + + if isinstance(self.image_paths, (str, Path)): + self.image_paths = [self.image_paths] + + normalized_image_paths = [] + for image_path in self.image_paths: + normalized_image_paths.append(str(self.validate_image_file(image_path))) + self.image_paths = normalized_image_paths + + async def upload_note_content(self, page: Page) -> None: + xiaohongshu_logger.info(f'[+]正在上传图文，共 {len(self.image_paths)} 张图片') + xiaohongshu_logger.info('[-] 正在打开图文发布页...') + await page.goto(XHS_PUBLISH_NOTE_URL) + await page.wait_for_url(XHS_PUBLISH_NOTE_URL) + + upload_input = page.locator('input[type="file"][accept*="image"]').first + if not await upload_input.count(): + upload_input = page.locator("div[class^='upload-content'] input[class='upload-input']").first + + await upload_input.wait_for(state="attached", timeout=30000) + await upload_input.set_input_files(self.image_paths) + + while True: + try: + title_container = page.locator('input[placeholder*="填写标题"]').first + await title_container.wait_for(state="visible", timeout=3000) + xiaohongshu_logger.info("[+] 检测到图文素材上传完成，已经可以填写内容") + break + except Exception: + xiaohongshu_logger.info(" [-] 图文素材仍在上传中，继续等待...") + await asyncio.sleep(1) + + xiaohongshu_logger.info(' [-] 正在填写标题、描述和话题...') + await self.fill_meta(page) + + if self.publish_strategy == XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: + await self.set_schedule_time_xiaohongshu(page, self.publish_date) + + while True: + try: + if self.publish_strategy == XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED: + await page.locator('button:has-text("定时发布")').click() + else: + await page.locator('button:has-text("发布")').click() + await page.wait_for_url( + XHS_PUBLISH_SUCCESS_URL_PATTERN, + timeout=3000 + ) + xiaohongshu_logger.success(" [-]图文发布成功") + break + except Exception: + xiaohongshu_logger.info(" [-] 图文正在发布中...") + await page.screenshot(full_page=True) + await asyncio.sleep(0.5) + + async def upload(self, playwright: Playwright) -> None: + xiaohongshu_logger.info(_msg("🧍", "小人先检查 cookie、图片和发布时间")) + await self.validate_upload_args() + xiaohongshu_logger.info(_msg("🥳", "图文上传前检查通过")) + + browser = await playwright.chromium.launch(headless=self.headless, channel="chrome") + context = await browser.new_context( + storage_state=self.account_file, + ) + context = await set_init_script(context) + + try: + page = await context.new_page() + await self.upload_note_content(page) + await context.storage_state(path=self.account_file) + xiaohongshu_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def xiaohongshu_upload_note(self): + async with async_playwright() as playwright: + await self.upload(playwright) + + async def main(self): + await self.xiaohongshu_upload_note()