小红书重构
1. 图文上传 2. 视频上传(封面,定时) 3. cli 4. skill 5. 无头模式
This commit is contained in:
home-dev-pookz
@@ -0,0 +1,184 @@
|
||||
import asyncio
|
||||
import tempfile
|
||||
import unittest
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import sau_cli
|
||||
|
||||
|
||||
class BrowserCliParserTests(unittest.TestCase):
|
||||
def test_build_parser_accepts_xiaohongshu_login(self):
|
||||
parser = sau_cli.build_parser()
|
||||
args = parser.parse_args(["xiaohongshu", "login", "--account", "creator"])
|
||||
self.assertEqual(args.platform, "xiaohongshu")
|
||||
self.assertEqual(args.action, "login")
|
||||
|
||||
def test_douyin_upload_video_accepts_desc(self):
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
video_path = Path(tmp_dir) / "demo.mp4"
|
||||
video_path.write_bytes(b"video")
|
||||
|
||||
parser = sau_cli.build_parser()
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"douyin",
|
||||
"upload-video",
|
||||
"--account",
|
||||
"creator",
|
||||
"--file",
|
||||
str(video_path),
|
||||
"--title",
|
||||
"标题",
|
||||
"--desc",
|
||||
"视频简介",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.desc, "视频简介")
|
||||
|
||||
def test_kuaishou_upload_note_accepts_title_and_note(self):
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
image_path = Path(tmp_dir) / "1.png"
|
||||
image_path.write_bytes(b"image")
|
||||
|
||||
parser = sau_cli.build_parser()
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"kuaishou",
|
||||
"upload-note",
|
||||
"--account",
|
||||
"creator",
|
||||
"--images",
|
||||
str(image_path),
|
||||
"--title",
|
||||
"图文标题",
|
||||
"--note",
|
||||
"图文正文",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.title, "图文标题")
|
||||
self.assertEqual(args.note, "图文正文")
|
||||
|
||||
def test_xiaohongshu_upload_video_defaults_to_headless(self):
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
video_path = Path(tmp_dir) / "demo.mp4"
|
||||
video_path.write_bytes(b"video")
|
||||
|
||||
parser = sau_cli.build_parser()
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"xiaohongshu",
|
||||
"upload-video",
|
||||
"--account",
|
||||
"creator",
|
||||
"--file",
|
||||
str(video_path),
|
||||
"--title",
|
||||
"视频标题",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertTrue(args.headless)
|
||||
|
||||
def test_xiaohongshu_upload_note_accepts_headed(self):
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
image_path = Path(tmp_dir) / "1.png"
|
||||
image_path.write_bytes(b"image")
|
||||
|
||||
parser = sau_cli.build_parser()
|
||||
args = parser.parse_args(
|
||||
[
|
||||
"xiaohongshu",
|
||||
"upload-note",
|
||||
"--account",
|
||||
"creator",
|
||||
"--images",
|
||||
str(image_path),
|
||||
"--title",
|
||||
"图文标题",
|
||||
"--note",
|
||||
"图文正文",
|
||||
"--headed",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertFalse(args.headless)
|
||||
|
||||
|
||||
class BrowserCliDispatchTests(unittest.TestCase):
|
||||
def test_dispatch_xiaohongshu_check_prints_valid(self):
|
||||
args = Namespace(platform="xiaohongshu", action="check", account="creator")
|
||||
with patch("sau_cli.check_xiaohongshu_account", new=AsyncMock(return_value=True)):
|
||||
code = asyncio.run(sau_cli.dispatch(args))
|
||||
self.assertEqual(code, 0)
|
||||
|
||||
def test_dispatch_douyin_upload_note_uses_new_request_fields(self):
|
||||
args = Namespace(
|
||||
platform="douyin",
|
||||
action="upload-note",
|
||||
account="creator",
|
||||
images=[Path("1.png")],
|
||||
title="图文标题",
|
||||
note="图文正文",
|
||||
tags="测试,图文",
|
||||
schedule=0,
|
||||
debug=False,
|
||||
headless=True,
|
||||
)
|
||||
with patch("sau_cli.upload_note", new=AsyncMock()) as mock_upload:
|
||||
asyncio.run(sau_cli.dispatch(args))
|
||||
|
||||
request = mock_upload.await_args.args[0]
|
||||
self.assertEqual(request.title, "图文标题")
|
||||
self.assertEqual(request.note, "图文正文")
|
||||
|
||||
def test_dispatch_xiaohongshu_upload_video_uses_headed_request(self):
|
||||
args = Namespace(
|
||||
platform="xiaohongshu",
|
||||
action="upload-video",
|
||||
account="creator",
|
||||
file=Path("demo.mp4"),
|
||||
title="视频标题",
|
||||
desc="视频简介",
|
||||
tags="测试,视频",
|
||||
schedule=0,
|
||||
thumbnail=None,
|
||||
debug=False,
|
||||
headless=False,
|
||||
)
|
||||
with patch("sau_cli.upload_xiaohongshu_video", new=AsyncMock()) as mock_upload:
|
||||
asyncio.run(sau_cli.dispatch(args))
|
||||
|
||||
request = mock_upload.await_args.args[0]
|
||||
self.assertEqual(request.title, "视频标题")
|
||||
self.assertEqual(request.description, "视频简介")
|
||||
self.assertFalse(request.headless)
|
||||
|
||||
def test_dispatch_xiaohongshu_upload_note_uses_headless_request(self):
|
||||
args = Namespace(
|
||||
platform="xiaohongshu",
|
||||
action="upload-note",
|
||||
account="creator",
|
||||
images=[Path("1.png"), Path("2.png")],
|
||||
title="图文标题",
|
||||
note="图文正文",
|
||||
tags="测试,图文",
|
||||
schedule=0,
|
||||
debug=False,
|
||||
headless=True,
|
||||
)
|
||||
with patch("sau_cli.upload_xiaohongshu_note", new=AsyncMock()) as mock_upload:
|
||||
asyncio.run(sau_cli.dispatch(args))
|
||||
|
||||
request = mock_upload.await_args.args[0]
|
||||
self.assertEqual(request.title, "图文标题")
|
||||
self.assertEqual(request.note, "图文正文")
|
||||
self.assertTrue(request.headless)
|
||||
self.assertEqual(len(request.image_files), 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,260 @@
|
||||
import asyncio
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import uploader.xiaohongshu_uploader.main as xhs_main
|
||||
|
||||
|
||||
class FakeLocator:
|
||||
def __init__(self, name, count=0, src=None, children=None):
|
||||
self.name = name
|
||||
self._count = count
|
||||
self._src = src
|
||||
self._children = children or {}
|
||||
|
||||
@property
|
||||
def first(self):
|
||||
return self
|
||||
|
||||
def locator(self, selector):
|
||||
return self._children.get(selector, FakeLocator(selector))
|
||||
|
||||
def get_by_text(self, text, exact=False):
|
||||
return self._children.get(f"text:{text}", FakeLocator(text))
|
||||
|
||||
def filter(self, **kwargs):
|
||||
return self
|
||||
|
||||
def nth(self, index):
|
||||
return self
|
||||
|
||||
async def count(self):
|
||||
return self._count
|
||||
|
||||
async def wait_for(self, **kwargs):
|
||||
return None
|
||||
|
||||
async def get_attribute(self, name):
|
||||
if name == "src":
|
||||
return self._src
|
||||
return None
|
||||
|
||||
async def fill(self, value):
|
||||
return None
|
||||
|
||||
async def click(self):
|
||||
return None
|
||||
|
||||
|
||||
class RecordingKeyboard:
|
||||
def __init__(self):
|
||||
self.actions = []
|
||||
|
||||
async def press(self, key):
|
||||
self.actions.append(("press", key))
|
||||
|
||||
async def type(self, text, delay=None):
|
||||
self.actions.append(("type", text, delay))
|
||||
|
||||
|
||||
class RecordingLocator(FakeLocator):
|
||||
def __init__(self, name):
|
||||
super().__init__(name, count=1)
|
||||
self.actions = []
|
||||
|
||||
async def fill(self, value):
|
||||
self.actions.append(("fill", value))
|
||||
|
||||
async def click(self):
|
||||
self.actions.append(("click",))
|
||||
|
||||
async def wait_for(self, **kwargs):
|
||||
self.actions.append(("wait_for", kwargs))
|
||||
|
||||
|
||||
class RecordingPage:
|
||||
def __init__(self):
|
||||
self.keyboard = RecordingKeyboard()
|
||||
self.locators = {
|
||||
'input[placeholder*="填写标题"]': RecordingLocator("title"),
|
||||
'p[data-placeholder*="输入正文描述"]': RecordingLocator("desc"),
|
||||
'#creator-editor-topic-container': RecordingLocator("topic-container"),
|
||||
'#creator-editor-topic-container .item': RecordingLocator("topic-item"),
|
||||
}
|
||||
|
||||
def locator(self, selector):
|
||||
return self.locators[selector]
|
||||
|
||||
|
||||
class XiaohongshuUploaderTests(unittest.TestCase):
|
||||
def test_find_xhs_qrcode_locator_prefers_scan_sibling_inside_login_box(self):
|
||||
qrcode_locator = FakeLocator("qrcode", count=1, src="data:image/png;base64,abc")
|
||||
scan_text_locator = FakeLocator(
|
||||
"scan-text",
|
||||
count=1,
|
||||
children={
|
||||
"xpath=..//following-sibling::div//img": qrcode_locator,
|
||||
},
|
||||
)
|
||||
login_box_locator = FakeLocator(
|
||||
"login-box",
|
||||
count=1,
|
||||
children={
|
||||
"div:has-text('扫一扫')": scan_text_locator,
|
||||
"text:APP扫一扫登录": scan_text_locator,
|
||||
},
|
||||
)
|
||||
page = FakeLocator(
|
||||
"page",
|
||||
children={
|
||||
"div[class*='login-box']": login_box_locator,
|
||||
".login-box-container": login_box_locator,
|
||||
},
|
||||
)
|
||||
|
||||
locator = asyncio.run(xhs_main._find_xhs_qrcode_locator(page))
|
||||
self.assertIs(locator, qrcode_locator)
|
||||
|
||||
def test_setup_returns_detail_when_cookie_invalid_without_handle(self):
|
||||
with patch("uploader.xiaohongshu_uploader.main.os.path.exists", return_value=False):
|
||||
result = asyncio.run(
|
||||
xhs_main.xiaohongshu_setup(
|
||||
"missing.json",
|
||||
handle=False,
|
||||
return_detail=True,
|
||||
)
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["status"], "cookie_invalid")
|
||||
|
||||
def test_setup_uses_login_flow_when_handle_is_true(self):
|
||||
login_result = {
|
||||
"success": True,
|
||||
"status": "success",
|
||||
"message": "ok",
|
||||
"account_file": "account.json",
|
||||
"qrcode": {"image_path": "qrcode.png"},
|
||||
"current_url": "https://creator.xiaohongshu.com/",
|
||||
}
|
||||
with patch("uploader.xiaohongshu_uploader.main.os.path.exists", return_value=False):
|
||||
with patch(
|
||||
"uploader.xiaohongshu_uploader.main.xiaohongshu_cookie_gen",
|
||||
new=AsyncMock(return_value=login_result),
|
||||
) as mock_login:
|
||||
result = asyncio.run(
|
||||
xhs_main.xiaohongshu_setup(
|
||||
"account.json",
|
||||
handle=True,
|
||||
return_detail=True,
|
||||
)
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
mock_login.assert_awaited_once()
|
||||
|
||||
def test_video_validate_upload_args_normalizes_video_and_thumbnail(self):
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
video_path = Path(tmp_dir) / "demo.mp4"
|
||||
thumbnail_path = Path(tmp_dir) / "demo.png"
|
||||
cookie_path = Path(tmp_dir) / "account.json"
|
||||
video_path.write_bytes(b"video")
|
||||
thumbnail_path.write_bytes(b"image")
|
||||
cookie_path.write_text("{}")
|
||||
|
||||
app = xhs_main.XiaoHongShuVideo(
|
||||
title="demo",
|
||||
file_path=str(video_path),
|
||||
tags=["xhs"],
|
||||
publish_date=0,
|
||||
account_file=str(cookie_path),
|
||||
thumbnail_path=str(thumbnail_path),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"uploader.xiaohongshu_uploader.main.cookie_auth",
|
||||
new=AsyncMock(return_value=True),
|
||||
):
|
||||
asyncio.run(app.validate_upload_args())
|
||||
|
||||
self.assertTrue(app.file_path.endswith("demo.mp4"))
|
||||
self.assertTrue(app.thumbnail_path.endswith("demo.png"))
|
||||
|
||||
def test_note_uploader_exists_and_validates_required_fields(self):
|
||||
note_cls = getattr(xhs_main, "XiaoHongShuNote")
|
||||
app = note_cls(
|
||||
image_paths=[],
|
||||
note="",
|
||||
tags=[],
|
||||
publish_date=0,
|
||||
account_file="account.json",
|
||||
)
|
||||
|
||||
with patch.object(app, "validate_base_args", new=AsyncMock(return_value=None)):
|
||||
with self.assertRaises(ValueError):
|
||||
asyncio.run(app.validate_upload_args())
|
||||
|
||||
def test_video_fill_meta_uses_desc_then_first_tag(self):
|
||||
app = xhs_main.XiaoHongShuVideo(
|
||||
title="标题内容",
|
||||
file_path="demo.mp4",
|
||||
tags=["话题1"],
|
||||
publish_date=0,
|
||||
account_file="account.json",
|
||||
desc="描述内容",
|
||||
)
|
||||
page = RecordingPage()
|
||||
|
||||
asyncio.run(app.fill_meta(page))
|
||||
|
||||
self.assertEqual(
|
||||
page.locators['input[placeholder*="填写标题"]'].actions,
|
||||
[("fill", "标题内容")],
|
||||
)
|
||||
self.assertEqual(
|
||||
page.locators['p[data-placeholder*="输入正文描述"]'].actions,
|
||||
[("click",)],
|
||||
)
|
||||
self.assertIn(("type", "描述内容", None), page.keyboard.actions)
|
||||
self.assertIn(("type", "#话题1", 30), page.keyboard.actions)
|
||||
self.assertEqual(
|
||||
page.locators['#creator-editor-topic-container .item'].actions,
|
||||
[("wait_for", {"state": "visible", "timeout": 2000}), ("click",)],
|
||||
)
|
||||
|
||||
def test_video_fill_meta_can_fill_first_tag_without_desc(self):
|
||||
app = xhs_main.XiaoHongShuVideo(
|
||||
title="标题内容",
|
||||
file_path="demo.mp4",
|
||||
tags=["话题1"],
|
||||
publish_date=0,
|
||||
account_file="account.json",
|
||||
)
|
||||
page = RecordingPage()
|
||||
|
||||
asyncio.run(app.fill_meta(page))
|
||||
|
||||
self.assertEqual(
|
||||
page.locators['p[data-placeholder*="输入正文描述"]'].actions,
|
||||
[("click",)],
|
||||
)
|
||||
self.assertNotIn(("type", "", None), page.keyboard.actions)
|
||||
self.assertIn(("type", "#话题1", 30), page.keyboard.actions)
|
||||
|
||||
def test_note_title_defaults_do_not_override_explicit_title(self):
|
||||
app = xhs_main.XiaoHongShuNote(
|
||||
image_paths=["a.png"],
|
||||
note="正文",
|
||||
tags=[],
|
||||
publish_date=0,
|
||||
account_file="account.json",
|
||||
title="显式标题",
|
||||
desc="图文正文",
|
||||
)
|
||||
|
||||
self.assertEqual(app.title, "显式标题")
|
||||
self.assertEqual(app.desc, "图文正文")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user