#!/usr/bin/env python3 """ nanobot集成接口 将小红书集成功能暴露给nanobot使用 """ import os import sys import json import logging from pathlib import Path from typing import Dict, List, Optional, Any # 添加当前目录到Python路径 sys.path.append(os.path.dirname(os.path.abspath(__file__))) from xiaohongshu_publisher import XiaohongshuPublisher, PublishResult from content_converter import ContentConverter, Article # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class XiaohongshuIntegration: """小红书集成主类""" def __init__(self, config_path: Optional[str] = None): """ 初始化集成 Args: config_path: 配置文件路径 """ self.config_path = config_path or self._get_default_config_path() self.config = self._load_config() # 初始化组件 self.publisher = XiaohongshuPublisher(self.config_path) self.converter = ContentConverter(self.config.get('content', {})) logger.info("小红书集成初始化完成") logger.info(f"配置文件: {self.config_path}") def _get_default_config_path(self) -> str: """获取默认配置文件路径""" skill_dir = Path(__file__).parent config_path = skill_dir / "config.json" if not config_path.exists(): # 创建默认配置 default_config = { "xiaohongshu": { "enabled": True, "cli_path": "xhs-cli", "username": None, "password": None, "max_images": 9, "add_watermark": False, "watermark_text": "技术资讯", "default_tags": ["#技术资讯", "#科技新闻", "#AI"], "publish_interval": 3600, "timeout": 30 }, "content": { "max_title_length": 30, "max_content_length": 1000, "auto_generate_tags": True, "add_source_info": True, "add_emoji": True, "max_tags": 10 }, "images": { "max_width": 1080, "max_height": 1920, "quality": 85, "format": "JPEG" } } with open(config_path, 'w', encoding='utf-8') as f: json.dump(default_config, f, indent=2, ensure_ascii=False) logger.info(f"创建默认配置文件: {config_path}") return str(config_path) def _load_config(self) -> Dict: """加载配置""" try: with open(self.config_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f"加载配置文件失败: {e}") return {} def publish_article(self, article_data: Dict) -> Dict[str, Any]: """ 发布文章到小红书 Args: article_data: 文章数据字典，包含： - title: 标题 - content: 内容 - source: 来源 - url: 原文链接 - images: 图片路径列表 - publish_date: 发布日期（可选） - author: 作者（可选） - category: 分类（可选） Returns: 发布结果字典 """ if not self.config.get('xiaohongshu', {}).get('enabled', True): return { 'success': False, 'error': '小红书集成未启用' } try: # 创建文章对象 article = Article( title=article_data.get('title', ''), content=article_data.get('content', ''), source=article_data.get('source', '未知来源'), url=article_data.get('url', ''), images=article_data.get('images', []), publish_date=article_data.get('publish_date'), author=article_data.get('author'), category=article_data.get('category') ) # 转换为小红书格式 xhs_content = self.converter.to_xiaohongshu(article) # 生成标签 tags = [] if self.config.get('content', {}).get('auto_generate_tags', True): tags = self.converter._generate_tags(article).replace('#', '').split() # 合并默认标签 default_tags = self.config.get('xiaohongshu', {}).get('default_tags', []) tags = list(set(tags + [tag.replace('#', '') for tag in default_tags])) # 发布到小红书 result = self.publisher.publish( title=article.title, content=xhs_content, images=article.images, tags=tags ) # 转换为字典格式 return { 'success': result.success, 'post_id': result.post_id, 'url': result.url, 'message': result.message, 'error': result.error, 'timestamp': result.timestamp, 'content_preview': xhs_content[:200] + '...' if len(xhs_content) > 200 else xhs_content, 'tags_used': tags } except Exception as e: logger.error(f"发布文章失败: {e}") return { 'success': False, 'error': str(e) } def test_connection(self) -> Dict[str, Any]: """测试连接""" try: status = self.publisher.get_status() # 尝试登录 login_success = False if status.get('username') and status.get('cli_available', False): login_success = self.publisher.login() return { 'success': True, 'status': status, 'login_success': login_success, 'config_loaded': bool(self.config), 'converter_ready': True } except Exception as e: return { 'success': False, 'error': str(e) } def get_status(self) -> Dict[str, Any]: """获取集成状态""" publisher_status = self.publisher.get_status() return { 'publisher': publisher_status, 'converter': { 'config': self.converter.config, 'emoji_count': len(self.converter.EMOJI_MAPPING), 'tag_mapping_count': len(self.converter.TAG_MAPPING) }, 'config': { 'path': self.config_path, 'enabled': self.config.get('xiaohongshu', {}).get('enabled', True) } } def update_config(self, new_config: Dict) -> bool: """ 更新配置 Args: new_config: 新的配置字典 Returns: 是否成功 """ try: # 合并配置 current_config = self._load_config() # 深度合并 def deep_update(target, source): for key, value in source.items(): if key in target and isinstance(target[key], dict) and isinstance(value, dict): deep_update(target[key], value) else: target[key] = value deep_update(current_config, new_config) # 保存配置 with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(current_config, f, indent=2, ensure_ascii=False) # 重新加载配置 self.config = current_config self.converter = ContentConverter(self.config.get('content', {})) logger.info("配置更新成功") return True except Exception as e: logger.error(f"更新配置失败: {e}") return False def convert_content_only(self, article_data: Dict) -> Dict[str, Any]: """ 仅转换内容（不发布） Args: article_data: 文章数据 Returns: 转换结果 """ try: article = Article( title=article_data.get('title', ''), content=article_data.get('content', ''), source=article_data.get('source', '未知来源'), url=article_data.get('url', ''), images=article_data.get('images', []), publish_date=article_data.get('publish_date'), author=article_data.get('author'), category=article_data.get('category') ) # 转换内容 xhs_content = self.converter.to_xiaohongshu(article) # 分析文章 analysis = self.converter.analyze_article(article) return { 'success': True, 'xhs_content': xhs_content, 'analysis': analysis, 'title': article.title, 'source': article.source, 'content_length': len(xhs_content), 'recommended_tags': analysis.get('recommended_tags', []) } except Exception as e: return { 'success': False, 'error': str(e) } # nanobot技能接口 class XiaohongshuSkill: """nanobot技能接口类""" def __init__(self): self.integration = None self.skill_dir = Path(__file__).parent def initialize(self): """初始化技能""" try: config_path = self.skill_dir / "config.json" self.integration = XiaohongshuIntegration(str(config_path)) # 测试连接 test_result = self.integration.test_connection() return { 'success': True, 'message': '小红书集成技能初始化成功', 'test_result': test_result, 'config_path': str(config_path) } except Exception as e: return { 'success': False, 'error': str(e), 'message': '小红书集成技能初始化失败' } def handle_command(self, command: str, args: Dict) -> Dict[str, Any]: """ 处理命令 Args: command: 命令名称 args: 命令参数 Returns: 命令执行结果 """ if not self.integration: init_result = self.initialize() if not init_result['success']: return init_result try: if command == 'publish': # 发布文章 return self.integration.publish_article(args) elif command == 'convert': # 转换内容 return self.integration.convert_content_only(args) elif command == 'status': # 获取状态 status = self.integration.get_status() return { 'success': True, 'status': status } elif command == 'test': # 测试连接 return self.integration.test_connection() elif command == 'update_config': # 更新配置 success = self.integration.update_config(args.get('config', {})) return { 'success': success, 'message': '配置更新成功' if success else '配置更新失败' } else: return { 'success': False, 'error': f'未知命令: {command}', 'available_commands': ['publish', 'convert', 'status', 'test', 'update_config'] } except Exception as e: return { 'success': False, 'error': str(e) } def get_skill_info(self) -> Dict[str, Any]: """获取技能信息""" return { 'name': 'xiaohongshu_integration', 'version': '1.0.0', 'description': '小红书平台集成技能', 'author': 'nanobot', 'commands': { 'publish': '发布文章到小红书', 'convert': '转换文章为小红书格式', 'status': '查看集成状态', 'test': '测试连接', 'update_config': '更新配置' }, 'config_file': str(self.skill_dir / "config.json"), 'requirements': ['Pillow'] } # 创建全局实例 xiaohongshu_skill = XiaohongshuSkill() # nanobot直接调用接口 def publish_to_xiaohongshu(article_data: Dict) -> Dict: """发布到小红书（直接调用）""" return xiaohongshu_skill.handle_command('publish', article_data) def convert_to_xiaohongshu_format(article_data: Dict) -> Dict: """转换为小红书格式（直接调用）""" return xiaohongshu_skill.handle_command('convert', article_data) def get_xiaohongshu_status() -> Dict: """获取小红书集成状态（直接调用）""" return xiaohongshu_skill.handle_command('status', {}) # 使用示例 if __name__ == "__main__": print("=" * 60) print("小红书集成技能测试") print("=" * 60) # 初始化技能 result = xiaohongshu_skill.initialize() print("初始化结果:", json.dumps(result, indent=2, ensure_ascii=False)) if result['success']: # 测试文章 test_article = { 'title': '测试：Python 3.12性能大幅提升', 'content': 'Python 3.12正式发布，带来了显著的性能改进。\n\n主要改进：\n1. 解释器性能提升25%\n2. 更好的错误信息\n3. 新的类型提示语法\n4. 改进的异步支持\n\n对于开发者来说，这是值得升级的版本。', 'source': 'Python官方博客', 'url': 'https://python.org/news', 'images': [], 'publish_date': '2026-03-15', 'author': 'Python核心团队', 'category': '编程语言' } # 测试转换 print("\n" + "=" * 60) print("测试内容转换:") print("=" * 60) convert_result = xiaohongshu_skill.handle_command('convert', test_article) if convert_result['success']: print("转换成功!") print("\n小红书格式内容预览:") print("-" * 40) print(convert_result.get('xhs_content', '')[:300] + '...') print("-" * 40) print("\n文章分析:") print(json.dumps(convert_result.get('analysis', {}), indent=2, ensure_ascii=False)) else: print(f"转换失败: {convert_result.get('error')}") # 测试状态 print("\n" + "=" * 60) print("测试状态查询:") print("=" * 60) status_result = xiaohongshu_skill.handle_command('status', {}) if status_result['success']: print("状态查询成功") # 简化显示 status = status_result.get('status', {}) publisher_status = status.get('publisher', {}) print(f"CLI可用: {publisher_status.get('cli_available', False)}") print(f"已登录: {publisher_status.get('is_logged_in', False)}") print(f"用户名: {publisher_status.get('username', '未设置')}") else: print(f"状态查询失败: {status_result.get('error')}") # 显示技能信息 print("\n" + "=" * 60) print("技能信息:") print("=" * 60) skill_info = xiaohongshu_skill.get_skill_info() print(json.dumps(skill_info, indent=2, ensure_ascii=False)) print("\n" + "=" * 60) print("测试完成") print("=" * 60)