354 lines
12 KiB
Python
354 lines
12 KiB
Python
|
"""
|
||
|
|
小红书发布示例
|
||
|
|
演示小红书平台的各种发布场景。
|
||
|
|
"""
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
# 添加项目根目录到Python路径
|
||
|
|
project_root = Path(__file__).parent.parent
|
||
|
|
sys.path.insert(0, str(project_root))
|
||
|
|
|
||
|
|
from core.publisher import Publisher
|
||
|
|
from core.models import ImageNote, VideoContent, PlatformType, AccountInfo
|
||
|
|
from utils.logger import setup_logger, get_logger
|
||
|
|
|
||
|
|
# 设置日志
|
||
|
|
setup_logger(level="INFO")
|
||
|
|
logger = get_logger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
async def xiaohongshu_image_note_example():
|
||
|
|
"""小红书图文笔记发布示例"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("小红书图文笔记发布示例")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
async with Publisher(headless=False) as publisher:
|
||
|
|
# 设置账号
|
||
|
|
account_name = "your_xiaohongshu_account" # 替换为实际账号名
|
||
|
|
print(f"1. 设置小红书账号: {account_name}")
|
||
|
|
|
||
|
|
success = await publisher.setup_platform("xiaohongshu", account_name)
|
||
|
|
if not success:
|
||
|
|
print("❌ 账号设置失败，请检查网络和账号信息")
|
||
|
|
return
|
||
|
|
|
||
|
|
print("✅ 账号设置成功")
|
||
|
|
|
||
|
|
# 创建多种类型的图文笔记
|
||
|
|
notes = [
|
||
|
|
# 美食分享笔记
|
||
|
|
ImageNote(
|
||
|
|
title="今日美食分享 🍜",
|
||
|
|
description="今天尝试了一家超级好吃的面馆，汤底浓郁，面条劲道，配上特制的辣椒酱，简直是绝配！强烈推荐给大家。#美食分享 #探店",
|
||
|
|
images=[
|
||
|
|
"media/images/food1.jpg",
|
||
|
|
"media/images/food2.jpg",
|
||
|
|
"media/images/food3.jpg"
|
||
|
|
],
|
||
|
|
tags=["美食", "探店", "面食", "推荐"]
|
||
|
|
),
|
||
|
|
|
||
|
|
# 旅行风景笔记
|
||
|
|
ImageNote(
|
||
|
|
title="周末游山玩水 🏔️",
|
||
|
|
description="这个周末去了附近的山里徒步，空气特别清新，风景也很美。爬山虽然有点累，但是看到山顶的风景一切都值得了。#旅行 #周末 #户外",
|
||
|
|
images=[
|
||
|
|
"media/images/travel1.jpg",
|
||
|
|
"media/images/travel2.jpg",
|
||
|
|
"media/images/travel3.jpg",
|
||
|
|
"media/images/travel4.jpg",
|
||
|
|
"media/images/travel5.jpg"
|
||
|
|
],
|
||
|
|
tags=["旅行", "户外", "徒步", "风景"]
|
||
|
|
),
|
||
|
|
|
||
|
|
# 生活记录笔记
|
||
|
|
ImageNote(
|
||
|
|
title="生活小确幸 ☀️",
|
||
|
|
description="今天收到了朋友送的花，心情特别好。生活中的小确幸就是这样简单，一朵花，一杯茶，一本好书，就是最美好的时光。#生活 #小确幸 #感恩",
|
||
|
|
images=[
|
||
|
|
"media/images/life1.jpg"
|
||
|
|
],
|
||
|
|
tags=["生活", "日常", "心情", "感恩"]
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
# 发布笔记
|
||
|
|
for i, note in enumerate(notes, 1):
|
||
|
|
print(f"\n{i}. 发布笔记: {note.title}")
|
||
|
|
try:
|
||
|
|
result = await publisher.publish("xiaohongshu", note, account_name)
|
||
|
|
|
||
|
|
if result.success:
|
||
|
|
print(f"✅ 发布成功！")
|
||
|
|
print(f" 任务ID: {result.task_id}")
|
||
|
|
print(f" 耗时: {result.duration:.2f}秒")
|
||
|
|
else:
|
||
|
|
print(f"❌ 发布失败: {result.message}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"❌ 发布异常: {e}")
|
||
|
|
|
||
|
|
# 发布间隔
|
||
|
|
if i < len(notes):
|
||
|
|
print("⏳ 等待5秒后发布下一篇...")
|
||
|
|
await asyncio.sleep(5)
|
||
|
|
|
||
|
|
|
||
|
|
async def xiaohongshu_video_note_example():
|
||
|
|
"""小红书视频笔记发布示例"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("小红书视频笔记发布示例")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
async with Publisher(headless=False) as publisher:
|
||
|
|
# 设置账号
|
||
|
|
account_name = "your_xiaohongshu_account" # 替换为实际账号名
|
||
|
|
print(f"1. 设置小红书账号: {account_name}")
|
||
|
|
|
||
|
|
success = await publisher.setup_platform("xiaohongshu", account_name)
|
||
|
|
if not success:
|
||
|
|
print("❌ 账号设置失败")
|
||
|
|
return
|
||
|
|
|
||
|
|
print("✅ 账号设置成功")
|
||
|
|
|
||
|
|
# 创建视频笔记
|
||
|
|
video_notes = [
|
||
|
|
# 生活vlog
|
||
|
|
VideoContent(
|
||
|
|
title="我的日常vlog 📸",
|
||
|
|
description="记录一下今天的生活，从早上起床到晚上睡觉的点点滴滴。希望大家喜欢这种分享形式。#vlog #日常 #生活记录",
|
||
|
|
video_path="media/videos/daily_vlog.mp4", # 替换为实际视频路径
|
||
|
|
tags=["vlog", "日常", "生活"]
|
||
|
|
),
|
||
|
|
|
||
|
|
# 美食制作
|
||
|
|
VideoContent(
|
||
|
|
title="美食制作教程 🍳",
|
||
|
|
description="今天教大家做一道超级简单又好吃的家常菜，新手也能轻松学会。详细的步骤都在视频里哦。#美食 #教程 #烹饪",
|
||
|
|
video_path="media/videos/cooking_tutorial.mp4", # 替换为实际视频路径
|
||
|
|
tags=["美食", "教程", "烹饪", "家常菜"]
|
||
|
|
)
|
||
|
|
]
|
||
|
|
|
||
|
|
# 发布视频笔记
|
||
|
|
for i, video_note in enumerate(video_notes, 1):
|
||
|
|
print(f"\n{i}. 发布视频笔记: {video_note.title}")
|
||
|
|
try:
|
||
|
|
result = await publisher.publish("xiaohongshu", video_note, account_name)
|
||
|
|
|
||
|
|
if result.success:
|
||
|
|
print(f"✅ 发布成功！")
|
||
|
|
print(f" 任务ID: {result.task_id}")
|
||
|
|
print(f" 耗时: {result.duration:.2f}秒")
|
||
|
|
else:
|
||
|
|
print(f"❌ 发布失败: {result.message}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"❌ 发布异常: {e}")
|
||
|
|
|
||
|
|
# 发布间隔
|
||
|
|
if i < len(video_notes):
|
||
|
|
print("⏳ 等待10秒后发布下一个视频...")
|
||
|
|
await asyncio.sleep(10)
|
||
|
|
|
||
|
|
|
||
|
|
async def xiaohongshu_batch_publish_example():
|
||
|
|
"""小红书批量发布示例"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("小红书批量发布示例")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# 批量准备内容
|
||
|
|
contents = [
|
||
|
|
ImageNote(
|
||
|
|
title=f"批量测试笔记 {i+1}",
|
||
|
|
description=f"这是第{i+1}篇批量发布的测试笔记，内容为{i+1}。",
|
||
|
|
images=[f"media/images/batch_test_{i+1}.jpg"], # 替换为实际图片路径
|
||
|
|
tags=[f"批量测试{i+1}", "自动化"]
|
||
|
|
)
|
||
|
|
for i in range(3)
|
||
|
|
]
|
||
|
|
|
||
|
|
async with Publisher(headless=False) as publisher:
|
||
|
|
# 设置账号
|
||
|
|
account_name = "your_xiaohongshu_account" # 替换为实际账号名
|
||
|
|
success = await publisher.setup_platform("xiaohongshu", account_name)
|
||
|
|
|
||
|
|
if not success:
|
||
|
|
print("❌ 账号设置失败")
|
||
|
|
return
|
||
|
|
|
||
|
|
print("✅ 账号设置成功")
|
||
|
|
print(f"开始批量发布 {len(contents)} 篇笔记...")
|
||
|
|
|
||
|
|
# 创建发布任务
|
||
|
|
from core.models import PublishTask
|
||
|
|
tasks = []
|
||
|
|
for i, content in enumerate(contents):
|
||
|
|
task = PublishTask(
|
||
|
|
id=f"xhs_batch_{i+1}",
|
||
|
|
platform=PlatformType.XIAOHONGSHU,
|
||
|
|
account=AccountInfo(
|
||
|
|
platform=PlatformType.XIAOHONGSHU,
|
||
|
|
username=account_name,
|
||
|
|
cookie_file=f"{account_name}.json"
|
||
|
|
),
|
||
|
|
content=content
|
||
|
|
)
|
||
|
|
tasks.append(task)
|
||
|
|
|
||
|
|
# 执行批量发布
|
||
|
|
try:
|
||
|
|
results = await publisher.batch_publish(tasks)
|
||
|
|
|
||
|
|
# 统计结果
|
||
|
|
success_count = sum(1 for r in results if r.success)
|
||
|
|
print(f"\n📊 批量发布完成:")
|
||
|
|
print(f" 成功: {success_count}/{len(results)}")
|
||
|
|
print(f" 失败: {len(results) - success_count}/{len(results)}")
|
||
|
|
|
||
|
|
# 详细结果
|
||
|
|
for result in results:
|
||
|
|
status = "✅" if result.success else "❌"
|
||
|
|
print(f" {status} {result.task_id}: {result.message}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"❌ 批量发布异常: {e}")
|
||
|
|
|
||
|
|
|
||
|
|
async def xiaohongshu_content_validation_example():
|
||
|
|
"""小红书内容验证示例"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("小红书内容验证示例")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
async with Publisher() as publisher:
|
||
|
|
# 测试各种内容验证场景
|
||
|
|
test_cases = [
|
||
|
|
# 有效内容
|
||
|
|
{
|
||
|
|
"name": "有效的图文笔记",
|
||
|
|
"content": ImageNote(
|
||
|
|
title="有效标题",
|
||
|
|
description="有效描述",
|
||
|
|
images=["test.jpg"]
|
||
|
|
),
|
||
|
|
"expected": True
|
||
|
|
},
|
||
|
|
|
||
|
|
# 无效内容 - 空标题
|
||
|
|
{
|
||
|
|
"name": "空标题",
|
||
|
|
"content": ImageNote(
|
||
|
|
title="",
|
||
|
|
description="有效描述",
|
||
|
|
images=["test.jpg"]
|
||
|
|
),
|
||
|
|
"expected": False
|
||
|
|
},
|
||
|
|
|
||
|
|
# 无效内容 - 空描述
|
||
|
|
{
|
||
|
|
"name": "空描述",
|
||
|
|
"content": ImageNote(
|
||
|
|
title="有效标题",
|
||
|
|
description="",
|
||
|
|
images=["test.jpg"]
|
||
|
|
),
|
||
|
|
"expected": False
|
||
|
|
},
|
||
|
|
|
||
|
|
# 无效内容 - 无图片
|
||
|
|
{
|
||
|
|
"name": "无图片",
|
||
|
|
"content": ImageNote(
|
||
|
|
title="有效标题",
|
||
|
|
description="有效描述",
|
||
|
|
images=[]
|
||
|
|
),
|
||
|
|
"expected": False
|
||
|
|
},
|
||
|
|
|
||
|
|
# 无效内容 - 图片过多
|
||
|
|
{
|
||
|
|
"name": "图片过多",
|
||
|
|
"content": ImageNote(
|
||
|
|
title="有效标题",
|
||
|
|
description="有效描述",
|
||
|
|
images=[f"test{i}.jpg" for i in range(15)] # 15张图片
|
||
|
|
),
|
||
|
|
"expected": False
|
||
|
|
}
|
||
|
|
]
|
||
|
|
|
||
|
|
print("开始内容验证测试...")
|
||
|
|
for i, test_case in enumerate(test_cases, 1):
|
||
|
|
print(f"\n{i}. 测试: {test_case['name']}")
|
||
|
|
|
||
|
|
try:
|
||
|
|
is_valid, error_msg = await publisher.validate_content(
|
||
|
|
"xiaohongshu",
|
||
|
|
test_case['content']
|
||
|
|
)
|
||
|
|
|
||
|
|
if is_valid == test_case['expected']:
|
||
|
|
print(f" ✅ 验证结果符合预期")
|
||
|
|
else:
|
||
|
|
print(f" ❌ 验证结果不符合预期")
|
||
|
|
print(f" 期望: {test_case['expected']}")
|
||
|
|
print(f" 实际: {is_valid}")
|
||
|
|
|
||
|
|
if not is_valid:
|
||
|
|
print(f" 错误信息: {error_msg}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f" ❌ 验证异常: {e}")
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
"""主函数"""
|
||
|
|
print("🚀 小红书发布示例")
|
||
|
|
print("=" * 60)
|
||
|
|
print("请确保:")
|
||
|
|
print("1. 替换示例中的账号名称")
|
||
|
|
print("2. 准备好图片和视频文件")
|
||
|
|
print("3. 确保网络连接正常")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 运行示例
|
||
|
|
print("选择要运行的示例:")
|
||
|
|
print("1. 图文笔记发布")
|
||
|
|
print("2. 视频笔记发布")
|
||
|
|
print("3. 批量发布")
|
||
|
|
print("4. 内容验证")
|
||
|
|
|
||
|
|
choice = input("\n请输入选择 (1-4): ").strip()
|
||
|
|
|
||
|
|
if choice == "1":
|
||
|
|
await xiaohongshu_image_note_example()
|
||
|
|
elif choice == "2":
|
||
|
|
await xiaohongshu_video_note_example()
|
||
|
|
elif choice == "3":
|
||
|
|
await xiaohongshu_batch_publish_example()
|
||
|
|
elif choice == "4":
|
||
|
|
await xiaohongshu_content_validation_example()
|
||
|
|
else:
|
||
|
|
print("❌ 无效选择")
|
||
|
|
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
print("\n⏹️ 用户中断")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"\n❌ 运行异常: {e}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|