Ubuntu
497 lines
17 KiB
Python
497 lines
17 KiB
Python
#!/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) |