""" 小红书发布示例 演示小红书平台的各种发布场景。 """ 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())